package com.devicehive.messages.bus;

import com.devicehive.model.DeviceCommand;
import com.devicehive.model.DeviceNotification;
import com.devicehive.service.DeviceCommandService;
import com.devicehive.service.HazelcastService;
import com.devicehive.util.LogExecutionTime;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.ITopic;
import com.hazelcast.core.Message;
import com.hazelcast.core.MessageListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.ejb.*;
import javax.enterprise.event.Observes;
import javax.enterprise.event.TransactionPhase;
import javax.inject.Inject;

import static javax.ejb.ConcurrencyManagementType.BEAN;


@Singleton
@ConcurrencyManagement(BEAN)
@Startup
@LogExecutionTime
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public class GlobalMessageBus {

    private static final Logger logger = LoggerFactory.getLogger(GlobalMessageBus.class);

    private static final String DEVICE_COMMAND = "DEVICE_COMMAND";
    private static final String DEVICE_COMMAND_UPDATE = "DEVICE_COMMAND_UPDATE";
    private static final String DEVICE_NOTIFICATION = "DEVICE_NOTIFICATION";

    @EJB
    private HazelcastService hazelcastService;

    @EJB
    private LocalMessageBus localMessageBus;

    private HazelcastInstance hazelcast;

    private String commandListener;
    private String commandUpdateListener;
    private String notificationListener;


    @PostConstruct
    protected void postConstruct() {
        hazelcast = hazelcastService.getHazelcast();

        logger.debug("Initializing topic {}...", DEVICE_COMMAND);
        ITopic<DeviceCommand> deviceCommandTopic = hazelcast.getTopic(DEVICE_COMMAND);
        commandListener = deviceCommandTopic.addMessageListener(new DeviceCommandListener());
        logger.debug("Done {}", DEVICE_COMMAND);

        logger.debug("Initializing topic {}...", DEVICE_COMMAND_UPDATE);
        ITopic<DeviceCommand> deviceCommandUpdateTopic = hazelcast.getTopic(DEVICE_COMMAND_UPDATE);
        commandUpdateListener = deviceCommandUpdateTopic.addMessageListener(new DeviceCommandUpdateListener());
        logger.debug("Done {}", DEVICE_COMMAND_UPDATE);

        logger.debug("Initializing topic {}...", DEVICE_NOTIFICATION);
        ITopic<DeviceNotification> deviceNotificationTopic = hazelcast.getTopic(DEVICE_NOTIFICATION);
        notificationListener = deviceNotificationTopic.addMessageListener(new DeviceNotificationListener());
        logger.debug("Done {}", DEVICE_NOTIFICATION);
    }

    @PreDestroy
    protected void preDestroy() {
        hazelcast.getTopic(DEVICE_COMMAND).removeMessageListener(commandListener);
        hazelcast.getTopic(DEVICE_COMMAND_UPDATE).removeMessageListener(commandUpdateListener);
        hazelcast.getTopic(DEVICE_NOTIFICATION).removeMessageListener(notificationListener);
    }

    @Asynchronous
    public void publishDeviceCommand(@Observes(during = TransactionPhase.AFTER_SUCCESS) @DeviceCommandService.Create DeviceCommand deviceCommand) {
        logger.debug("Sending device command {}", deviceCommand.getId());
        localMessageBus.submitDeviceCommand(deviceCommand);
        hazelcast.getTopic(DEVICE_COMMAND).publish(deviceCommand);
        logger.debug("Sent");
    }

    @Asynchronous
    public void publishDeviceCommandUpdate(@Observes(during = TransactionPhase.AFTER_SUCCESS) @DeviceCommandService.Update DeviceCommand deviceCommandUpdate) {
        logger.debug("Sending device command update {}", deviceCommandUpdate.getId());
        localMessageBus.submitDeviceCommandUpdate(deviceCommandUpdate);
        hazelcast.getTopic(DEVICE_COMMAND_UPDATE).publish(deviceCommandUpdate);
        logger.debug("Sent");
    }

    @Asynchronous
    public void publishDeviceNotification(@Observes(during = TransactionPhase.AFTER_SUCCESS) DeviceNotification deviceNotification) {
        logger.debug("Sending device notification {}", deviceNotification.getId());
        localMessageBus.submitDeviceNotification(deviceNotification);
        hazelcast.getTopic(DEVICE_NOTIFICATION).publish(deviceNotification);
        logger.debug("Sent");
    }

    private  class DeviceCommandListener implements MessageListener<DeviceCommand> {



        @Override
        public void onMessage(Message<DeviceCommand> deviceCommandMessage) {
            if (!deviceCommandMessage.getPublishingMember().localMember()) {
                logger.debug("Received device command {}", deviceCommandMessage.getMessageObject().getId());
                localMessageBus.submitDeviceCommand(deviceCommandMessage.getMessageObject());
            }
        }
    }

    private class DeviceCommandUpdateListener implements MessageListener<DeviceCommand> {



        @Override
        public void onMessage(Message<DeviceCommand> deviceCommandMessage) {
            if (!deviceCommandMessage.getPublishingMember().localMember()) {
                logger.debug("Received device command update {}", deviceCommandMessage.getMessageObject().getId());
                localMessageBus.submitDeviceCommandUpdate(deviceCommandMessage.getMessageObject());
            }
        }
    }


    private  class DeviceNotificationListener implements MessageListener<DeviceNotification> {


        @Override
        public void onMessage(Message<DeviceNotification> deviceNotificationMessage) {
            if (!deviceNotificationMessage.getPublishingMember().localMember()) {
                logger.debug("Received device notification{}", deviceNotificationMessage.getMessageObject().getId());
                localMessageBus.submitDeviceNotification(deviceNotificationMessage.getMessageObject());
            }
        }
    }


}

@@ -578,9 +578,218 @@
 ejb.
-*
+Asynchronous;%0Aimport javax.ejb.ConcurrencyManagement;%0Aimport javax.ejb.EJB;%0Aimport javax.ejb.Singleton;%0Aimport javax.ejb.Startup;%0Aimport javax.ejb.TransactionAttribute;%0Aimport javax.ejb.TransactionAttributeType
 ;%0Aim
@@ -877,36 +877,8 @@
 ase;
-%0Aimport javax.inject.Inject;
 %0A%0Aim
@@ -1185,17 +1185,16 @@
 class);%0A
-%0A
     priv
@@ -1410,17 +1410,16 @@
 ATION%22;%0A
-%0A
     @EJB
@@ -1466,17 +1466,16 @@
 ervice;%0A
-%0A
     @EJB
@@ -1520,17 +1520,16 @@
 ageBus;%0A
-%0A
     priv
@@ -1561,17 +1561,16 @@
 elcast;%0A
-%0A
     priv
@@ -1681,17 +1681,16 @@
 tener;%0A%0A
-%0A
     @Pos
@@ -3112,32 +3112,93 @@
   @Asynchronous%0A
+    @TransactionAttribute(TransactionAttributeType.REQUIRED)%0A
     public void 
@@ -3297,16 +3297,53 @@
 e.Create
+%0A                                    
  DeviceC
@@ -3614,32 +3614,93 @@
   @Asynchronous%0A
+    @TransactionAttribute(TransactionAttributeType.REQUIRED)%0A
     public void 
@@ -3726,16 +3726,29 @@
 dUpdate(
+%0A            
 @Observe
@@ -3818,16 +3818,28 @@
 e.Update
+%0A           
  DeviceC
@@ -4158,24 +4158,85 @@
 synchronous%0A
+    @TransactionAttribute(TransactionAttributeType.REQUIRED)%0A
     public v
@@ -4265,16 +4265,29 @@
 ication(
+%0A            
 @Observe
@@ -4642,33 +4642,32 @@
  %7D%0A%0A    private 
- 
 class DeviceComm
@@ -4716,33 +4716,32 @@
 iceCommand%3E %7B%0A%0A%0A
-%0A
         @Overrid
@@ -5222,17 +5222,16 @@
 nd%3E %7B%0A%0A%0A
-%0A
         
@@ -5641,17 +5641,16 @@
 %0A    %7D%0A%0A
-%0A
     priv
@@ -5653,17 +5653,16 @@
 private 
- 
 class De

Fix for global message bus (transactions fixed)