WIFI-767: Cloud Back End for Events Integration

Processing ClientSession metrics from AP, publishing to SystemEvents

opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloud.java

@@ -66,6 +66,7 @@ import com.telecominfraproject.wlan.opensync.external.integration.models.Opensyn
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncAPVIFState;
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncAWLANNode;
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncWifiAssociatedClients;
+import com.telecominfraproject.wlan.opensync.external.integration.utils.MqttStatsPublisher;
 import com.telecominfraproject.wlan.opensync.ovsdb.dao.models.enumerations.DhcpFpDeviceType;
 import com.telecominfraproject.wlan.opensync.util.OvsdbStringConstants;
 import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
@@ -130,7 +131,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     @Autowired
     private FirmwareServiceInterface firmwareServiceInterface;
     @Autowired
-    private OpensyncExternalIntegrationMqttMessageProcessor mqttMessageProcessor;
+    private MqttStatsPublisher mqttMessageProcessor;
 
     @Autowired
     private OpensyncCloudGatewayController gatewayController;

opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisher.java

@@ -1,4 +1,4 @@
-package com.telecominfraproject.wlan.opensync.external.integration;
+package com.telecominfraproject.wlan.opensync.external.integration.utils;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -30,7 +30,6 @@ import com.telecominfraproject.wlan.client.session.models.ClientSessionDetails;
 import com.telecominfraproject.wlan.client.session.models.ClientSessionMetricDetails;
 import com.telecominfraproject.wlan.cloudeventdispatcher.CloudEventDispatcherInterface;
 import com.telecominfraproject.wlan.core.model.equipment.ChannelBandwidth;
-import com.telecominfraproject.wlan.core.model.equipment.ChannelHopReason;
 import com.telecominfraproject.wlan.core.model.equipment.DetectedAuthMode;
 import com.telecominfraproject.wlan.core.model.equipment.MacAddress;
 import com.telecominfraproject.wlan.core.model.equipment.NeighborScanPacketType;
@@ -40,8 +39,9 @@ import com.telecominfraproject.wlan.core.model.equipment.SecurityType;
 import com.telecominfraproject.wlan.core.model.equipment.WiFiSessionUtility;
 import com.telecominfraproject.wlan.core.model.utils.DecibelUtils;
 import com.telecominfraproject.wlan.equipment.EquipmentServiceInterface;
-import com.telecominfraproject.wlan.equipment.models.ApElementConfiguration;
 import com.telecominfraproject.wlan.equipment.models.Equipment;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSession;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSessionMapInterface;
 import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
 import com.telecominfraproject.wlan.profile.ProfileServiceInterface;
 import com.telecominfraproject.wlan.profile.models.Profile;
@@ -78,7 +78,6 @@ import com.telecominfraproject.wlan.status.models.Status;
 import com.telecominfraproject.wlan.status.models.StatusCode;
 import com.telecominfraproject.wlan.status.models.StatusDataType;
 import com.telecominfraproject.wlan.status.network.models.NetworkAdminStatusData;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeChannelHopEvent;
 import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeEventType;
 import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallReportEvent;
 import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallStartEvent;
@@ -95,7 +94,6 @@ import sts.OpensyncStats.AssocType;
 import sts.OpensyncStats.CallReport;
 import sts.OpensyncStats.CallStart;
 import sts.OpensyncStats.CallStop;
-import sts.OpensyncStats.ChannelSwitchReason;
 import sts.OpensyncStats.Client;
 import sts.OpensyncStats.ClientReport;
 import sts.OpensyncStats.DNSProbeMetric;
@@ -135,10 +133,10 @@ import wc.stats.IpDnsTelemetry.WCStatsReport;
 
 @org.springframework.context.annotation.Profile("opensync_cloud_config")
 @Component
-public class OpensyncExternalIntegrationMqttMessageProcessor {
+public class MqttStatsPublisher {
+
+    private static final Logger LOG = LoggerFactory.getLogger(MqttStatsPublisher.class);
 
-    private static final Logger LOG = LoggerFactory.getLogger(OpensyncExternalIntegrationMqttMessageProcessor.class);
-    
     @Autowired
     private EquipmentServiceInterface equipmentServiceInterface;
     @Autowired
@@ -150,9 +148,11 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
     @Autowired
     private ClientServiceInterface clientServiceInterface;
     @Autowired
-    private CloudEventDispatcherInterface equipmentMetricsCollectorInterface;
+    private CloudEventDispatcherInterface cloudEventDispatcherInterface;
+    @Autowired
+    private RealtimeEventPublisher realtimeEventPublisher;
 
-    void processMqttMessage(String topic, WCStatsReport wcStatsReport) {
+    public void processMqttMessage(String topic, WCStatsReport wcStatsReport) {
         LOG.info("Received WCStatsReport {}", wcStatsReport.toString());
 
         LOG.debug("Received report on topic {}", topic);
@@ -192,7 +192,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
-    void processMqttMessage(String topic, Report report) {
+    public void processMqttMessage(String topic, Report report) {
         LOG.info("Received report on topic {} for ap {}", topic, report.getNodeID());
         int customerId = extractCustomerIdFromTopic(topic);
         String apId = extractApIdFromTopic(topic);
@@ -251,12 +251,12 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
         if (!metricRecordList.isEmpty()) {
             LOG.debug("Publishing Metrics {}", metricRecordList);
-            equipmentMetricsCollectorInterface.publishMetrics(metricRecordList);
+            cloudEventDispatcherInterface.publishMetrics(metricRecordList);
         }
 
     }
 
-    void processMqttMessage(String topic, FlowReport flowReport) {
+    public void processMqttMessage(String topic, FlowReport flowReport) {
 
         LOG.info("Received report on topic {}", topic);
         int customerId = extractCustomerIdFromTopic(topic);
@@ -323,9 +323,9 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
             }
 
-            publishChannelHopEvents(customerId, equipmentId, e);
+            realtimeEventPublisher.publishChannelHopEvents(customerId, equipmentId, e);
-            
+
-            // TODO: add DHCP transaction processing, for now just log it 
+            // TODO: add DHCP transaction processing, for now just log it
             for (DhcpTransaction dhcpTransaction : e.getDhcpTransactionList()) {
                 LOG.info("DhcpTransaction {}", dhcpTransaction);
             }
@@ -334,205 +334,135 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
-    private void publishChannelHopEvents(int customerId, long equipmentId, EventReport e) {
+  
-        
-        LOG.info("publishChannelHopEvents for customerId {} equipmentId {}");
-        
-        List<SystemEvent> events = new ArrayList<>();
-        
-        for (sts.OpensyncStats.EventReport.ChannelSwitchEvent channelSwitchEvent : e.getChannelSwitchList()) {
-            Equipment equipment = equipmentServiceInterface.getOrNull(equipmentId);
-            if (equipment == null)
-                continue;
-            RadioType radioType = null;
-            Long timestamp = null;
-            ChannelHopReason reason = null;
-            Integer channel = null;
-            if (channelSwitchEvent.hasBand()) {                   
-               radioType = OvsdbToWlanCloudTypeMappingUtility
-                .getRadioTypeFromOpensyncStatsRadioBandType(channelSwitchEvent.getBand());            
-            }
-            if (RadioType.isUnsupported(radioType)) {
-                LOG.warn("publishChannelHopEvents:RadioType {} is unsupported, cannot send RealTimeChannelHopEvent for {}", radioType, channelSwitchEvent);
-                continue;
-            }
-            if (channelSwitchEvent.hasTimestampMs()) {
-                timestamp = channelSwitchEvent.getTimestampMs();
-            }
-            if (timestamp == null) {
-                LOG.warn("publishChannelHopEvents:timestamp is null, cannot send RealTimeChannelHopEvent for {}", channelSwitchEvent);
-                continue;  
-            }
-            
-            if (channelSwitchEvent.hasReason()) {
-                if (channelSwitchEvent.getReason().equals(ChannelSwitchReason.high_interference)) reason = ChannelHopReason.HighInterference;
-                else if (channelSwitchEvent.getReason().equals(ChannelSwitchReason.radar_detected)) reason = ChannelHopReason.RadarDetected;
-            }
-            if (ChannelHopReason.isUnsupported(reason)) {
-                LOG.warn("publishChannelHopEvents:reason {} is unsupported, cannot send RealTimeChannelHopEvent for {}", channelSwitchEvent.getReason(), channelSwitchEvent);
-                continue; 
-            }
-            if (channelSwitchEvent.hasChannel()) {
-                channel = channelSwitchEvent.getChannel();
-            }
-            if (channel == null) {
-                LOG.warn("publishChannelHopEvents:channel is null, cannot send RealTimeChannelHopEvent for {}", channelSwitchEvent);
-                continue; 
-            }
-
-            RealTimeChannelHopEvent channelHopEvent = new RealTimeChannelHopEvent(RealTimeEventType.Channel_Hop, customerId, equipmentId, radioType, channel, ((ApElementConfiguration)equipment.getDetails()).getRadioMap().get(radioType).getChannelNumber(), reason, timestamp);
-            
-            events.add(channelHopEvent);
-            
-            LOG.debug("publishChannelHopEvents:Adding ChannelHopEvent to bulk list {}", channelHopEvent);
-        }
-        
-
-        if (events.size() > 0) {
-            LOG.info("publishChannelHopEvents:publishEventsBulk: {}", events);
-            equipmentMetricsCollectorInterface.publishEventsBulk(events);
-        } else {
-            LOG.info("publishChannelHopEvents:No ChannelHopEvents in report");
-        }
-    }
-
     protected void processClientConnectEvent(int customerId, long equipmentId, long locationId, EventReport e,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientConnectEvent clientConnectEvent = apEventClientSession.getClientConnectEvent();
+        
+        realtimeEventPublisher.publishClientConnectSuccessEvent(customerId, equipmentId, clientConnectEvent);
 
-        if (clientConnectEvent.hasStaMac()) {
+        com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
-            com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
+                MacAddress.valueOf(clientConnectEvent.getStaMac()));
-                    MacAddress.valueOf(clientConnectEvent.getStaMac()));
+        if (client == null) {
-            if (client == null) {
+            client = new com.telecominfraproject.wlan.client.models.Client();
-                client = new com.telecominfraproject.wlan.client.models.Client();
 
-                client.setCustomerId(customerId);
+            client.setCustomerId(customerId);
-                client.setMacAddress(MacAddress.valueOf(clientConnectEvent.getStaMac()));
+            client.setMacAddress(MacAddress.valueOf(clientConnectEvent.getStaMac()));
 
-                client.setDetails(new ClientInfoDetails());
+            client.setDetails(new ClientInfoDetails());
 
-                client = clientServiceInterface.create(client);
+            client = clientServiceInterface.create(client);
-
-            }
-
-            ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId,
-                    MacAddress.valueOf(clientConnectEvent.getStaMac()));
-
-            if (clientSession == null) {
-                clientSession = new ClientSession();
-            }
-
-            clientSession.setCustomerId(customerId);
-            clientSession.setEquipmentId(equipmentId);
-            clientSession.setLocationId(locationId);
-            clientSession.setMacAddress(MacAddress.valueOf(clientConnectEvent.getStaMac()));
-
-            ClientSessionDetails clientSessionDetails = new ClientSessionDetails();
-            clientSessionDetails.setSessionId(apEventClientSession.getSessionId());
-
-            if (clientConnectEvent.hasFbtUsed()) {
-
-                // TODO: mapping?
-
-            }
-            if (clientConnectEvent.hasEvTimeBootupInUsAssoc()) {
-                clientSessionDetails.setAssocTimestamp(clientConnectEvent.getEvTimeBootupInUsAssoc());
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsAuth()) {
-                clientSessionDetails.setAuthTimestamp(clientConnectEvent.getEvTimeBootupInUsAuth());
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsEapol()) {
-                ClientEapDetails eapDetails = new ClientEapDetails();
-                eapDetails.setEapSuccessTimestamp(clientConnectEvent.getEvTimeBootupInUsEapol());
-                clientSessionDetails.setEapDetails(eapDetails);
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsFirstRx()) {
-                clientSessionDetails.setFirstDataRcvdTimestamp(clientConnectEvent.getEvTimeBootupInUsFirstRx());
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsFirstTx()) {
-                clientSessionDetails.setFirstDataSentTimestamp(clientConnectEvent.getEvTimeBootupInUsFirstTx());
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsIp()) {
-                clientSessionDetails.setIpTimestamp(clientConnectEvent.getEvTimeBootupInUsIp());
-            }
-
-            if (clientConnectEvent.hasEvTimeBootupInUsPortEnable()) {
-                clientSessionDetails.setPortEnabledTimestamp(clientConnectEvent.getEvTimeBootupInUsPortEnable());
-            }
-
-            if (clientConnectEvent.hasCltId()) {
-                clientSessionDetails.setHostname(clientConnectEvent.getCltId());
-            }
-
-            if (clientConnectEvent.hasSecType()) {
-                clientSessionDetails.setSecurityType(OvsdbToWlanCloudTypeMappingUtility
-                        .getCloudSecurityTypeFromOpensyncStats(clientConnectEvent.getSecType()));
-            }
-
-            if (clientConnectEvent.hasBand()) {
-                clientSessionDetails.setRadioType(OvsdbToWlanCloudTypeMappingUtility
-                        .getRadioTypeFromOpensyncStatsRadioBandType(clientConnectEvent.getBand()));
-            }
-
-            if (clientConnectEvent.hasAssocType()) {
-                clientSessionDetails
-                .setIsReassociation(clientConnectEvent.getAssocType().equals(AssocType.REASSOC));
-
-            }
-
-            if (clientConnectEvent.hasAssocRssi()) {
-                clientSessionDetails.setAssocRssi(clientConnectEvent.getAssocRssi());
-            }
-
-            if (clientConnectEvent.hasSsid()) {
-                clientSessionDetails.setSsid(clientConnectEvent.getSsid());
-            }
-
-            if (clientConnectEvent.hasUsing11K()) {
-                clientSessionDetails.setIs11KUsed(clientConnectEvent.getUsing11K());
-            }
-
-            if (clientConnectEvent.hasUsing11R()) {
-                clientSessionDetails.setIs11RUsed(clientConnectEvent.getUsing11R());
-
-            }
-
-            if (clientConnectEvent.hasUsing11V()) {
-                clientSessionDetails.setIs11VUsed(clientConnectEvent.getUsing11V());
-            }
-
-            if (clientConnectEvent.hasIpAddr()) {
-                try {
-                    clientSessionDetails
-                    .setIpAddress(InetAddress.getByAddress(clientConnectEvent.getIpAddr().toByteArray()));
-                } catch (UnknownHostException e1) {
-                    LOG.error("Invalid Ip Address for client {}", clientConnectEvent.getIpAddr(), e);
-                }
-            }
-            clientSessionDetails.setAssociationState(AssociationState._802_11_Associated);
-
-            if (clientSession.getDetails() == null) {
-                clientSession.setDetails(clientSessionDetails);
-            } else {
-                clientSession.getDetails().mergeSession(clientSessionDetails);
-            }
-
-            clientSession = clientServiceInterface.updateSession(clientSession);
 
         }
 
+        ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId,
+                MacAddress.valueOf(clientConnectEvent.getStaMac()));
+
+        if (clientSession == null) {
+            clientSession = new ClientSession();
+        }
+
+        clientSession.setCustomerId(customerId);
+        clientSession.setEquipmentId(equipmentId);
+        clientSession.setLocationId(locationId);
+        clientSession.setMacAddress(MacAddress.valueOf(clientConnectEvent.getStaMac()));
+
+        ClientSessionDetails clientSessionDetails = new ClientSessionDetails();
+        clientSessionDetails.setSessionId(clientConnectEvent.getSessionId());
+
+        if (clientConnectEvent.hasEvTimeBootupInUsAssoc()) {
+            clientSessionDetails.setAssocTimestamp(clientConnectEvent.getEvTimeBootupInUsAssoc());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsAuth()) {
+            clientSessionDetails.setAuthTimestamp(clientConnectEvent.getEvTimeBootupInUsAuth());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsEapol()) {
+            ClientEapDetails eapDetails = new ClientEapDetails();
+            eapDetails.setEapSuccessTimestamp(clientConnectEvent.getEvTimeBootupInUsEapol());
+            clientSessionDetails.setEapDetails(eapDetails);
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsFirstRx()) {
+            clientSessionDetails.setFirstDataRcvdTimestamp(clientConnectEvent.getEvTimeBootupInUsFirstRx());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsFirstTx()) {
+            clientSessionDetails.setFirstDataSentTimestamp(clientConnectEvent.getEvTimeBootupInUsFirstTx());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsIp()) {
+            clientSessionDetails.setIpTimestamp(clientConnectEvent.getEvTimeBootupInUsIp());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsPortEnable()) {
+            clientSessionDetails.setPortEnabledTimestamp(clientConnectEvent.getEvTimeBootupInUsPortEnable());
+        }
+
+        if (clientConnectEvent.hasCltId()) {
+            clientSessionDetails.setHostname(clientConnectEvent.getCltId());
+        }
+
+        if (clientConnectEvent.hasSecType()) {
+            clientSessionDetails.setSecurityType(OvsdbToWlanCloudTypeMappingUtility
+                    .getCloudSecurityTypeFromOpensyncStats(clientConnectEvent.getSecType()));
+        }
+
+        clientSessionDetails.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                .getRadioTypeFromOpensyncStatsRadioBandType(clientConnectEvent.getBand()));
+
+        if (clientConnectEvent.hasAssocType()) {
+
+            clientSessionDetails.setIsReassociation(clientConnectEvent.getAssocType().equals(AssocType.REASSOC));
+        }
+
+        if (clientConnectEvent.hasAssocRssi()) {
+            clientSessionDetails.setAssocRssi(clientConnectEvent.getAssocRssi());
+        }
+
+        clientSessionDetails.setSsid(clientConnectEvent.getSsid());
+
+        if (clientConnectEvent.hasUsing11K()) {
+            clientSessionDetails.setIs11KUsed(clientConnectEvent.getUsing11K());
+        }
+
+        if (clientConnectEvent.hasUsing11R()) {
+            clientSessionDetails.setIs11RUsed(clientConnectEvent.getUsing11R());
+
+        }
+
+        if (clientConnectEvent.hasUsing11V()) {
+            clientSessionDetails.setIs11VUsed(clientConnectEvent.getUsing11V());
+        }
+
+        if (clientConnectEvent.hasIpAddr()) {
+            try {
+                clientSessionDetails
+                        .setIpAddress(InetAddress.getByAddress(clientConnectEvent.getIpAddr().toByteArray()));
+            } catch (UnknownHostException e1) {
+                LOG.error("Invalid Ip Address for client {}", clientConnectEvent.getIpAddr(), e);
+            }
+        }
+
+        clientSessionDetails.setAssociationState(AssociationState._802_11_Associated);
+
+        if (clientSession.getDetails() == null) {
+            clientSession.setDetails(clientSessionDetails);
+        } else {
+            clientSession.getDetails().mergeSession(clientSessionDetails);
+        }
+
+        clientSession = clientServiceInterface.updateSession(clientSession);
 
     }
 
+    
+
     protected void processClientDisconnectEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientDisconnectEvent clientDisconnectEvent = apEventClientSession.getClientDisconnectEvent();
-
+        realtimeEventPublisher.publishClientDisconnectEvent(customerId, equipmentId, clientDisconnectEvent);
         if (clientDisconnectEvent.hasStaMac()) {
 
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
@@ -615,66 +545,70 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
     protected void processClientAuthEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientAuthEvent clientAuthEvent = apEventClientSession.getClientAuthEvent();
-            if (clientAuthEvent.hasStaMac()) {
+        realtimeEventPublisher.publishClientAuthSystemEvent(customerId, equipmentId, clientAuthEvent);
+        if (clientAuthEvent.hasStaMac()) {
 
-                com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
+            com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
-                        MacAddress.valueOf(clientAuthEvent.getStaMac()));
+                    MacAddress.valueOf(clientAuthEvent.getStaMac()));
-                if (client == null) {
+            if (client == null) {
-                    client = new com.telecominfraproject.wlan.client.models.Client();
+                client = new com.telecominfraproject.wlan.client.models.Client();
 
-                    client.setCustomerId(customerId);
+                client.setCustomerId(customerId);
-                    client.setMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
+                client.setMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
 
-                    client.setDetails(new ClientInfoDetails());
+                client.setDetails(new ClientInfoDetails());
 
-                    client = clientServiceInterface.create(client);
+                client = clientServiceInterface.create(client);
 
-                }
-
-                ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId,
-                        MacAddress.valueOf(clientAuthEvent.getStaMac()));
-
-                if (clientSession == null) {
-                    clientSession = new ClientSession();
-                }
-
-                clientSession.setCustomerId(customerId);
-                clientSession.setEquipmentId(equipmentId);
-                clientSession.setLocationId(locationId);
-                clientSession.setMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
-
-                ClientSessionDetails clientSessionDetails = new ClientSessionDetails();
-
-                clientSessionDetails.setSessionId(apEventClientSession.getSessionId());
-
-                if (clientAuthEvent.hasBand()) {
-                    clientSessionDetails.setRadioType(OvsdbToWlanCloudTypeMappingUtility
-                            .getRadioTypeFromOpensyncStatsRadioBandType(clientAuthEvent.getBand()));
-                }
-                if (clientAuthEvent.hasSsid()) {
-                    clientSessionDetails.setSsid(clientAuthEvent.getSsid());
-                }
-                if (clientAuthEvent.hasAuthStatus()) {
-                    clientSessionDetails.setAssociationState(AssociationState._802_11_Authenticated);
-                }
-
-                if (clientSession.getDetails() == null) {
-                    clientSession.setDetails(clientSessionDetails);
-                } else {
-                    clientSession.getDetails().mergeSession(clientSessionDetails);
-                }
-
-                clientSession = clientServiceInterface.updateSession(clientSession);
-
-            } else {
-                LOG.warn("Cannot update client or client session when no client mac address is present");
             }
-        
+
+            ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId,
+                    MacAddress.valueOf(clientAuthEvent.getStaMac()));
+
+            if (clientSession == null) {
+                clientSession = new ClientSession();
+            }
+
+            clientSession.setCustomerId(customerId);
+            clientSession.setEquipmentId(equipmentId);
+            clientSession.setLocationId(locationId);
+            clientSession.setMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
+
+            ClientSessionDetails clientSessionDetails = new ClientSessionDetails();
+
+            clientSessionDetails.setSessionId(apEventClientSession.getSessionId());
+
+            if (clientAuthEvent.hasBand()) {
+                clientSessionDetails.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                        .getRadioTypeFromOpensyncStatsRadioBandType(clientAuthEvent.getBand()));
+            }
+            if (clientAuthEvent.hasSsid()) {
+                clientSessionDetails.setSsid(clientAuthEvent.getSsid());
+            }
+            if (clientAuthEvent.hasAuthStatus()) {
+                clientSessionDetails.setAssociationState(AssociationState._802_11_Authenticated);
+            }
+
+            if (clientSession.getDetails() == null) {
+                clientSession.setDetails(clientSessionDetails);
+            } else {
+                clientSession.getDetails().mergeSession(clientSessionDetails);
+            }
+
+            clientSession = clientServiceInterface.updateSession(clientSession);
+
+        } else {
+            LOG.warn("Cannot update client or client session when no client mac address is present");
+        }
+
     }
 
+   
     protected void processClientAssocEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientAssocEvent clientAssocEvent = apEventClientSession.getClientAssocEvent();
+        realtimeEventPublisher.publishClientAssocEvent(customerId, equipmentId, clientAssocEvent);
+
         if (clientAssocEvent.hasStaMac()) {
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
                     MacAddress.valueOf(clientAssocEvent.getStaMac()));
@@ -745,6 +679,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
+   
+
     protected void processClientFailureEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientFailureEvent clientFailureEvent = apEventClientSession.getClientFailureEvent();
@@ -807,9 +743,12 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
+   
+
     protected void processClientFirstDataEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientFirstDataEvent clientFirstDataEvent = apEventClientSession.getClientFirstDataEvent();
+        realtimeEventPublisher.publishClientFirstDataEvent(customerId, equipmentId, clientFirstDataEvent);
         if (clientFirstDataEvent.hasStaMac()) {
 
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
@@ -866,9 +805,11 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
+
     protected void processClientIdEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientIdEvent clientIdEvent = apEventClientSession.getClientIdEvent();
+        realtimeEventPublisher.publishClientIdEvent(customerId, equipmentId, clientIdEvent);
         if (clientIdEvent.hasCltMac()) {
 
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
@@ -915,9 +856,12 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
+
+
     protected void processClientIpEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientIpEvent clientIpEvent = apEventClientSession.getClientIpEvent();
+        realtimeEventPublisher.publishClientIpEvent(customerId, equipmentId, clientIpEvent);
         if (clientIpEvent.hasStaMac()) {
 
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
@@ -951,8 +895,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
             clientSessionDetails.setSessionId(apEventClientSession.getSessionId());
 
             try {
-                clientSessionDetails
+                clientSessionDetails.setIpAddress(InetAddress.getByAddress(clientIpEvent.getIpAddr().toByteArray()));
-                .setIpAddress(InetAddress.getByAddress(clientIpEvent.getIpAddr().toByteArray()));
             } catch (UnknownHostException e1) {
                 // TODO Auto-generated catch block
                 e1.printStackTrace();
@@ -975,6 +918,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
     protected void processClientTimeoutEvent(int customerId, long equipmentId, long locationId,
             sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
         ClientTimeoutEvent clientTimeoutEvent = apEventClientSession.getClientTimeoutEvent();
+        realtimeEventPublisher.publishClientTimeoutEvent(customerId, equipmentId, clientTimeoutEvent);
         if (clientTimeoutEvent.hasStaMac()) {
 
             com.telecominfraproject.wlan.client.models.Client client = clientServiceInterface.getOrNull(customerId,
@@ -1039,6 +983,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
     }
 
+   
+
     void populateSipCallReport(List<ServiceMetric> metricRecordList, Report report, int customerId, long equipmentId,
             String apId, long locationId) {
         // only in case it is not there, we will just use the time when we
@@ -1069,13 +1015,13 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
         }
 
         if (eventsList.size() > 0) {
-            equipmentMetricsCollectorInterface.publishEventsBulk(eventsList);
+            cloudEventDispatcherInterface.publishEventsBulk(eventsList);
         }
 
     }
 
     protected void processRealTimeSipCallReportEvent(int customerId, long equipmentId, long eventTimestamp,
-                                                     List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
         if (videoVoiceReport.hasCallReport()) {
 
             CallReport callReport = videoVoiceReport.getCallReport();
@@ -1084,7 +1030,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
                     eventTimestamp);
 
             if (callReport.hasClientMac() && callReport.getClientMac().isValidUtf8()) {
-                cloudSipCallReportEvent.setClientMacAddress(MacAddress.valueOf(callReport.getClientMac().toStringUtf8()));
+                cloudSipCallReportEvent
+                        .setClientMacAddress(MacAddress.valueOf(callReport.getClientMac().toStringUtf8()));
             }
             cloudSipCallReportEvent.setStatuses(processRtpFlowStats(callReport.getStatsList()));
             cloudSipCallReportEvent.setEventType(RealTimeEventType.SipCallReport);
@@ -1120,14 +1067,14 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
     private SIPCallReportReason getCallReportReason(CallReport.CallReportReason reason) {
         if (reason != null) {
             switch (reason) {
-                case ROAMED_TO:
+            case ROAMED_TO:
-                    return SIPCallReportReason.ROAMED_TO;
+                return SIPCallReportReason.ROAMED_TO;
-                case GOT_PUBLISH:
+            case GOT_PUBLISH:
-                    return SIPCallReportReason.GOT_PUBLISH;
+                return SIPCallReportReason.GOT_PUBLISH;
-                case ROAMED_FROM:
+            case ROAMED_FROM:
-                    return SIPCallReportReason.ROAMED_FROM;
+                return SIPCallReportReason.ROAMED_FROM;
-                default:
+            default:
-                    return SIPCallReportReason.UNSUPPORTED;
+                return SIPCallReportReason.UNSUPPORTED;
             }
         }
         return SIPCallReportReason.UNSUPPORTED;
@@ -1143,7 +1090,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
                     eventTimestamp);
 
             if (apCallStart.hasClientMac() && apCallStart.getClientMac().isValidUtf8()) {
-                cloudSipCallStartEvent.setClientMacAddress(MacAddress.valueOf(apCallStart.getClientMac().toStringUtf8()));
+                cloudSipCallStartEvent
+                        .setClientMacAddress(MacAddress.valueOf(apCallStart.getClientMac().toStringUtf8()));
             }
 
             if (apCallStart.hasDeviceInfo()) {
@@ -1278,7 +1226,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
             }
 
             if (apStreamVideoServer.hasClientMac() && apStreamVideoServer.getClientMac().isValidUtf8()) {
-                rtsStartEvent.setClientMacAddress(MacAddress.valueOf(apStreamVideoServer.getClientMac().toStringUtf8()));
+                rtsStartEvent
+                        .setClientMacAddress(MacAddress.valueOf(apStreamVideoServer.getClientMac().toStringUtf8()));
             }
 
             if (apStreamVideoServer.hasSessionId()) {
@@ -1294,7 +1243,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
         }
     }
 
-    private List<com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats> processRtpFlowStats(List<OpensyncStats.RtpFlowStats> stats) {
+    private List<com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats> processRtpFlowStats(
+            List<OpensyncStats.RtpFlowStats> stats) {
         List<com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats> cloudRtpFlowStatsList = new ArrayList<>();
         for (RtpFlowStats apRtpFlowStats : stats) {
 
@@ -1322,34 +1272,34 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
             if (apRtpFlowStats.hasDirection()) {
                 switch (apRtpFlowStats.getDirection()) {
-                    case RTP_DOWNSTREAM:
+                case RTP_DOWNSTREAM:
-                        cloudRtpStats.setDirection(
+                    cloudRtpStats.setDirection(
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.DOWNSTREAM);
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.DOWNSTREAM);
-                        break;
+                    break;
-                    case RTP_UPSTREAM:
+                case RTP_UPSTREAM:
-                        cloudRtpStats.setDirection(
+                    cloudRtpStats.setDirection(
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UPSTREAM);
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UPSTREAM);
-                        break;
+                    break;
-                    default:
+                default:
-                        cloudRtpStats.setDirection(
+                    cloudRtpStats.setDirection(
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UNSUPPORTED);
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UNSUPPORTED);
                 }
             }
 
             if (apRtpFlowStats.hasRtpFlowType()) {
                 switch (apRtpFlowStats.getRtpFlowType()) {
-                    case RTP_VIDEO:
+                case RTP_VIDEO:
-                        cloudRtpStats.setFlowType(
+                    cloudRtpStats
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VIDEO);
+                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VIDEO);
-                        break;
+                    break;
-                    case RTP_VOICE:
+                case RTP_VOICE:
-                        cloudRtpStats.setFlowType(
+                    cloudRtpStats
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VOICE);
+                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VOICE);
-                        break;
+                    break;
-                    default:
+                default:
-                        cloudRtpStats.setFlowType(
+                    cloudRtpStats.setFlowType(
-                                com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.UNSUPPORTED);
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.UNSUPPORTED);
-                        break;
+                    break;
                 }
             }
 
@@ -1676,13 +1626,13 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
                 RadioUtilization radioUtil = new RadioUtilization();
                 radioUtil.setTimestampSeconds((int) ((survey.getTimestampMs()) / 1000));
                 int pctBusyTx = busyTx / totalDurationMs;
-            	checkIfOutOfBound("pctBusyTx", pctBusyTx, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
+                checkIfOutOfBound("pctBusyTx", pctBusyTx, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
-            
+
                 radioUtil.setAssocClientTx(pctBusyTx);
                 int pctBusyRx = busyRx / totalDurationMs;
                 checkIfOutOfBound("pctBusyRx", pctBusyRx, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
                 radioUtil.setAssocClientRx(pctBusyRx);
-                
+
                 double pctIBSS = (busyTx + busySelf) / totalDurationMs;
                 radioUtil.setIbss(pctIBSS);
                 int nonWifi = (busy - (busyTx + busyRx)) / totalDurationMs;
@@ -1727,23 +1677,23 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
         updateDeviceStatusRadioUtilizationReport(customerId, equipmentId, radioUtilizationReport);
     }
-    
+
-    private void checkIfOutOfBound(String checkedType, int checkedValue, Survey survey, int totalDurationMs, 
+    private void checkIfOutOfBound(String checkedType, int checkedValue, Survey survey, int totalDurationMs, int busyTx,
-    		int busyTx, int busyRx, int busy, int busySelf) {
+            int busyRx, int busy, int busySelf) {
-    	if (checkedValue > 100 || checkedValue < 0) {
+        if (checkedValue > 100 || checkedValue < 0) {
-    		LOG.warn("Calculated value for {} {} is out of bounds on totalDurationMs {} for survey.getBand {}. busyTx {} busyRx {} busy {} busySelf {} "
+            LOG.warn(
-        			+ " survey.getTimestampMs {}, survey.getSurveyListList {}",
+                    "Calculated value for {} {} is out of bounds on totalDurationMs {} for survey.getBand {}. busyTx {} busyRx {} busy {} busySelf {} "
-        			checkedType, checkedValue, totalDurationMs, survey.getBand(), busyTx, busyRx, busy, busySelf, 
+                            + " survey.getTimestampMs {}, survey.getSurveyListList {}",
-        			survey.getTimestampMs(), survey.getSurveyListList());
+                    checkedType, checkedValue, totalDurationMs, survey.getBand(), busyTx, busyRx, busy, busySelf,
-    	}
+                    survey.getTimestampMs(), survey.getSurveyListList());
+        }
     }
 
     private void updateNetworkAdminStatusReport(int customerId, long equipmentId, ApNodeMetrics apNodeMetrics) {
         apNodeMetrics.getNetworkProbeMetrics().forEach(n -> {
 
-            
             LOG.info("Update NetworkAdminStatusReport for NetworkProbeMetrics {}", n.toString());
-            
+
             Status networkAdminStatus = statusServiceInterface.getOrNull(customerId, equipmentId,
                     StatusDataType.NETWORK_ADMIN);
 
@@ -1756,9 +1706,9 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
                 networkAdminStatus.setDetails(new NetworkAdminStatusData());
                 networkAdminStatus = statusServiceInterface.update(networkAdminStatus);
             }
-            
+
             NetworkAdminStatusData statusData = (NetworkAdminStatusData) networkAdminStatus.getDetails();
-            
+
             if (n.getDnsState() == null) {
                 LOG.debug("No DnsState present in networkProbeMetrics, DnsState and CloudLinkStatus set to 'normal");
                 statusData.setDnsStatus(StatusCode.normal);
@@ -1791,7 +1741,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
     }
 
     private static StatusCode stateUpDownErrorToStatusCode(StateUpDownError state) {
-        
+
         switch (state) {
         case enabled:
             return StatusCode.normal;
@@ -1887,7 +1837,6 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
                 }
 
-
             }
 
             if (networkProbe.hasVlanProbe()) {
@@ -1938,7 +1887,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
             for (Client cl : clReport.getClientListList()) {
 
                 if (cl.getMacAddress() == null) {
-                    LOG.debug("No mac address for Client {}, cannot set device mac address for client in ClientMetrics.",
+                    LOG.debug(
+                            "No mac address for Client {}, cannot set device mac address for client in ClientMetrics.",
                             cl);
                     continue;
                 }
@@ -2161,7 +2111,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
                         || ssidConfig.getSecureMode().equals(SecureMode.wpa2OnlyPSK)) {
                     latestClientSessionDetails.setSecurityType(SecurityType.PSK);
                 } else if (ssidConfig.getSecureMode().equals(SecureMode.wpa3OnlySAE)
-                        || ssidConfig.getSecureMode().equals(SecureMode.wpa3MixedSAE)){
+                        || ssidConfig.getSecureMode().equals(SecureMode.wpa3MixedSAE)) {
                     latestClientSessionDetails.setSecurityType(SecurityType.SAE);
                 } else if (ssidConfig.getSecureMode().equals(SecureMode.wpa2Radius)
                         || ssidConfig.getSecureMode().equals(SecureMode.wpaRadius)
@@ -2476,16 +2426,15 @@ public class OpensyncExternalIntegrationMqttMessageProcessor {
 
         ProfileContainer profileContainer = new ProfileContainer(
                 profileServiceInterface.getProfileWithChildren(profileId));
-        
+
         Profile rfProfile = profileContainer.getChildOfTypeOrNull(profileId, ProfileType.rf);
         RfConfiguration rfConfig = null;
         if (rfProfile != null) {
-         rfConfig = (RfConfiguration) profileContainer.getChildOfTypeOrNull(profileId, ProfileType.rf)
+            rfConfig = (RfConfiguration) profileContainer.getChildOfTypeOrNull(profileId, ProfileType.rf).getDetails();
-                .getDetails();
         }
-        
+
         if (rfConfig == null) {
-            LOG.warn("Cannot get RfConfiguration for customerId {} equipmentId {}", customerId,equipmentId);
+            LOG.warn("Cannot get RfConfiguration for customerId {} equipmentId {}", customerId, equipmentId);
             return;
         }
 

opensync-ext-cloud/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/RealtimeEventPublisher.java

@@ -0,0 +1,480 @@
+package com.telecominfraproject.wlan.opensync.external.integration.utils;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import com.telecominfraproject.wlan.client.models.events.realtime.ClientConnectSuccessEvent;
+import com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent.DisconnectFrameType;
+import com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent.DisconnectInitiator;
+import com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent.ClientTimeoutReason;
+import com.telecominfraproject.wlan.cloudeventdispatcher.CloudEventDispatcherInterface;
+import com.telecominfraproject.wlan.core.model.equipment.ChannelHopReason;
+import com.telecominfraproject.wlan.core.model.equipment.MacAddress;
+import com.telecominfraproject.wlan.core.model.equipment.RadioType;
+import com.telecominfraproject.wlan.equipment.EquipmentServiceInterface;
+import com.telecominfraproject.wlan.equipment.models.ApElementConfiguration;
+import com.telecominfraproject.wlan.equipment.models.Equipment;
+import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeChannelHopEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeEventType;
+import com.telecominfraproject.wlan.systemevent.models.SystemEvent;
+import com.telecominfraproject.wlan.systemevent.models.SystemEventRecord;
+
+import sts.OpensyncStats.AssocType;
+import sts.OpensyncStats.CTReasonType;
+import sts.OpensyncStats.ChannelSwitchReason;
+import sts.OpensyncStats.DeviceType;
+import sts.OpensyncStats.EventReport;
+import sts.OpensyncStats.FrameType;
+import sts.OpensyncStats.EventReport.ClientAssocEvent;
+import sts.OpensyncStats.EventReport.ClientAuthEvent;
+import sts.OpensyncStats.EventReport.ClientConnectEvent;
+import sts.OpensyncStats.EventReport.ClientDisconnectEvent;
+import sts.OpensyncStats.EventReport.ClientFailureEvent;
+import sts.OpensyncStats.EventReport.ClientFirstDataEvent;
+import sts.OpensyncStats.EventReport.ClientIdEvent;
+import sts.OpensyncStats.EventReport.ClientIpEvent;
+import sts.OpensyncStats.EventReport.ClientTimeoutEvent;
+
+@org.springframework.context.annotation.Profile("opensync_cloud_config")
+@Component
+public class RealtimeEventPublisher {
+
+    @Autowired
+    private CloudEventDispatcherInterface cloudEventDispatcherInterface;
+    
+    @Autowired
+    private EquipmentServiceInterface equipmentServiceInterface;
+
+    private static final Logger LOG = LoggerFactory
+            .getLogger(RealtimeEventPublisher.class);
+    
+    void publishChannelHopEvents(int customerId, long equipmentId, EventReport e) {
+
+        LOG.info("publishChannelHopEvents for customerId {} equipmentId {}");
+
+        List<SystemEvent> events = new ArrayList<>();
+        List<SystemEventRecord> eventRecords = new ArrayList<>();
+
+        for (sts.OpensyncStats.EventReport.ChannelSwitchEvent channelSwitchEvent : e.getChannelSwitchList()) {
+            Equipment equipment = equipmentServiceInterface.getOrNull(equipmentId);
+            if (equipment == null)
+                continue;
+            RadioType radioType = null;
+            Long timestamp = null;
+            ChannelHopReason reason = null;
+            Integer channel = null;
+            if (channelSwitchEvent.hasBand()) {
+                radioType = OvsdbToWlanCloudTypeMappingUtility
+                        .getRadioTypeFromOpensyncStatsRadioBandType(channelSwitchEvent.getBand());
+            }
+            if (RadioType.isUnsupported(radioType)) {
+                LOG.warn(
+                        "publishChannelHopEvents:RadioType {} is unsupported, cannot send RealTimeChannelHopEvent for {}",
+                        radioType, channelSwitchEvent);
+                continue;
+            }
+            if (channelSwitchEvent.hasTimestampMs()) {
+                timestamp = channelSwitchEvent.getTimestampMs();
+            }
+            if (timestamp == null) {
+                LOG.warn("publishChannelHopEvents:timestamp is null, cannot send RealTimeChannelHopEvent for {}",
+                        channelSwitchEvent);
+                continue;
+            }
+
+            if (channelSwitchEvent.hasReason()) {
+                if (channelSwitchEvent.getReason().equals(ChannelSwitchReason.high_interference))
+                    reason = ChannelHopReason.HighInterference;
+                else if (channelSwitchEvent.getReason().equals(ChannelSwitchReason.radar_detected))
+                    reason = ChannelHopReason.RadarDetected;
+            }
+            if (ChannelHopReason.isUnsupported(reason)) {
+                LOG.warn("publishChannelHopEvents:reason {} is unsupported, cannot send RealTimeChannelHopEvent for {}",
+                        channelSwitchEvent.getReason(), channelSwitchEvent);
+                continue;
+            }
+            if (channelSwitchEvent.hasChannel()) {
+                channel = channelSwitchEvent.getChannel();
+            }
+            if (channel == null) {
+                LOG.warn("publishChannelHopEvents:channel is null, cannot send RealTimeChannelHopEvent for {}",
+                        channelSwitchEvent);
+                continue;
+            }
+
+            RealTimeChannelHopEvent channelHopEvent = new RealTimeChannelHopEvent(RealTimeEventType.Channel_Hop,
+                    customerId, equipmentId, radioType, channel,
+                    ((ApElementConfiguration) equipment.getDetails()).getRadioMap().get(radioType).getChannelNumber(),
+                    reason, timestamp);
+
+            events.add(channelHopEvent);
+            eventRecords.add(new SystemEventRecord(channelHopEvent));
+            LOG.debug("publishChannelHopEvents:Adding ChannelHopEvent to bulk list {}", channelHopEvent);
+        }
+
+        if (events.size() > 0) {
+            LOG.info("publishChannelHopEvents:publishEventsBulk: {}", events);
+            cloudEventDispatcherInterface.publishEventsBulk(events);
+        } else {
+            LOG.info("publishChannelHopEvents:No ChannelHopEvents in report");
+        }
+    }
+
+
+    void publishClientConnectSuccessEvent(int customerId, long equipmentId, ClientConnectEvent clientConnectEvent) {
+        ClientConnectSuccessEvent clientEvent = new ClientConnectSuccessEvent();
+        clientEvent.setMacAddress(MacAddress.valueOf(clientConnectEvent.getStaMac()));
+        clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                .getRadioTypeFromOpensyncStatsRadioBandType(clientConnectEvent.getBand()));
+        clientEvent.setSsid(clientConnectEvent.getSsid());
+        clientEvent.setSessionId(clientConnectEvent.getSessionId());
+
+        if (clientConnectEvent.hasFbtUsed()) {
+            clientEvent.setFbtUsed(clientConnectEvent.getFbtUsed());
+        }
+        if (clientConnectEvent.hasEvTimeBootupInUsAssoc()) {
+            clientEvent.setAssocTs(clientConnectEvent.getEvTimeBootupInUsAssoc());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsAuth()) {
+            clientEvent.setAuthTs(clientConnectEvent.getEvTimeBootupInUsAuth());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsEapol()) {
+            clientEvent.setEapolTs(clientConnectEvent.getEvTimeBootupInUsEapol());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsFirstRx()) {
+            clientEvent.setFirstDataRxTs(clientConnectEvent.getEvTimeBootupInUsFirstRx());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsFirstTx()) {
+            clientEvent.setFirstDataTxTs(clientConnectEvent.getEvTimeBootupInUsFirstTx());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsIp()) {
+            clientEvent.setIpAcquisitionTs(clientConnectEvent.getEvTimeBootupInUsIp());
+        }
+
+        if (clientConnectEvent.hasEvTimeBootupInUsPortEnable()) {
+            clientEvent.setPortEnabledTs(clientConnectEvent.getEvTimeBootupInUsPortEnable());
+        }
+
+        if (clientConnectEvent.hasCltId()) {
+            clientEvent.setHostName(clientConnectEvent.getCltId());
+        }
+
+        if (clientConnectEvent.hasSecType()) {
+            clientEvent.setSecurityType(OvsdbToWlanCloudTypeMappingUtility
+                    .getCloudSecurityTypeFromOpensyncStats(clientConnectEvent.getSecType()));
+        }
+
+        if (clientConnectEvent.hasAssocType()) {
+            clientEvent.setReassociation(clientConnectEvent.getAssocType().equals(AssocType.REASSOC));
+        }
+
+        if (clientConnectEvent.hasAssocRssi()) {
+            clientEvent.setAssocRSSI(clientConnectEvent.getAssocRssi());
+        }
+
+        if (clientConnectEvent.hasUsing11K()) {
+            clientEvent.setUsing11k(clientConnectEvent.getUsing11K());
+        }
+
+        if (clientConnectEvent.hasUsing11R()) {
+            clientEvent.setUsing11r(clientConnectEvent.getUsing11R());
+        }
+
+        if (clientConnectEvent.hasUsing11V()) {
+            clientEvent.setUsing11v(clientConnectEvent.getUsing11V());
+        }
+
+        if (clientConnectEvent.hasIpAddr()) {
+            try {
+                clientEvent.setIpAddr(InetAddress.getByAddress(clientConnectEvent.getIpAddr().toByteArray()));
+
+            } catch (UnknownHostException e1) {
+                LOG.error("Invalid Ip Address for client {}", clientConnectEvent.getIpAddr(), e1);
+            }
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+
+    void publishClientDisconnectEvent(int customerId, long equipmentId, ClientDisconnectEvent clientDisconnectEvent) {
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent clientEvent;
+
+        if (clientDisconnectEvent.hasTimestampMs()) {
+            long timestampMs = clientDisconnectEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent(
+                    timestampMs);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent(
+                    System.currentTimeMillis());
+
+        }
+
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientDisconnectEvent.getStaMac()));
+        clientEvent.setSessionId(clientDisconnectEvent.getSessionId());
+        clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                .getRadioTypeFromOpensyncStatsRadioBandType(clientDisconnectEvent.getBand()));
+        clientEvent.setSsid(clientDisconnectEvent.getSsid());
+
+        if (clientDisconnectEvent.hasDevType()) {
+
+            clientEvent.setInitiator(
+                    clientDisconnectEvent.getDevType().equals(DeviceType.DEV_AP) ? DisconnectInitiator.AccessPoint
+                            : DisconnectInitiator.Client);
+
+        }
+        if (clientDisconnectEvent.hasFrType()) {
+            clientEvent.setFrameType(
+                    clientDisconnectEvent.getFrType().equals(FrameType.FT_DEAUTH) ? DisconnectFrameType.Deauth
+                            : DisconnectFrameType.Disassoc);
+        }
+        if (clientDisconnectEvent.hasInternalRc()) {
+            clientEvent.setInternalReasonCode(clientDisconnectEvent.getInternalRc());
+        }
+        if (clientDisconnectEvent.hasLrcvUpTsInUs()) {
+            clientEvent.setLastRecvTime(clientDisconnectEvent.getLrcvUpTsInUs());
+        }
+        if (clientDisconnectEvent.hasLsentUpTsInUs()) {
+            clientEvent.setLastSentTime(clientDisconnectEvent.getLsentUpTsInUs());
+        }
+        if (clientDisconnectEvent.hasReason()) {
+            clientEvent.setReasonCode(clientDisconnectEvent.getReason());
+        }
+        if (clientDisconnectEvent.hasRssi()) {
+            clientEvent.setRssi(clientDisconnectEvent.getRssi());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+
+    void publishClientAuthSystemEvent(int customerId, long equipmentId, ClientAuthEvent clientAuthEvent) {
+
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientAuthEvent clientEvent;
+
+        if (clientAuthEvent.hasTimestampMs()) {
+            long timestamp = clientAuthEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAuthEvent(timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAuthEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientAuthEvent.getSessionId());
+        clientEvent.setSsid(clientAuthEvent.getSsid());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
+        clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                .getRadioTypeFromOpensyncStatsRadioBandType(clientAuthEvent.getBand()));
+
+        if (clientAuthEvent.hasAuthStatus()) {
+            clientEvent.setAuthStatus(clientAuthEvent.getAuthStatus());
+        }
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+
+    void publishClientAssocEvent(int customerId, long equipmentId, ClientAssocEvent clientAssocEvent) {
+
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientAssocEvent clientEvent;
+
+        if (clientAssocEvent.hasTimestampMs()) {
+            long timestamp = clientAssocEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAssocEvent(timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAssocEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientAssocEvent.getSessionId());
+        clientEvent.setSsid(clientAssocEvent.getSsid());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientAssocEvent.getStaMac()));
+        clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                .getRadioTypeFromOpensyncStatsRadioBandType(clientAssocEvent.getBand()));
+
+        if (clientAssocEvent.hasAssocType()) {
+            clientEvent.setReassociation(clientAssocEvent.getAssocType().equals(AssocType.REASSOC));
+        }
+
+        if (clientAssocEvent.hasInternalSc()) {
+            clientEvent.setInternalSC(clientAssocEvent.getInternalSc());
+        }
+
+        if (clientAssocEvent.hasRssi()) {
+            clientEvent.setRssi(clientAssocEvent.getRssi());
+        }
+
+        if (clientAssocEvent.hasStatus()) {
+            clientEvent.setStatus(clientAssocEvent.getStatus());
+        }
+
+        if (clientAssocEvent.hasUsing11K()) {
+            clientEvent.setUsing11k(clientAssocEvent.getUsing11K());
+        }
+
+        if (clientAssocEvent.hasUsing11R()) {
+            clientEvent.setUsing11r(clientAssocEvent.getUsing11R());
+        }
+
+        if (clientAssocEvent.hasUsing11V()) {
+            clientEvent.setUsing11v(clientAssocEvent.getUsing11V());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+
+    void publishClientFailureEvent(int customerId, long equipmentId, ClientFailureEvent clientFailureEvent) {
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientFailureEvent clientEvent;
+
+        if (clientFailureEvent.hasTimestampMs()) {
+            long timestamp = clientFailureEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFailureEvent(timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFailureEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientFailureEvent.getSessionId());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientFailureEvent.getStaMac()));
+        clientEvent.setSsid(clientFailureEvent.getSsid());
+
+        if (clientFailureEvent.hasReasonStr()) {
+            clientEvent.setReasonString(clientFailureEvent.getReasonStr());
+        }
+
+        if (clientFailureEvent.hasReasonCode()) {
+            clientEvent.setReasonCode(clientFailureEvent.getReasonCode());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+    }
+    
+    void publishClientFirstDataEvent(int customerId, long equipmentId, ClientFirstDataEvent clientFirstDataEvent) {
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientFirstDataEvent clientEvent;
+
+        if (clientFirstDataEvent.hasTimestampMs()) {
+            long timestamp = clientFirstDataEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFirstDataEvent(
+                    timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFirstDataEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientFirstDataEvent.getSessionId());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientFirstDataEvent.getStaMac()));
+
+        if (clientFirstDataEvent.hasFdataTxUpTsInUs()) {
+            clientEvent.setFirstDataSentTs(clientFirstDataEvent.getFdataTxUpTsInUs());
+        }
+
+        if (clientFirstDataEvent.hasFdataRxUpTsInUs()) {
+            clientEvent.setFirstDataRcvdTs(clientFirstDataEvent.getFdataRxUpTsInUs());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+
+    void publishClientIdEvent(int customerId, long equipmentId, ClientIdEvent clientIdEvent) {
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientIdEvent clientEvent;
+
+        if (clientIdEvent.hasTimestampMs()) {
+            long timestamp = clientIdEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIdEvent(timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIdEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientIdEvent.getSessionId());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientIdEvent.getCltMac()));
+        if (clientIdEvent.hasCltId()) {
+            clientEvent.setUserId(clientIdEvent.getCltId());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+    }
+    
+    void publishClientIpEvent(int customerId, long equipmentId, ClientIpEvent clientIpEvent) {
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientIpAddressEvent clientEvent;
+
+        if (clientIpEvent.hasTimestampMs()) {
+            long timestamp = clientIpEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIpAddressEvent(
+                    timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIpAddressEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientIpEvent.getSessionId());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientIpEvent.getStaMac()));
+        if (clientIpEvent.hasIpAddr()) {
+            clientEvent.setIpAddr(clientIpEvent.getIpAddr().toByteArray());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+    }
+    
+    void publishClientTimeoutEvent(int customerId, long equipmentId, ClientTimeoutEvent clientTimeoutEvent) {
+
+        com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent clientEvent;
+
+        if (clientTimeoutEvent.hasTimestampMs()) {
+            long timestamp = clientTimeoutEvent.getTimestampMs();
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent(timestamp);
+        } else {
+            clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent(
+                    System.currentTimeMillis());
+        }
+
+        clientEvent.setSessionId(clientTimeoutEvent.getSessionId());
+        clientEvent.setDeviceMacAddress(MacAddress.valueOf(clientTimeoutEvent.getStaMac()));
+        if (clientTimeoutEvent.hasRCode()) {
+            clientEvent.setTimeoutReason(clientTimeoutEvent.getRCode().equals(CTReasonType.CTR_IDLE_TOO_LONG)
+                    ? ClientTimeoutReason.IdleTooLong
+                    : ClientTimeoutReason.FailedProbe);
+        }
+        if (clientTimeoutEvent.hasLastRcvUpTsInUs()) {
+            clientEvent.setLastRecvTime(clientTimeoutEvent.getLastRcvUpTsInUs());
+        }
+        if (clientTimeoutEvent.hasLastSentUpTsInUs()) {
+            clientEvent.setLastSentTime(clientTimeoutEvent.getLastSentUpTsInUs());
+        }
+
+        clientEvent.setCustomerId(customerId);
+        clientEvent.setEquipmentId(equipmentId);
+        cloudEventDispatcherInterface.publishEvent(clientEvent);
+
+    }
+}

opensync-ext-cloud/src/test/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationCloudTest.java

@@ -66,6 +66,7 @@ import com.telecominfraproject.wlan.opensync.external.integration.controller.Ope
 import com.telecominfraproject.wlan.opensync.external.integration.models.ConnectNodeInfo;
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncAPRadioState;
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncAPVIFState;
+import com.telecominfraproject.wlan.opensync.external.integration.utils.MqttStatsPublisher;
 import com.telecominfraproject.wlan.profile.ProfileServiceInterface;
 import com.telecominfraproject.wlan.profile.models.Profile;
 import com.telecominfraproject.wlan.profile.models.ProfileType;
@@ -125,7 +126,7 @@ public class OpensyncExternalIntegrationCloudTest {
     @MockBean(answer = Answers.RETURNS_MOCKS)
     OpensyncCloudGatewayController gatewayController;
     @MockBean
-    OpensyncExternalIntegrationMqttMessageProcessor opensyncExternalIntegrationMqttProcessor;
+    MqttStatsPublisher opensyncExternalIntegrationMqttProcessor;
 
     @Autowired
     OpensyncExternalIntegrationCloud opensyncExternalIntegrationCloud;
@@ -461,16 +462,6 @@ public class OpensyncExternalIntegrationCloudTest {
 
     }
 
-    @Test
-    public void testExtractApIdFromTopic() {
-
-        String topic = "/ap/Test_Client_21P10C68818122/opensync";
-
-        assertEquals("Test_Client_21P10C68818122",
-                OpensyncExternalIntegrationMqttMessageProcessor.extractApIdFromTopic(topic));
-
-    }
-
     @Test
     public void testProcessMqttMessageStringReport() {
 

opensync-ext-cloud/src/test/java/com/telecominfraproject/wlan/opensync/external/integration/utils/MqttStatsPublisherTest.java

@@ -1,4 +1,4 @@
-package com.telecominfraproject.wlan.opensync.external.integration;
+package com.telecominfraproject.wlan.opensync.external.integration.utils;
 
 import static org.junit.Assert.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -7,7 +7,6 @@ import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.List;
 
-import com.telecominfraproject.wlan.systemevent.models.SystemEvent;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
@@ -44,6 +43,9 @@ import com.telecominfraproject.wlan.equipment.models.ApElementConfiguration;
 import com.telecominfraproject.wlan.equipment.models.Equipment;
 import com.telecominfraproject.wlan.firmware.FirmwareServiceInterface;
 import com.telecominfraproject.wlan.location.service.LocationServiceInterface;
+import com.telecominfraproject.wlan.opensync.external.integration.OpensyncExternalIntegrationCloud;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSession;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSessionMapInterface;
 import com.telecominfraproject.wlan.opensync.external.integration.controller.OpensyncCloudGatewayController;
 import com.telecominfraproject.wlan.profile.ProfileServiceInterface;
 import com.telecominfraproject.wlan.routing.RoutingServiceInterface;
@@ -73,12 +75,12 @@ import sts.OpensyncStats.VLANMetrics;
 
 @RunWith(SpringRunner.class)
 @ActiveProfiles(profiles = { "integration_test", })
-@SpringBootTest(webEnvironment = WebEnvironment.NONE, classes = OpensyncExternalIntegrationMqttMessageProcessorTest.class)
+@SpringBootTest(webEnvironment = WebEnvironment.NONE, classes = MqttStatsPublisherTest.class)
 @Import(value = { AlarmServiceInterface.class, OpensyncExternalIntegrationCloud.class,
-        OpensyncExternalIntegrationMqttMessageProcessorTest.Config.class,
+        MqttStatsPublisherTest.Config.class,
 
 })
-public class OpensyncExternalIntegrationMqttMessageProcessorTest {
+public class MqttStatsPublisherTest {
 
     @MockBean
     AlarmServiceInterface alarmServiceInterface;
@@ -104,9 +106,10 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
     FirmwareServiceInterface firmwareServiceInterface;
     @MockBean(answer = Answers.RETURNS_MOCKS)
     OpensyncCloudGatewayController gatewayController;
-
+    @MockBean
+    RealtimeEventPublisher realtimeEventPublisher;
     @Autowired
-    OpensyncExternalIntegrationMqttMessageProcessor opensyncExternalIntegrationMqttProcessor;
+    MqttStatsPublisher opensyncExternalIntegrationMqttProcessor;
 
     MockitoSession mockito;
 
@@ -114,8 +117,8 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
     static class Config {
 
         @Bean
-        public OpensyncExternalIntegrationMqttMessageProcessor opensyncExternalIntegrationMqttMessageProcessor() {
+        public MqttStatsPublisher mqttStatsPublisher() {
-            return new OpensyncExternalIntegrationMqttMessageProcessor();
+            return new MqttStatsPublisher();
         }
 
     }
@@ -133,14 +136,12 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
 
     }
 
-
     @Test
     public void testExtractApIdFromTopic() {
 
         String topic = "/ap/Test_Client_21P10C68818122/opensync";
 
-        assertEquals("Test_Client_21P10C68818122",
+        assertEquals("Test_Client_21P10C68818122", MqttStatsPublisher.extractApIdFromTopic(topic));
-                OpensyncExternalIntegrationMqttMessageProcessor.extractApIdFromTopic(topic));
 
     }
 
@@ -173,15 +174,15 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
         Equipment equipment = new Equipment();
 
         equipment.setDetails(ApElementConfiguration.createWithDefaults());
-        
+
         equipment.setId(1L);
 
         Mockito.when(
                 equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
                 .thenReturn(equipment);
-        
+
         equipment.setProfileId(0L);
-        
+
         Mockito.when(equipmentServiceInterface.getOrNull(1L)).thenReturn(equipment);
         Mockito.when(equipmentServiceInterface.get(1L)).thenReturn(equipment);
 
@@ -259,51 +260,56 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
         // TODO: implement me when wcs stats reports supported
     }
 
-    // Verify CallStart, CallStop and CallReport are properly processed and published as
+    // Verify CallStart, CallStop and CallReport are properly processed and
+    // published as
     // respective events (5 events in total)
     @Test
     public void testProcessMqttMessageStringMultipleVideoVoiceReport() {
-        OpensyncStats.VideoVoiceReport.Builder callStartVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+        OpensyncStats.VideoVoiceReport.Builder callStartVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder()
-                setCallStart(getDefaultCallStart());
+                .setCallStart(getDefaultCallStart());
-        OpensyncStats.VideoVoiceReport.Builder callReportGotPublishVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+        OpensyncStats.VideoVoiceReport.Builder callReportGotPublishVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH, 121, 1028, 1316, 1888, 298, 2, 100, 200));
+                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH,
-        OpensyncStats.VideoVoiceReport.Builder callReportRoamedToVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+                        121, 1028, 1316, 1888, 298, 2, 100, 200));
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO, 123, 1020, 1116, 1345, 223, 0, 102, 203));
+        OpensyncStats.VideoVoiceReport.Builder callReportRoamedToVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-        OpensyncStats.VideoVoiceReport.Builder callReportRoamedFromVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO,
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_FROM, 122, 1029, 1300, 1234, 111, 3, 101, 201));
+                        123, 1020, 1116, 1345, 223, 0, 102, 203));
-        OpensyncStats.VideoVoiceReport.Builder callStopVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+        OpensyncStats.VideoVoiceReport.Builder callReportRoamedFromVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-                setCallStop(getDefaultCallStop());
+                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_FROM,
+                        122, 1029, 1300, 1234, 111, 3, 101, 201));
+        OpensyncStats.VideoVoiceReport.Builder callStopVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder()
+                .setCallStop(getDefaultCallStop());
         // Create report with multiple voiceReports in one
-        Report multipleVoiceReportsInOneReport = Report.newBuilder().addVideoVoiceReport(callStartVoiceReportBuilder).
+        Report multipleVoiceReportsInOneReport = Report.newBuilder().addVideoVoiceReport(callStartVoiceReportBuilder)
-                addVideoVoiceReport(callReportGotPublishVoiceReportBuilder).
+                .addVideoVoiceReport(callReportGotPublishVoiceReportBuilder)
-                addVideoVoiceReport(callReportRoamedFromVoiceReportBuilder).
+                .addVideoVoiceReport(callReportRoamedFromVoiceReportBuilder)
-                addVideoVoiceReport(callReportRoamedToVoiceReportBuilder).
+                .addVideoVoiceReport(callReportRoamedToVoiceReportBuilder)
-                addVideoVoiceReport(callStopVoiceReportBuilder).
+                .addVideoVoiceReport(callStopVoiceReportBuilder).setNodeID("1").build();
-                setNodeID("1").
-                build();
 
-        opensyncExternalIntegrationMqttProcessor.populateSipCallReport(null, multipleVoiceReportsInOneReport, 1, 2L, "TestAP", 12L);
+        opensyncExternalIntegrationMqttProcessor.populateSipCallReport(null, multipleVoiceReportsInOneReport, 1, 2L,
+                "TestAP", 12L);
 
         Mockito.verify(equipmentMetricsCollectorInterface, Mockito.times(1)).publishEventsBulk(Mockito.anyList());
     }
 
     // Create report with 1 voiceReports that contains multiple Calls
-    // Note that GOT_PUBLISH and ROAMED_FROM are ignored in the CallReport. ROAMED_TO overwrites them
+    // Note that GOT_PUBLISH and ROAMED_FROM are ignored in the CallReport.
+    // ROAMED_TO overwrites them
     @Test
     public void testProcessMqttMessageStringOneVideoVoiceReport() {
 
-        OpensyncStats.VideoVoiceReport.Builder videoVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder().
+        OpensyncStats.VideoVoiceReport.Builder videoVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder()
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH, 121, 1028, 1316,1888, 298, 2, 100,200)).
+                .setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH, 121, 1028,
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_FROM, 122, 1029, 1300,1234, 111, 3, 101,201)).
+                        1316, 1888, 298, 2, 100, 200))
-                setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO, 123, 1020, 1116,1345, 223, 0, 102,203)).
+                .setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_FROM, 122, 1029,
-                setCallStart(getDefaultCallStart()).
+                        1300, 1234, 111, 3, 101, 201))
-                setCallStop(getDefaultCallStop());
+                .setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO, 123, 1020,
-        Report oneVoiceReportWithMultipleCallsInOneReport = Report.getDefaultInstance().toBuilder().
+                        1116, 1345, 223, 0, 102, 203))
-                addVideoVoiceReport(videoVoiceReportBuilder).
+                .setCallStart(getDefaultCallStart()).setCallStop(getDefaultCallStop());
-                setNodeID("1").
+        Report oneVoiceReportWithMultipleCallsInOneReport = Report.getDefaultInstance().toBuilder()
-                build();
+                .addVideoVoiceReport(videoVoiceReportBuilder).setNodeID("1").build();
 
-        opensyncExternalIntegrationMqttProcessor.populateSipCallReport(null, oneVoiceReportWithMultipleCallsInOneReport, 1, 2L, "TestAP", 12L);
+        opensyncExternalIntegrationMqttProcessor.populateSipCallReport(null, oneVoiceReportWithMultipleCallsInOneReport,
+                1, 2L, "TestAP", 12L);
 
         Mockito.verify(equipmentMetricsCollectorInterface, Mockito.times(1)).publishEventsBulk(Mockito.anyList());
     }
@@ -393,13 +399,13 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
         callStopBuilder.setSessionId(123L);
         callStopBuilder.setWifiSessionId(1234L);
         callStopBuilder.setReason(OpensyncStats.CallStop.CallStopReason.BYE_OK);
-        callStopBuilder.addStats(getRtpFlowStats(121, 1380, 1400,3000, 119, 3, 205,350));
+        callStopBuilder.addStats(getRtpFlowStats(121, 1380, 1400, 3000, 119, 3, 205, 350));
 
         return callStopBuilder.build();
     }
-    
+
     private OpensyncStats.RtpFlowStats getRtpFlowStats(int codec, int jitter, int latency, int totalPackets,
-                                                       int totalPacketsLost, int mos, int firstRtpSeq, int lastRtpSeq) {
+            int totalPacketsLost, int mos, int firstRtpSeq, int lastRtpSeq) {
         OpensyncStats.RtpFlowStats.Builder rtpFlowStatsBuilder = OpensyncStats.RtpFlowStats.newBuilder();
         rtpFlowStatsBuilder.setCodec(codec);
         rtpFlowStatsBuilder.setBlockCodecs(ByteString.copyFrom(new byte[] { (byte) 0xe6, 0x1 }));
@@ -417,8 +423,7 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
     }
 
     private OpensyncStats.CallReport getDefaultCallReport(OpensyncStats.CallReport.CallReportReason reason, int codec,
-                                                          int jitter, int latency, int totalPackets, int totalPacketsLost,
+            int jitter, int latency, int totalPackets, int totalPacketsLost, int mos, int firstRtpSeq, int lastRtpSeq) {
-                                                          int mos, int firstRtpSeq, int lastRtpSeq) {
         OpensyncStats.CallReport.Builder callReportBuilder = OpensyncStats.CallReport.newBuilder();
         callReportBuilder.setBand(RadioBandType.BAND5G);
         callReportBuilder.setChannel(40);
@@ -429,15 +434,14 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
         callReportBuilder.setSessionId(123L);
         callReportBuilder.setWifiSessionId(1234L);
         callReportBuilder.setReason(reason);
-        callReportBuilder.addStats(getRtpFlowStats(codec, jitter, latency, totalPackets, totalPacketsLost,
+        callReportBuilder.addStats(
-                mos, firstRtpSeq, lastRtpSeq));
+                getRtpFlowStats(codec, jitter, latency, totalPackets, totalPacketsLost, mos, firstRtpSeq, lastRtpSeq));
 
         return callReportBuilder.build();
     }
 
     private List<EventReport> getOpensyncStatsEventReportsList() {
 
-
         List<ClientAssocEvent> clientAssocEventList = new ArrayList<>();
         sts.OpensyncStats.EventReport.ClientAssocEvent.Builder clientAssocBuilder = EventReport.ClientAssocEvent
                 .getDefaultInstance().toBuilder();
@@ -463,22 +467,22 @@ public class OpensyncExternalIntegrationMqttMessageProcessorTest {
         clientSessionBuilder.setClientAssocEvent(clientAssocBuilder.build());
         List<sts.OpensyncStats.EventReport.ClientSession> clientSessionList = new ArrayList<>();
         clientSessionList.add(clientSessionBuilder.build());
-        
+
-        sts.OpensyncStats.EventReport.ChannelSwitchEvent.Builder channelSwitchEventBuilder = sts.OpensyncStats.EventReport.ChannelSwitchEvent.getDefaultInstance().toBuilder();
+        sts.OpensyncStats.EventReport.ChannelSwitchEvent.Builder channelSwitchEventBuilder = sts.OpensyncStats.EventReport.ChannelSwitchEvent
-        channelSwitchEventBuilder.setBand(RadioBandType.BAND5GL).setChannel(40).setReason(ChannelSwitchReason.high_interference).setTimestampMs(System.currentTimeMillis());
+                .getDefaultInstance().toBuilder();
+        channelSwitchEventBuilder.setBand(RadioBandType.BAND5GL).setChannel(40)
+                .setReason(ChannelSwitchReason.high_interference).setTimestampMs(System.currentTimeMillis());
 
         List<ChannelSwitchEvent> channelSwitchEventList = new ArrayList<>();
         channelSwitchEventList.add(channelSwitchEventBuilder.build());
-        
+
         eventReportBuilder.addAllClientSession(clientSessionList);
         eventReportBuilder.addAllChannelSwitch(channelSwitchEventList);
 
         eventReportList.add(eventReportBuilder.build());
 
-
         return eventReportList;
 
-
     }
 
     private List<ClientReport> getOpensyncStatsClientReportsList() {