WIFI-767: Cloud Back End for Events Integration

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

@@ -78,22 +78,9 @@ 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.RealTimeEventType;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallReportEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallStartEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallStopEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStartEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStartSessionEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStopEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.SIPCallReportReason;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.SipCallStopReason;
-import com.telecominfraproject.wlan.systemevent.models.SystemEvent;
 
 import sts.OpensyncStats;
 import sts.OpensyncStats.AssocType;
-import sts.OpensyncStats.CallReport;
-import sts.OpensyncStats.CallStart;
-import sts.OpensyncStats.CallStop;
 import sts.OpensyncStats.Client;
 import sts.OpensyncStats.ClientReport;
 import sts.OpensyncStats.DNSProbeMetric;
@@ -116,15 +103,10 @@ import sts.OpensyncStats.NetworkProbe;
 import sts.OpensyncStats.RADIUSMetrics;
 import sts.OpensyncStats.RadioBandType;
 import sts.OpensyncStats.Report;
-import sts.OpensyncStats.RtpFlowStats;
-import sts.OpensyncStats.StreamingVideoServerDetected;
-import sts.OpensyncStats.StreamingVideoSessionStart;
-import sts.OpensyncStats.StreamingVideoStop;
 import sts.OpensyncStats.Survey;
 import sts.OpensyncStats.Survey.SurveySample;
 import sts.OpensyncStats.SurveyType;
 import sts.OpensyncStats.VLANMetrics;
-import sts.OpensyncStats.VideoVoiceReport;
 import traffic.NetworkMetadata;
 import traffic.NetworkMetadata.FlowReport;
 import wc.stats.IpDnsTelemetry;
@@ -238,9 +220,8 @@ public class MqttStatsPublisher {
             populateNeighbourScanReports(metricRecordList, report, customerId, equipmentId, locationId);
             populateChannelInfoReports(metricRecordList, report, customerId, equipmentId, locationId, profileId);
             populateApSsidMetrics(metricRecordList, report, customerId, equipmentId, apId, locationId);
-            // TODO: uncomment when AP support present
+
-            populateSipCallReport(metricRecordList, report, customerId, equipmentId, apId, locationId);
+            publishEvents(report, customerId, equipmentId, apId, locationId);
-            processEventReport(report, customerId, equipmentId, apId, locationId);
             // handleRssiMetrics(metricRecordList, report, customerId,
             // equipmentId, locationId);
 
@@ -294,7 +275,9 @@ public class MqttStatsPublisher {
 
     }
 
-    void processEventReport(Report report, int customerId, long equipmentId, String apId, long locationId) {
+    void publishEvents(Report report, int customerId, long equipmentId, String apId, long locationId) {
+
+        realtimeEventPublisher.publishSipCallEvents(customerId, equipmentId, report.getVideoVoiceReportList());
 
         report.getEventReportList().stream().forEach(e -> {
 
@@ -984,440 +967,6 @@ public class MqttStatsPublisher {
 
     }
 
-    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
-        // received the report/event
-        long eventTimestamp = System.currentTimeMillis();
-
-        List<SystemEvent> eventsList = new ArrayList<>();
-        for (VideoVoiceReport videoVoiceReport : report.getVideoVoiceReportList()) {
-
-            if (videoVoiceReport.hasTimestampMs()) {
-                eventTimestamp = videoVoiceReport.getTimestampMs();
-            }
-
-            LOG.debug("Received VideoVoiceReport {} for SIP call", videoVoiceReport);
-
-            processRealTimeSipCallReportEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-            processRealTimeSipCallStartEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-            processRealTimeSipCallStopEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-            processRtsStartEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-            processRtsStartSessionEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-            processRtsStopEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
-
-        }
-
-        if (eventsList.size() > 0) {
-            cloudEventDispatcherInterface.publishEventsBulk(eventsList);
-        }
-
-    }
-
-    protected void processRealTimeSipCallReportEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasCallReport()) {
-
-            CallReport callReport = videoVoiceReport.getCallReport();
-
-            RealTimeSipCallReportEvent cloudSipCallReportEvent = new RealTimeSipCallReportEvent(customerId, equipmentId,
-                    eventTimestamp);
-
-            if (callReport.hasClientMac() && callReport.getClientMac().isValidUtf8()) {
-                cloudSipCallReportEvent
-                        .setClientMacAddress(MacAddress.valueOf(callReport.getClientMac().toStringUtf8()));
-            }
-            cloudSipCallReportEvent.setStatuses(processRtpFlowStats(callReport.getStatsList()));
-            cloudSipCallReportEvent.setEventType(RealTimeEventType.SipCallReport);
-
-            cloudSipCallReportEvent.setSipCallId(callReport.getWifiSessionId());
-            cloudSipCallReportEvent.setAssociationId(callReport.getSessionId());
-
-            if (callReport.hasReason()) {
-                cloudSipCallReportEvent.setReportReason(getCallReportReason(callReport.getReason()));
-            }
-            if (callReport.hasProviderDomain()) {
-                cloudSipCallReportEvent.setProviderDomain(callReport.getProviderDomain());
-            }
-
-            if (callReport.getCodecsCount() > 0) {
-                cloudSipCallReportEvent.setCodecs(callReport.getCodecsList());
-            }
-
-            if (callReport.hasChannel()) {
-                cloudSipCallReportEvent.setChannel(callReport.getChannel());
-            }
-
-            if (callReport.hasBand()) {
-                cloudSipCallReportEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
-                        .getRadioTypeFromOpensyncStatsRadioBandType(callReport.getBand()));
-            }
-
-            eventsList.add(cloudSipCallReportEvent);
-
-        }
-    }
-
-    private SIPCallReportReason getCallReportReason(CallReport.CallReportReason reason) {
-        if (reason != null) {
-            switch (reason) {
-            case ROAMED_TO:
-                return SIPCallReportReason.ROAMED_TO;
-            case GOT_PUBLISH:
-                return SIPCallReportReason.GOT_PUBLISH;
-            case ROAMED_FROM:
-                return SIPCallReportReason.ROAMED_FROM;
-            default:
-                return SIPCallReportReason.UNSUPPORTED;
-            }
-        }
-        return SIPCallReportReason.UNSUPPORTED;
-    }
-
-    protected void processRealTimeSipCallStartEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasCallStart()) {
-
-            CallStart apCallStart = videoVoiceReport.getCallStart();
-
-            RealTimeSipCallStartEvent cloudSipCallStartEvent = new RealTimeSipCallStartEvent(customerId, equipmentId,
-                    eventTimestamp);
-
-            if (apCallStart.hasClientMac() && apCallStart.getClientMac().isValidUtf8()) {
-                cloudSipCallStartEvent
-                        .setClientMacAddress(MacAddress.valueOf(apCallStart.getClientMac().toStringUtf8()));
-            }
-
-            if (apCallStart.hasDeviceInfo()) {
-                cloudSipCallStartEvent.setDeviceInfo(apCallStart.getDeviceInfo());
-            }
-
-            if (apCallStart.hasProviderDomain()) {
-                cloudSipCallStartEvent.setProviderDomain(apCallStart.getProviderDomain());
-            }
-
-            if (apCallStart.hasSessionId()) {
-                cloudSipCallStartEvent.setAssociationId(apCallStart.getSessionId());
-            }
-
-            if (apCallStart.hasWifiSessionId()) {
-                cloudSipCallStartEvent.setAssociationId(apCallStart.getWifiSessionId());
-            }
-
-            if (apCallStart.getCodecsCount() > 0) {
-                cloudSipCallStartEvent.setCodecs(apCallStart.getCodecsList());
-            }
-
-            if (apCallStart.hasChannel()) {
-                cloudSipCallStartEvent.setChannel(apCallStart.getChannel());
-            }
-
-            if (apCallStart.hasBand()) {
-                cloudSipCallStartEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
-                        .getRadioTypeFromOpensyncStatsRadioBandType(apCallStart.getBand()));
-            }
-
-            eventsList.add(cloudSipCallStartEvent);
-
-        }
-    }
-
-    protected void processRealTimeSipCallStopEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasCallStop()) {
-
-            CallStop apCallStop = videoVoiceReport.getCallStop();
-
-            RealTimeSipCallStopEvent cloudSipCallStopEvent = new RealTimeSipCallStopEvent(customerId, equipmentId,
-                    eventTimestamp);
-
-            if (apCallStop.hasCallDuration()) {
-
-                cloudSipCallStopEvent.setCallDuration(apCallStop.getCallDuration());
-
-            }
-
-            if (apCallStop.hasClientMac() && apCallStop.getClientMac().isValidUtf8()) {
-
-                cloudSipCallStopEvent.setClientMacAddress(MacAddress.valueOf(apCallStop.getClientMac().toStringUtf8()));
-
-            }
-
-            if (apCallStop.hasReason()) {
-
-                switch (apCallStop.getReason()) {
-                case BYE_OK:
-                    cloudSipCallStopEvent.setReason(SipCallStopReason.BYE_OK);
-                    break;
-                case CALL_DROPPED:
-                    cloudSipCallStopEvent.setReason(SipCallStopReason.DROPPED);
-                    break;
-
-                default:
-                    cloudSipCallStopEvent.setReason(SipCallStopReason.UNSUPPORTED);
-                }
-
-            }
-
-            if (apCallStop.hasSessionId()) {
-
-                cloudSipCallStopEvent.setAssociationId(apCallStop.getSessionId());
-
-            }
-
-            if (apCallStop.hasWifiSessionId()) {
-                cloudSipCallStopEvent.setSipCallId(apCallStop.getWifiSessionId());
-
-            }
-
-            if (apCallStop.getStatsCount() > 0) {
-                cloudSipCallStopEvent.setStatuses(processRtpFlowStats(apCallStop.getStatsList()));
-            }
-
-            if (apCallStop.hasProviderDomain()) {
-                cloudSipCallStopEvent.setProviderDomain(apCallStop.getProviderDomain());
-            }
-
-            if (apCallStop.getCodecsCount() > 0) {
-                cloudSipCallStopEvent.setCodecs(apCallStop.getCodecsList());
-            }
-
-            if (apCallStop.hasChannel()) {
-                cloudSipCallStopEvent.setChannel(apCallStop.getChannel());
-            }
-
-            if (apCallStop.hasBand()) {
-                cloudSipCallStopEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
-                        .getRadioTypeFromOpensyncStatsRadioBandType(apCallStop.getBand()));
-            }
-
-            eventsList.add(cloudSipCallStopEvent);
-
-        }
-    }
-
-    protected void processRtsStartEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasStreamVideoServer()) {
-            StreamingVideoServerDetected apStreamVideoServer = videoVoiceReport.getStreamVideoServer();
-            RealTimeStreamingStartEvent rtsStartEvent = new RealTimeStreamingStartEvent(customerId, equipmentId,
-                    eventTimestamp);
-            if (apStreamVideoServer.hasServerIp()) {
-                try {
-                    rtsStartEvent
-                            .setServerIp(InetAddress.getByAddress(apStreamVideoServer.getServerIp().toByteArray()));
-                } catch (UnknownHostException e) {
-                    LOG.error("Cannot get IP Address from {}", apStreamVideoServer.getServerIp(), e);
-                }
-            }
-            if (apStreamVideoServer.hasStreamingVideoType()) {
-                rtsStartEvent.setType(OvsdbToWlanCloudTypeMappingUtility
-                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoServer.getStreamingVideoType()));
-            }
-
-            if (apStreamVideoServer.hasServerDnsName()) {
-                rtsStartEvent.setServerDnsName(apStreamVideoServer.getServerDnsName());
-            }
-
-            if (apStreamVideoServer.hasClientMac() && apStreamVideoServer.getClientMac().isValidUtf8()) {
-                rtsStartEvent
-                        .setClientMacAddress(MacAddress.valueOf(apStreamVideoServer.getClientMac().toStringUtf8()));
-            }
-
-            if (apStreamVideoServer.hasSessionId()) {
-                rtsStartEvent.setSessionId(apStreamVideoServer.getSessionId());
-            }
-
-            if (apStreamVideoServer.hasVideoSessionId()) {
-                rtsStartEvent.setVideoSessionId(apStreamVideoServer.getVideoSessionId());
-            }
-
-            eventsList.add(rtsStartEvent);
-
-        }
-    }
-
-    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) {
-
-            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats cloudRtpStats = new com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats();
-
-            if (apRtpFlowStats.hasCodec()) {
-                cloudRtpStats.setCodec(apRtpFlowStats.getCodec());
-            }
-
-            if (apRtpFlowStats.hasBlockCodecs()) {
-                cloudRtpStats.setBlockCodecs(apRtpFlowStats.getBlockCodecs().toByteArray());
-            }
-
-            if (apRtpFlowStats.hasLatency()) {
-                cloudRtpStats.setLatency(apRtpFlowStats.getLatency());
-            }
-
-            if (apRtpFlowStats.hasRtpSeqFirst()) {
-                cloudRtpStats.setFirstRTPSeq(apRtpFlowStats.getRtpSeqFirst());
-            }
-
-            if (apRtpFlowStats.hasRtpSeqLast()) {
-                cloudRtpStats.setLastRTPSeq(apRtpFlowStats.getRtpSeqLast());
-            }
-
-            if (apRtpFlowStats.hasDirection()) {
-                switch (apRtpFlowStats.getDirection()) {
-                case RTP_DOWNSTREAM:
-                    cloudRtpStats.setDirection(
-                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.DOWNSTREAM);
-                    break;
-                case RTP_UPSTREAM:
-                    cloudRtpStats.setDirection(
-                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UPSTREAM);
-                    break;
-                default:
-                    cloudRtpStats.setDirection(
-                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UNSUPPORTED);
-                }
-            }
-
-            if (apRtpFlowStats.hasRtpFlowType()) {
-                switch (apRtpFlowStats.getRtpFlowType()) {
-                case RTP_VIDEO:
-                    cloudRtpStats
-                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VIDEO);
-                    break;
-                case RTP_VOICE:
-                    cloudRtpStats
-                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VOICE);
-                    break;
-                default:
-                    cloudRtpStats.setFlowType(
-                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.UNSUPPORTED);
-                    break;
-                }
-            }
-
-            if (apRtpFlowStats.hasJitter()) {
-                cloudRtpStats.setJitter(apRtpFlowStats.getJitter());
-            }
-
-            if (apRtpFlowStats.hasTotalPacketsSent()) {
-                cloudRtpStats.setTotalPacket(apRtpFlowStats.getTotalPacketsSent());
-            }
-
-            if (apRtpFlowStats.hasTotalPacketsLost()) {
-                cloudRtpStats.setTotalPacketLost(apRtpFlowStats.getTotalPacketsLost());
-            }
-
-            if (apRtpFlowStats.hasMosx100()) {
-                cloudRtpStats.setMosMultipliedBy100(apRtpFlowStats.getMosx100());
-            }
-
-            if (apRtpFlowStats.hasPacketLossConsec()) {
-                cloudRtpStats.setPacketLossConsecutive(apRtpFlowStats.getPacketLossConsec());
-            }
-
-            if (apRtpFlowStats.hasPacketLossPercent()) {
-                cloudRtpStats.setPacketLossPercentage(apRtpFlowStats.getPacketLossPercent());
-            }
-
-            cloudRtpFlowStatsList.add(cloudRtpStats);
-
-        }
-        return cloudRtpFlowStatsList;
-    }
-
-    protected void processRtsStartSessionEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasStreamVideoSessionStart()) {
-            StreamingVideoSessionStart apStreamVideoSessionStart = videoVoiceReport.getStreamVideoSessionStart();
-            RealTimeStreamingStartSessionEvent rtsStartSessionEvent = new RealTimeStreamingStartSessionEvent(customerId,
-                    equipmentId, eventTimestamp);
-            if (apStreamVideoSessionStart.hasClientMac() && apStreamVideoSessionStart.getClientMac().isValidUtf8()) {
-                rtsStartSessionEvent.setClientMacAddress(
-                        MacAddress.valueOf(apStreamVideoSessionStart.getClientMac().toStringUtf8()));
-            }
-
-            if (apStreamVideoSessionStart.hasServerIp()) {
-                try {
-                    rtsStartSessionEvent.setServerIp(
-                            InetAddress.getByAddress(apStreamVideoSessionStart.getServerIp().toByteArray()));
-                } catch (UnknownHostException e) {
-                    LOG.error("Cannot get IP Address from {}", apStreamVideoSessionStart.getServerIp(), e);
-
-                }
-            }
-
-            if (apStreamVideoSessionStart.hasSessionId()) {
-                rtsStartSessionEvent.setSessionId(apStreamVideoSessionStart.getSessionId());
-
-            }
-
-            if (apStreamVideoSessionStart.hasStreamingVideoType()) {
-                rtsStartSessionEvent.setType(OvsdbToWlanCloudTypeMappingUtility
-                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoSessionStart.getStreamingVideoType()));
-
-            }
-
-            if (apStreamVideoSessionStart.hasVideoSessionId()) {
-                rtsStartSessionEvent.setVideoSessionId(apStreamVideoSessionStart.getVideoSessionId());
-            }
-            eventsList.add(rtsStartSessionEvent);
-        }
-    }
-
-    protected void processRtsStopEvent(int customerId, long equipmentId, long eventTimestamp,
-            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
-        if (videoVoiceReport.hasStreamVideoStop()) {
-            StreamingVideoStop apStreamVideoStop = videoVoiceReport.getStreamVideoStop();
-            RealTimeStreamingStopEvent rtsStopEvent = new RealTimeStreamingStopEvent(customerId, equipmentId,
-                    eventTimestamp);
-            if (apStreamVideoStop.hasClientMac() && apStreamVideoStop.getClientMac().isValidUtf8()) {
-                rtsStopEvent.setClientMacAddress(MacAddress.valueOf(apStreamVideoStop.getClientMac().toStringUtf8()));
-            }
-
-            if (apStreamVideoStop.hasDurationSec()) {
-                rtsStopEvent.setDurationInSecs(apStreamVideoStop.getDurationSec());
-            }
-
-            if (apStreamVideoStop.hasServerIp()) {
-                try {
-                    rtsStopEvent.setServerIp(InetAddress.getByAddress(apStreamVideoStop.getServerIp().toByteArray()));
-                } catch (UnknownHostException e) {
-                    LOG.error("Cannot get IP Address from {}", apStreamVideoStop.getServerIp(), e);
-
-                }
-            }
-
-            if (apStreamVideoStop.hasSessionId()) {
-                rtsStopEvent.setSessionId(apStreamVideoStop.getSessionId());
-            }
-
-            if (apStreamVideoStop.hasStreamingVideoType()) {
-
-                rtsStopEvent.setType(OvsdbToWlanCloudTypeMappingUtility
-                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoStop.getStreamingVideoType()));
-
-            }
-
-            if (apStreamVideoStop.hasTotalBytes()) {
-                rtsStopEvent.setTotalBytes(apStreamVideoStop.getTotalBytes());
-            }
-
-            if (apStreamVideoStop.hasVideoSessionId()) {
-                rtsStopEvent.setVideoSessionId(apStreamVideoStop.getVideoSessionId());
-            }
-
-            eventsList.add(rtsStopEvent);
-
-        }
-    }
-
     void populateApNodeMetrics(List<ServiceMetric> metricRecordList, Report report, int customerId, long equipmentId,
             long locationId) {
         LOG.info("populateApNodeMetrics for Customer {} Equipment {}", customerId, equipmentId);

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

@@ -26,10 +26,22 @@ 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.equipment.realtime.RealTimeSipCallReportEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallStartEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeSipCallStopEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStartEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStartSessionEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeStreamingStopEvent;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.SIPCallReportReason;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.SipCallStopReason;
 import com.telecominfraproject.wlan.systemevent.models.SystemEvent;
 
+import sts.OpensyncStats;
 import sts.OpensyncStats.AssocType;
 import sts.OpensyncStats.CTReasonType;
+import sts.OpensyncStats.CallReport;
+import sts.OpensyncStats.CallStart;
+import sts.OpensyncStats.CallStop;
 import sts.OpensyncStats.ChannelSwitchReason;
 import sts.OpensyncStats.DeviceType;
 import sts.OpensyncStats.EventReport;
@@ -51,6 +63,11 @@ import sts.OpensyncStats.EventReport.DhcpOfferEvent;
 import sts.OpensyncStats.EventReport.DhcpRequestEvent;
 import sts.OpensyncStats.EventReport.DhcpTransaction;
 import sts.OpensyncStats.FrameType;
+import sts.OpensyncStats.RtpFlowStats;
+import sts.OpensyncStats.StreamingVideoServerDetected;
+import sts.OpensyncStats.StreamingVideoSessionStart;
+import sts.OpensyncStats.StreamingVideoStop;
+import sts.OpensyncStats.VideoVoiceReport;
 
 @org.springframework.context.annotation.Profile("opensync_cloud_config")
 @Component
@@ -564,4 +581,438 @@ public class RealtimeEventPublisher {
             }
         }
     }
+    
+    void publishSipCallEvents(int customerId, long equipmentId, List<VideoVoiceReport> sipCallReportList) {
+        // only in case it is not there, we will just use the time when we
+        // received the report/event
+        long eventTimestamp = System.currentTimeMillis();
+
+        List<SystemEvent> eventsList = new ArrayList<>();
+        for (VideoVoiceReport videoVoiceReport : sipCallReportList) {
+
+            if (videoVoiceReport.hasTimestampMs()) {
+                eventTimestamp = videoVoiceReport.getTimestampMs();
+            }
+
+            LOG.debug("Received VideoVoiceReport {} for SIP call", videoVoiceReport);
+
+            processRealTimeSipCallReportEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+            processRealTimeSipCallStartEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+            processRealTimeSipCallStopEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+            processRtsStartEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+            processRtsStartSessionEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+            processRtsStopEvent(customerId, equipmentId, eventTimestamp, eventsList, videoVoiceReport);
+
+        }
+
+        if (eventsList.size() > 0) {
+            cloudEventDispatcherInterface.publishEventsBulk(eventsList);
+        }
+
+    }
+
+    protected void processRealTimeSipCallReportEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasCallReport()) {
+
+            CallReport callReport = videoVoiceReport.getCallReport();
+
+            RealTimeSipCallReportEvent cloudSipCallReportEvent = new RealTimeSipCallReportEvent(customerId, equipmentId,
+                    eventTimestamp);
+
+            if (callReport.hasClientMac() && callReport.getClientMac().isValidUtf8()) {
+                cloudSipCallReportEvent
+                        .setClientMacAddress(MacAddress.valueOf(callReport.getClientMac().toStringUtf8()));
+            }
+            cloudSipCallReportEvent.setStatuses(processRtpFlowStats(callReport.getStatsList()));
+            cloudSipCallReportEvent.setEventType(RealTimeEventType.SipCallReport);
+
+            cloudSipCallReportEvent.setSipCallId(callReport.getWifiSessionId());
+            cloudSipCallReportEvent.setAssociationId(callReport.getSessionId());
+
+            if (callReport.hasReason()) {
+                cloudSipCallReportEvent.setReportReason(getCallReportReason(callReport.getReason()));
+            }
+            if (callReport.hasProviderDomain()) {
+                cloudSipCallReportEvent.setProviderDomain(callReport.getProviderDomain());
+            }
+
+            if (callReport.getCodecsCount() > 0) {
+                cloudSipCallReportEvent.setCodecs(callReport.getCodecsList());
+            }
+
+            if (callReport.hasChannel()) {
+                cloudSipCallReportEvent.setChannel(callReport.getChannel());
+            }
+
+            if (callReport.hasBand()) {
+                cloudSipCallReportEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                        .getRadioTypeFromOpensyncStatsRadioBandType(callReport.getBand()));
+            }
+
+            eventsList.add(cloudSipCallReportEvent);
+
+        }
+    }
+
+    private SIPCallReportReason getCallReportReason(CallReport.CallReportReason reason) {
+        if (reason != null) {
+            switch (reason) {
+            case ROAMED_TO:
+                return SIPCallReportReason.ROAMED_TO;
+            case GOT_PUBLISH:
+                return SIPCallReportReason.GOT_PUBLISH;
+            case ROAMED_FROM:
+                return SIPCallReportReason.ROAMED_FROM;
+            default:
+                return SIPCallReportReason.UNSUPPORTED;
+            }
+        }
+        return SIPCallReportReason.UNSUPPORTED;
+    }
+
+    protected void processRealTimeSipCallStartEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasCallStart()) {
+
+            CallStart apCallStart = videoVoiceReport.getCallStart();
+
+            RealTimeSipCallStartEvent cloudSipCallStartEvent = new RealTimeSipCallStartEvent(customerId, equipmentId,
+                    eventTimestamp);
+
+            if (apCallStart.hasClientMac() && apCallStart.getClientMac().isValidUtf8()) {
+                cloudSipCallStartEvent
+                        .setClientMacAddress(MacAddress.valueOf(apCallStart.getClientMac().toStringUtf8()));
+            }
+
+            if (apCallStart.hasDeviceInfo()) {
+                cloudSipCallStartEvent.setDeviceInfo(apCallStart.getDeviceInfo());
+            }
+
+            if (apCallStart.hasProviderDomain()) {
+                cloudSipCallStartEvent.setProviderDomain(apCallStart.getProviderDomain());
+            }
+
+            if (apCallStart.hasSessionId()) {
+                cloudSipCallStartEvent.setAssociationId(apCallStart.getSessionId());
+            }
+
+            if (apCallStart.hasWifiSessionId()) {
+                cloudSipCallStartEvent.setAssociationId(apCallStart.getWifiSessionId());
+            }
+
+            if (apCallStart.getCodecsCount() > 0) {
+                cloudSipCallStartEvent.setCodecs(apCallStart.getCodecsList());
+            }
+
+            if (apCallStart.hasChannel()) {
+                cloudSipCallStartEvent.setChannel(apCallStart.getChannel());
+            }
+
+            if (apCallStart.hasBand()) {
+                cloudSipCallStartEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                        .getRadioTypeFromOpensyncStatsRadioBandType(apCallStart.getBand()));
+            }
+
+            eventsList.add(cloudSipCallStartEvent);
+
+        }
+    }
+
+    protected void processRealTimeSipCallStopEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasCallStop()) {
+
+            CallStop apCallStop = videoVoiceReport.getCallStop();
+
+            RealTimeSipCallStopEvent cloudSipCallStopEvent = new RealTimeSipCallStopEvent(customerId, equipmentId,
+                    eventTimestamp);
+
+            if (apCallStop.hasCallDuration()) {
+
+                cloudSipCallStopEvent.setCallDuration(apCallStop.getCallDuration());
+
+            }
+
+            if (apCallStop.hasClientMac() && apCallStop.getClientMac().isValidUtf8()) {
+
+                cloudSipCallStopEvent.setClientMacAddress(MacAddress.valueOf(apCallStop.getClientMac().toStringUtf8()));
+
+            }
+
+            if (apCallStop.hasReason()) {
+
+                switch (apCallStop.getReason()) {
+                case BYE_OK:
+                    cloudSipCallStopEvent.setReason(SipCallStopReason.BYE_OK);
+                    break;
+                case CALL_DROPPED:
+                    cloudSipCallStopEvent.setReason(SipCallStopReason.DROPPED);
+                    break;
+
+                default:
+                    cloudSipCallStopEvent.setReason(SipCallStopReason.UNSUPPORTED);
+                }
+
+            }
+
+            if (apCallStop.hasSessionId()) {
+
+                cloudSipCallStopEvent.setAssociationId(apCallStop.getSessionId());
+
+            }
+
+            if (apCallStop.hasWifiSessionId()) {
+                cloudSipCallStopEvent.setSipCallId(apCallStop.getWifiSessionId());
+
+            }
+
+            if (apCallStop.getStatsCount() > 0) {
+                cloudSipCallStopEvent.setStatuses(processRtpFlowStats(apCallStop.getStatsList()));
+            }
+
+            if (apCallStop.hasProviderDomain()) {
+                cloudSipCallStopEvent.setProviderDomain(apCallStop.getProviderDomain());
+            }
+
+            if (apCallStop.getCodecsCount() > 0) {
+                cloudSipCallStopEvent.setCodecs(apCallStop.getCodecsList());
+            }
+
+            if (apCallStop.hasChannel()) {
+                cloudSipCallStopEvent.setChannel(apCallStop.getChannel());
+            }
+
+            if (apCallStop.hasBand()) {
+                cloudSipCallStopEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
+                        .getRadioTypeFromOpensyncStatsRadioBandType(apCallStop.getBand()));
+            }
+
+            eventsList.add(cloudSipCallStopEvent);
+
+        }
+    }
+
+    protected void processRtsStartEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasStreamVideoServer()) {
+            StreamingVideoServerDetected apStreamVideoServer = videoVoiceReport.getStreamVideoServer();
+            RealTimeStreamingStartEvent rtsStartEvent = new RealTimeStreamingStartEvent(customerId, equipmentId,
+                    eventTimestamp);
+            if (apStreamVideoServer.hasServerIp()) {
+                try {
+                    rtsStartEvent
+                            .setServerIp(InetAddress.getByAddress(apStreamVideoServer.getServerIp().toByteArray()));
+                } catch (UnknownHostException e) {
+                    LOG.error("Cannot get IP Address from {}", apStreamVideoServer.getServerIp(), e);
+                }
+            }
+            if (apStreamVideoServer.hasStreamingVideoType()) {
+                rtsStartEvent.setType(OvsdbToWlanCloudTypeMappingUtility
+                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoServer.getStreamingVideoType()));
+            }
+
+            if (apStreamVideoServer.hasServerDnsName()) {
+                rtsStartEvent.setServerDnsName(apStreamVideoServer.getServerDnsName());
+            }
+
+            if (apStreamVideoServer.hasClientMac() && apStreamVideoServer.getClientMac().isValidUtf8()) {
+                rtsStartEvent
+                        .setClientMacAddress(MacAddress.valueOf(apStreamVideoServer.getClientMac().toStringUtf8()));
+            }
+
+            if (apStreamVideoServer.hasSessionId()) {
+                rtsStartEvent.setSessionId(apStreamVideoServer.getSessionId());
+            }
+
+            if (apStreamVideoServer.hasVideoSessionId()) {
+                rtsStartEvent.setVideoSessionId(apStreamVideoServer.getVideoSessionId());
+            }
+
+            eventsList.add(rtsStartEvent);
+
+        }
+    }
+
+    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) {
+
+            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats cloudRtpStats = new com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowStats();
+
+            if (apRtpFlowStats.hasCodec()) {
+                cloudRtpStats.setCodec(apRtpFlowStats.getCodec());
+            }
+
+            if (apRtpFlowStats.hasBlockCodecs()) {
+                cloudRtpStats.setBlockCodecs(apRtpFlowStats.getBlockCodecs().toByteArray());
+            }
+
+            if (apRtpFlowStats.hasLatency()) {
+                cloudRtpStats.setLatency(apRtpFlowStats.getLatency());
+            }
+
+            if (apRtpFlowStats.hasRtpSeqFirst()) {
+                cloudRtpStats.setFirstRTPSeq(apRtpFlowStats.getRtpSeqFirst());
+            }
+
+            if (apRtpFlowStats.hasRtpSeqLast()) {
+                cloudRtpStats.setLastRTPSeq(apRtpFlowStats.getRtpSeqLast());
+            }
+
+            if (apRtpFlowStats.hasDirection()) {
+                switch (apRtpFlowStats.getDirection()) {
+                case RTP_DOWNSTREAM:
+                    cloudRtpStats.setDirection(
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.DOWNSTREAM);
+                    break;
+                case RTP_UPSTREAM:
+                    cloudRtpStats.setDirection(
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UPSTREAM);
+                    break;
+                default:
+                    cloudRtpStats.setDirection(
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowDirection.UNSUPPORTED);
+                }
+            }
+
+            if (apRtpFlowStats.hasRtpFlowType()) {
+                switch (apRtpFlowStats.getRtpFlowType()) {
+                case RTP_VIDEO:
+                    cloudRtpStats
+                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VIDEO);
+                    break;
+                case RTP_VOICE:
+                    cloudRtpStats
+                            .setFlowType(com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.VOICE);
+                    break;
+                default:
+                    cloudRtpStats.setFlowType(
+                            com.telecominfraproject.wlan.systemevent.equipment.realtime.RtpFlowType.UNSUPPORTED);
+                    break;
+                }
+            }
+
+            if (apRtpFlowStats.hasJitter()) {
+                cloudRtpStats.setJitter(apRtpFlowStats.getJitter());
+            }
+
+            if (apRtpFlowStats.hasTotalPacketsSent()) {
+                cloudRtpStats.setTotalPacket(apRtpFlowStats.getTotalPacketsSent());
+            }
+
+            if (apRtpFlowStats.hasTotalPacketsLost()) {
+                cloudRtpStats.setTotalPacketLost(apRtpFlowStats.getTotalPacketsLost());
+            }
+
+            if (apRtpFlowStats.hasMosx100()) {
+                cloudRtpStats.setMosMultipliedBy100(apRtpFlowStats.getMosx100());
+            }
+
+            if (apRtpFlowStats.hasPacketLossConsec()) {
+                cloudRtpStats.setPacketLossConsecutive(apRtpFlowStats.getPacketLossConsec());
+            }
+
+            if (apRtpFlowStats.hasPacketLossPercent()) {
+                cloudRtpStats.setPacketLossPercentage(apRtpFlowStats.getPacketLossPercent());
+            }
+
+            cloudRtpFlowStatsList.add(cloudRtpStats);
+
+        }
+        return cloudRtpFlowStatsList;
+    }
+
+    protected void processRtsStartSessionEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasStreamVideoSessionStart()) {
+            StreamingVideoSessionStart apStreamVideoSessionStart = videoVoiceReport.getStreamVideoSessionStart();
+            RealTimeStreamingStartSessionEvent rtsStartSessionEvent = new RealTimeStreamingStartSessionEvent(customerId,
+                    equipmentId, eventTimestamp);
+            if (apStreamVideoSessionStart.hasClientMac() && apStreamVideoSessionStart.getClientMac().isValidUtf8()) {
+                rtsStartSessionEvent.setClientMacAddress(
+                        MacAddress.valueOf(apStreamVideoSessionStart.getClientMac().toStringUtf8()));
+            }
+
+            if (apStreamVideoSessionStart.hasServerIp()) {
+                try {
+                    rtsStartSessionEvent.setServerIp(
+                            InetAddress.getByAddress(apStreamVideoSessionStart.getServerIp().toByteArray()));
+                } catch (UnknownHostException e) {
+                    LOG.error("Cannot get IP Address from {}", apStreamVideoSessionStart.getServerIp(), e);
+
+                }
+            }
+
+            if (apStreamVideoSessionStart.hasSessionId()) {
+                rtsStartSessionEvent.setSessionId(apStreamVideoSessionStart.getSessionId());
+
+            }
+
+            if (apStreamVideoSessionStart.hasStreamingVideoType()) {
+                rtsStartSessionEvent.setType(OvsdbToWlanCloudTypeMappingUtility
+                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoSessionStart.getStreamingVideoType()));
+
+            }
+
+            if (apStreamVideoSessionStart.hasVideoSessionId()) {
+                rtsStartSessionEvent.setVideoSessionId(apStreamVideoSessionStart.getVideoSessionId());
+            }
+            eventsList.add(rtsStartSessionEvent);
+        }
+    }
+
+    protected void processRtsStopEvent(int customerId, long equipmentId, long eventTimestamp,
+            List<SystemEvent> eventsList, VideoVoiceReport videoVoiceReport) {
+        if (videoVoiceReport.hasStreamVideoStop()) {
+            StreamingVideoStop apStreamVideoStop = videoVoiceReport.getStreamVideoStop();
+            RealTimeStreamingStopEvent rtsStopEvent = new RealTimeStreamingStopEvent(customerId, equipmentId,
+                    eventTimestamp);
+            if (apStreamVideoStop.hasClientMac() && apStreamVideoStop.getClientMac().isValidUtf8()) {
+                rtsStopEvent.setClientMacAddress(MacAddress.valueOf(apStreamVideoStop.getClientMac().toStringUtf8()));
+            }
+
+            if (apStreamVideoStop.hasDurationSec()) {
+                rtsStopEvent.setDurationInSecs(apStreamVideoStop.getDurationSec());
+            }
+
+            if (apStreamVideoStop.hasServerIp()) {
+                try {
+                    rtsStopEvent.setServerIp(InetAddress.getByAddress(apStreamVideoStop.getServerIp().toByteArray()));
+                } catch (UnknownHostException e) {
+                    LOG.error("Cannot get IP Address from {}", apStreamVideoStop.getServerIp(), e);
+
+                }
+            }
+
+            if (apStreamVideoStop.hasSessionId()) {
+                rtsStopEvent.setSessionId(apStreamVideoStop.getSessionId());
+            }
+
+            if (apStreamVideoStop.hasStreamingVideoType()) {
+
+                rtsStopEvent.setType(OvsdbToWlanCloudTypeMappingUtility
+                        .getCloudStreamingVideoTypeFromApReport(apStreamVideoStop.getStreamingVideoType()));
+
+            }
+
+            if (apStreamVideoStop.hasTotalBytes()) {
+                rtsStopEvent.setTotalBytes(apStreamVideoStop.getTotalBytes());
+            }
+
+            if (apStreamVideoStop.hasVideoSessionId()) {
+                rtsStopEvent.setVideoSessionId(apStreamVideoStop.getVideoSessionId());
+            }
+
+            eventsList.add(rtsStopEvent);
+
+        }
+    }
+
 }

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

@@ -57,7 +57,6 @@ import com.telecominfraproject.wlan.status.equipment.report.models.ActiveBSSIDs;
 import com.telecominfraproject.wlan.status.models.Status;
 import com.telecominfraproject.wlan.status.models.StatusDataType;
 
-import sts.OpensyncStats;
 import sts.OpensyncStats.AssocType;
 import sts.OpensyncStats.ChannelSwitchReason;
 import sts.OpensyncStats.Client;
@@ -260,60 +259,6 @@ public class MqttStatsPublisherTest {
         // TODO: implement me when wcs stats reports supported
     }
 
-    // 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()
-                .setCallStart(getDefaultCallStart());
-        OpensyncStats.VideoVoiceReport.Builder callReportGotPublishVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH,
-                        121, 1028, 1316, 1888, 298, 2, 100, 200));
-        OpensyncStats.VideoVoiceReport.Builder callReportRoamedToVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO,
-                        123, 1020, 1116, 1345, 223, 0, 102, 203));
-        OpensyncStats.VideoVoiceReport.Builder callReportRoamedFromVoiceReportBuilder = OpensyncStats.VideoVoiceReport
-                .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)
-                .addVideoVoiceReport(callReportGotPublishVoiceReportBuilder)
-                .addVideoVoiceReport(callReportRoamedFromVoiceReportBuilder)
-                .addVideoVoiceReport(callReportRoamedToVoiceReportBuilder)
-                .addVideoVoiceReport(callStopVoiceReportBuilder).setNodeID("1").build();
-
-        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
-    @Test
-    public void testProcessMqttMessageStringOneVideoVoiceReport() {
-
-        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.ROAMED_FROM, 122, 1029,
-                        1300, 1234, 111, 3, 101, 201))
-                .setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO, 123, 1020,
-                        1116, 1345, 223, 0, 102, 203))
-                .setCallStart(getDefaultCallStart()).setCallStop(getDefaultCallStop());
-        Report oneVoiceReportWithMultipleCallsInOneReport = Report.getDefaultInstance().toBuilder()
-                .addVideoVoiceReport(videoVoiceReportBuilder).setNodeID("1").build();
-
-        opensyncExternalIntegrationMqttProcessor.populateSipCallReport(null, oneVoiceReportWithMultipleCallsInOneReport,
-                1, 2L, "TestAP", 12L);
-
-        Mockito.verify(equipmentMetricsCollectorInterface, Mockito.times(1)).publishEventsBulk(Mockito.anyList());
-    }
-
     @Test
     public void testpopulateNetworkProbeMetrics() throws Exception {
 
@@ -371,75 +316,6 @@ public class MqttStatsPublisherTest {
         bssidList.add(activeBssid3);
         return bssidList;
     }
-
-    private OpensyncStats.CallStart getDefaultCallStart() {
-        OpensyncStats.CallStart.Builder callStartBuilder = OpensyncStats.CallStart.newBuilder();
-        callStartBuilder.setBand(RadioBandType.BAND5G);
-        callStartBuilder.setChannel(40);
-        callStartBuilder.addCodecs("110 opus/48000/2");
-        callStartBuilder.addCodecs("102 iLBC/8000");
-        callStartBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
-        callStartBuilder.setDeviceInfo("Test Device");
-        callStartBuilder.setProviderDomain("skype");
-        callStartBuilder.setSessionId(123L);
-        callStartBuilder.setWifiSessionId(1234L);
-
-        return callStartBuilder.build();
-    }
-
-    private OpensyncStats.CallStop getDefaultCallStop() {
-        OpensyncStats.CallStop.Builder callStopBuilder = OpensyncStats.CallStop.newBuilder();
-        callStopBuilder.setBand(RadioBandType.BAND5G);
-        callStopBuilder.setChannel(40);
-        callStopBuilder.addCodecs("110 opus/48000/2");
-        callStopBuilder.addCodecs("102 iLBC/8000");
-        callStopBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
-        callStopBuilder.setCallDuration(1230);
-        callStopBuilder.setProviderDomain("skype");
-        callStopBuilder.setSessionId(123L);
-        callStopBuilder.setWifiSessionId(1234L);
-        callStopBuilder.setReason(OpensyncStats.CallStop.CallStopReason.BYE_OK);
-        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) {
-        OpensyncStats.RtpFlowStats.Builder rtpFlowStatsBuilder = OpensyncStats.RtpFlowStats.newBuilder();
-        rtpFlowStatsBuilder.setCodec(codec);
-        rtpFlowStatsBuilder.setBlockCodecs(ByteString.copyFrom(new byte[] { (byte) 0xe6, 0x1 }));
-        rtpFlowStatsBuilder.setDirection(OpensyncStats.RtpFlowStats.RtpFlowDirection.RTP_DOWNSTREAM);
-        rtpFlowStatsBuilder.setRtpFlowType(OpensyncStats.RtpFlowStats.RtpFlowType.RTP_VOICE);
-        rtpFlowStatsBuilder.setJitter(jitter);
-        rtpFlowStatsBuilder.setLatency(latency);
-        rtpFlowStatsBuilder.setTotalPacketsSent(totalPackets);
-        rtpFlowStatsBuilder.setTotalPacketsLost(totalPacketsLost);
-        rtpFlowStatsBuilder.setMosx100(mos);
-        rtpFlowStatsBuilder.setRtpSeqFirst(firstRtpSeq);
-        rtpFlowStatsBuilder.setRtpSeqLast(lastRtpSeq);
-
-        return rtpFlowStatsBuilder.build();
-    }
-
-    private OpensyncStats.CallReport getDefaultCallReport(OpensyncStats.CallReport.CallReportReason reason, int codec,
-            int jitter, int latency, int totalPackets, int totalPacketsLost, int mos, int firstRtpSeq, int lastRtpSeq) {
-        OpensyncStats.CallReport.Builder callReportBuilder = OpensyncStats.CallReport.newBuilder();
-        callReportBuilder.setBand(RadioBandType.BAND5G);
-        callReportBuilder.setChannel(40);
-        callReportBuilder.addCodecs("110 opus/48000/2");
-        callReportBuilder.addCodecs("102 iLBC/8000");
-        callReportBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
-        callReportBuilder.setProviderDomain("skype");
-        callReportBuilder.setSessionId(123L);
-        callReportBuilder.setWifiSessionId(1234L);
-        callReportBuilder.setReason(reason);
-        callReportBuilder.addStats(
-                getRtpFlowStats(codec, jitter, latency, totalPackets, totalPacketsLost, mos, firstRtpSeq, lastRtpSeq));
-
-        return callReportBuilder.build();
-    }
-
     private List<EventReport> getOpensyncStatsEventReportsList() {
 
         List<ClientAssocEvent> clientAssocEventList = new ArrayList<>();

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

@@ -0,0 +1,222 @@
+package com.telecominfraproject.wlan.opensync.external.integration.utils;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.MockitoSession;
+import org.mockito.quality.Strictness;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import com.google.protobuf.ByteString;
+import com.telecominfraproject.wlan.cloudeventdispatcher.CloudEventDispatcherInterface;
+import com.telecominfraproject.wlan.equipment.EquipmentServiceInterface;
+import com.telecominfraproject.wlan.opensync.external.integration.OpensyncExternalIntegrationCloud;
+
+import sts.OpensyncStats;
+import sts.OpensyncStats.RadioBandType;
+import sts.OpensyncStats.Report;
+@RunWith(SpringRunner.class)
+@ActiveProfiles(profiles = { "integration_test", })
+@SpringBootTest(webEnvironment = WebEnvironment.NONE, classes = RealtimeEventPublisherTest.class)
+@Import(value = { OpensyncExternalIntegrationCloud.class, RealtimeEventPublisherTest.Config.class,
+
+})
+public class RealtimeEventPublisherTest {
+    
+    @MockBean
+    private CloudEventDispatcherInterface cloudEventDispatcherInterface;
+
+    @MockBean
+    private EquipmentServiceInterface equipmentServiceInterface;
+
+    @Autowired
+    RealtimeEventPublisher realtimeEventPublisher;
+    MockitoSession mockito;
+
+    @Configuration
+    static class Config {
+
+        @Bean
+        public RealtimeEventPublisher realtimeEventPublisher() {
+            return new RealtimeEventPublisher();
+        }
+
+    }
+
+    @Before
+    public void setUp() throws Exception {
+
+        mockito = Mockito.mockitoSession().initMocks(this).strictness(Strictness.STRICT_STUBS).startMocking();
+
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        mockito.finishMocking();
+
+    }
+
+    @Ignore
+    void testPublishChannelHopEvents() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientConnectSuccessEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientDisconnectEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientAuthSystemEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientAssocEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientFailureEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientFirstDataEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientIdEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientIpEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishClientTimeoutEvent() {
+        // TODO: implement
+    }
+
+    @Ignore
+    void testPublishDhcpTransactionEvents() {
+        // TODO: implement
+    }
+
+    @Test
+    public void testPublishSipCallEvents() throws Exception {
+
+        OpensyncStats.VideoVoiceReport.Builder callStartVoiceReportBuilder = OpensyncStats.VideoVoiceReport.newBuilder()
+                .setCallStart(getDefaultCallStart());
+        OpensyncStats.VideoVoiceReport.Builder callReportGotPublishVoiceReportBuilder = OpensyncStats.VideoVoiceReport
+                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.GOT_PUBLISH,
+                        121, 1028, 1316, 1888, 298, 2, 100, 200));
+        OpensyncStats.VideoVoiceReport.Builder callReportRoamedToVoiceReportBuilder = OpensyncStats.VideoVoiceReport
+                .newBuilder().setCallReport(getDefaultCallReport(OpensyncStats.CallReport.CallReportReason.ROAMED_TO,
+                        123, 1020, 1116, 1345, 223, 0, 102, 203));
+        OpensyncStats.VideoVoiceReport.Builder callReportRoamedFromVoiceReportBuilder = OpensyncStats.VideoVoiceReport
+                .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)
+                .addVideoVoiceReport(callReportGotPublishVoiceReportBuilder)
+                .addVideoVoiceReport(callReportRoamedFromVoiceReportBuilder)
+                .addVideoVoiceReport(callReportRoamedToVoiceReportBuilder)
+                .addVideoVoiceReport(callStopVoiceReportBuilder).setNodeID("1").build();
+
+        realtimeEventPublisher.publishSipCallEvents(1, 2L,multipleVoiceReportsInOneReport.getVideoVoiceReportList());
+
+        Mockito.verify(cloudEventDispatcherInterface, Mockito.times(1)).publishEventsBulk(Mockito.anyList());
+    }
+   
+    private OpensyncStats.CallStart getDefaultCallStart() {
+        OpensyncStats.CallStart.Builder callStartBuilder = OpensyncStats.CallStart.newBuilder();
+        callStartBuilder.setBand(RadioBandType.BAND5G);
+        callStartBuilder.setChannel(40);
+        callStartBuilder.addCodecs("110 opus/48000/2");
+        callStartBuilder.addCodecs("102 iLBC/8000");
+        callStartBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
+        callStartBuilder.setDeviceInfo("Test Device");
+        callStartBuilder.setProviderDomain("skype");
+        callStartBuilder.setSessionId(123L);
+        callStartBuilder.setWifiSessionId(1234L);
+
+        return callStartBuilder.build();
+    }
+
+    private OpensyncStats.CallStop getDefaultCallStop() {
+        OpensyncStats.CallStop.Builder callStopBuilder = OpensyncStats.CallStop.newBuilder();
+        callStopBuilder.setBand(RadioBandType.BAND5G);
+        callStopBuilder.setChannel(40);
+        callStopBuilder.addCodecs("110 opus/48000/2");
+        callStopBuilder.addCodecs("102 iLBC/8000");
+        callStopBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
+        callStopBuilder.setCallDuration(1230);
+        callStopBuilder.setProviderDomain("skype");
+        callStopBuilder.setSessionId(123L);
+        callStopBuilder.setWifiSessionId(1234L);
+        callStopBuilder.setReason(OpensyncStats.CallStop.CallStopReason.BYE_OK);
+        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) {
+        OpensyncStats.RtpFlowStats.Builder rtpFlowStatsBuilder = OpensyncStats.RtpFlowStats.newBuilder();
+        rtpFlowStatsBuilder.setCodec(codec);
+        rtpFlowStatsBuilder.setBlockCodecs(ByteString.copyFrom(new byte[] { (byte) 0xe6, 0x1 }));
+        rtpFlowStatsBuilder.setDirection(OpensyncStats.RtpFlowStats.RtpFlowDirection.RTP_DOWNSTREAM);
+        rtpFlowStatsBuilder.setRtpFlowType(OpensyncStats.RtpFlowStats.RtpFlowType.RTP_VOICE);
+        rtpFlowStatsBuilder.setJitter(jitter);
+        rtpFlowStatsBuilder.setLatency(latency);
+        rtpFlowStatsBuilder.setTotalPacketsSent(totalPackets);
+        rtpFlowStatsBuilder.setTotalPacketsLost(totalPacketsLost);
+        rtpFlowStatsBuilder.setMosx100(mos);
+        rtpFlowStatsBuilder.setRtpSeqFirst(firstRtpSeq);
+        rtpFlowStatsBuilder.setRtpSeqLast(lastRtpSeq);
+
+        return rtpFlowStatsBuilder.build();
+    }
+
+    private OpensyncStats.CallReport getDefaultCallReport(OpensyncStats.CallReport.CallReportReason reason, int codec,
+            int jitter, int latency, int totalPackets, int totalPacketsLost, int mos, int firstRtpSeq, int lastRtpSeq) {
+        OpensyncStats.CallReport.Builder callReportBuilder = OpensyncStats.CallReport.newBuilder();
+        callReportBuilder.setBand(RadioBandType.BAND5G);
+        callReportBuilder.setChannel(40);
+        callReportBuilder.addCodecs("110 opus/48000/2");
+        callReportBuilder.addCodecs("102 iLBC/8000");
+        callReportBuilder.setClientMac(ByteString.copyFrom("C0:9A:D0:76:A9:69".getBytes()));
+        callReportBuilder.setProviderDomain("skype");
+        callReportBuilder.setSessionId(123L);
+        callReportBuilder.setWifiSessionId(1234L);
+        callReportBuilder.setReason(reason);
+        callReportBuilder.addStats(
+                getRtpFlowStats(codec, jitter, latency, totalPackets, totalPacketsLost, mos, firstRtpSeq, lastRtpSeq));
+
+        return callReportBuilder.build();
+    }
+
+
+}