[WIFI-3166] Session ID is a -ve number in opensyncgw logs and on UI logs

Signed-off-by: Mike Hansen <mike.hansen@connectus.ai>

opensync-ext-cloud/pom.xml

@@ -11,11 +11,11 @@
 	<name>opensync-ext-cloud</name>
 	<description>Configuration interface that provides config from the cloud services.</description>
 	<dependencies>
-<!-- 		<dependency> -->
-<!-- 			<groupId>com.telecominfraproject.wlan</groupId> -->
-<!-- 			<artifactId>opensync-ext-interface</artifactId> -->
-<!-- 			<version>0.0.1-SNAPSHOT</version> -->
-<!-- 		</dependency> -->
+		<dependency>
+			<groupId>com.telecominfraproject.wlan</groupId>
+			<artifactId>opensync-ext-interface</artifactId>
+			<version>1.2.0-SNAPSHOT</version>
+		</dependency>
 		<dependency>
 			<groupId>com.telecominfraproject.wlan</groupId>
 			<artifactId>opensync-gateway</artifactId>

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

@@ -76,7 +76,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.external.integration.utils.StatsPublisherInterface;
 import com.telecominfraproject.wlan.opensync.ovsdb.dao.models.enumerations.DhcpFpDeviceType;
 import com.telecominfraproject.wlan.opensync.util.OvsdbStringConstants;
 import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
@@ -103,6 +103,7 @@ import com.telecominfraproject.wlan.status.equipment.models.EquipmentProtocolSta
 import com.telecominfraproject.wlan.status.equipment.models.EquipmentUpgradeState;
 import com.telecominfraproject.wlan.status.equipment.models.EquipmentUpgradeState.FailureReason;
 import com.telecominfraproject.wlan.status.equipment.models.EquipmentUpgradeStatusData;
+import com.telecominfraproject.wlan.status.equipment.models.LedStatus;
 import com.telecominfraproject.wlan.status.equipment.models.VLANStatusData;
 import com.telecominfraproject.wlan.status.equipment.report.models.ActiveBSSID;
 import com.telecominfraproject.wlan.status.equipment.report.models.ActiveBSSIDs;
@@ -115,12 +116,8 @@ 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.ApcElectionEvent;
-import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeEventType;
 import com.telecominfraproject.wlan.systemevent.equipment.realtime.ApcElectionEvent.ApcMode;
-
-import sts.OpensyncStats.Report;
-import traffic.NetworkMetadata.FlowReport;
-import wc.stats.IpDnsTelemetry.WCStatsReport;
+import com.telecominfraproject.wlan.systemevent.equipment.realtime.RealTimeEventType;
 
 @org.springframework.context.annotation.Profile("opensync_cloud_config")
 @Component
@@ -147,7 +144,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     @Autowired
     private FirmwareServiceInterface firmwareServiceInterface;
     @Autowired
-    private MqttStatsPublisher mqttMessageProcessor;
+    private StatsPublisherInterface mqttMessageProcessor;
     @Autowired
     private AlarmServiceInterface alarmServiceInterface;
 
@@ -182,7 +179,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
 
     @Override
     public void apConnected(String apId, ConnectNodeInfo connectNodeInfo) {
-    	
+
         Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
 
         try {
@@ -213,10 +210,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 }
 
                 long locationId = autoProvisionedLocationId;
-                if ((customer.getDetails() != null) && (customer.getDetails().getAutoProvisioning() != null)
-                        && customer.getDetails().getAutoProvisioning().isEnabled()) {
-                    locationId = customer.getDetails().getAutoProvisioning().getLocationId();
-                }
+                locationId = customer.getDetails().getAutoProvisioning().getLocationId();
 
                 try {
                     Location location = locationServiceInterface.get(locationId);
@@ -224,7 +218,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 } catch (Exception e) {
                     LOG.error("Cannot auto-provision equipment because customer location with id {} cannot be found", locationId);
                     throw new IllegalStateException("Cannot auto-provision equipment because customer location cannot be found : " + locationId);
-
                 }
 
                 ce.setSerial(connectNodeInfo.serialNumber);
@@ -253,7 +246,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                     }
                     if (!radioType.equals(RadioType.UNSUPPORTED)) {
                         advancedRadioConfiguration = RadioConfiguration.createWithDefaults(radioType);
-
                         advancedRadioMap.put(radioType, advancedRadioConfiguration);
                         radioConfiguration = ElementRadioConfiguration.createWithDefaults(radioType);
                         radioMap.put(radioType, radioConfiguration);
@@ -389,7 +381,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 }
 
             }
-
+            
             EquipmentRoutingRecord equipmentRoutingRecord = gatewayController.registerCustomerEquipment(ce.getName(), ce.getCustomerId(), ce.getId());
 
             // Status and client cleanup, when AP reconnects or has been
@@ -406,7 +398,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
             ovsdbSession.setRoutingId(equipmentRoutingRecord.getId());
             ovsdbSession.setEquipmentId(ce.getId());
-            ovsdbSession.setCustomerId(ce.getCustomerId());
 
             LOG.debug("Equipment {}", ce);
             LOG.info("AP {} got connected to the gateway", apId);
@@ -501,7 +492,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     private void updateApStatus(Equipment ce, ConnectNodeInfo connectNodeInfo) {
 
         try {
-
             Status statusRecord = statusServiceInterface.getOrNull(ce.getCustomerId(), ce.getId(), StatusDataType.EQUIPMENT_ADMIN);
             if (statusRecord == null) {
                 statusRecord = new Status();
@@ -512,6 +502,9 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             }
 
             ((EquipmentAdminStatusData) statusRecord.getDetails()).setStatusCode(StatusCode.normal);
+            if (((EquipmentAdminStatusData) statusRecord.getDetails()).getLedStatus() == null) {
+                ((EquipmentAdminStatusData) statusRecord.getDetails()).setLedStatus(LedStatus.UNKNOWN);
+            }
             // Update the equipment admin status
             statusRecord = statusServiceInterface.update(statusRecord);
 
@@ -891,20 +884,20 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     private void updateApDisconnectedStatus(String apId, Equipment ce) {
         LOG.info("updateApDisconnectedStatus disconnected AP {}", apId);
         try {
-        	Alarm disconnectedAlarm = new Alarm();
+            Alarm disconnectedAlarm = new Alarm();
             disconnectedAlarm.setCustomerId(ce.getCustomerId());
             disconnectedAlarm.setEquipmentId(ce.getId());
             disconnectedAlarm.setAlarmCode(AlarmCode.Disconnected);
-            
+
             AlarmDetails alarmDetails = new AlarmDetails();
-    		alarmDetails.setMessage(AlarmCode.Disconnected.getDescription());
-    		alarmDetails.setAffectedEquipmentIds(List.of(ce.getId()));
-    		alarmDetails.setGeneratedBy(apId);
-    		
-    		disconnectedAlarm.setDetails(alarmDetails);
-            
-    		alarmServiceInterface.create(disconnectedAlarm);
-            
+            alarmDetails.setMessage(AlarmCode.Disconnected.getDescription());
+            alarmDetails.setAffectedEquipmentIds(List.of(ce.getId()));
+            alarmDetails.setGeneratedBy(apId);
+
+            disconnectedAlarm.setDetails(alarmDetails);
+
+            alarmServiceInterface.create(disconnectedAlarm);
+
             Status statusRecord = new Status();
             statusRecord.setCustomerId(ce.getCustomerId());
             statusRecord.setEquipmentId(ce.getId());
@@ -969,11 +962,11 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             if (ovsdbSession == null) {
                 throw new IllegalStateException("AP is not connected " + apId);
             }
-            int customerId = ovsdbSession.getCustomerId();
             Equipment equipmentConfig = equipmentServiceInterface.getByInventoryIdOrNull(apId);
             if (equipmentConfig == null) {
                 throw new IllegalStateException("Cannot retrieve configuration for " + apId);
             }
+            int customerId = equipmentConfig.getCustomerId();
 
             ret = new OpensyncAPConfig();
 
@@ -1062,11 +1055,6 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
         return ret;
     }
 
-    @Override
-    public void processMqttMessage(String topic, Report report) {
-        mqttMessageProcessor.processMqttMessage(topic, report);
-    }
-
     @Override
     public void wifiVIFStateDbTableUpdate(List<OpensyncAPVIFState> vifStateTables, String apId) {
         LOG.debug("Received Wifi_VIF_State table update for AP {}", apId);
@@ -1078,20 +1066,22 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
 
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("wifiVIFStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment apNode = equipmentServiceInterface.getOrNull(equipmentId);
         if (apNode == null) {
             LOG.debug("wifiVIFStateDbTableUpdate::Cannot get EquipmentId for AP {}", apId);
             return; // we don't have the required info to get the
             // radio type yet
         }
+
+        int customerId = apNode.getCustomerId();
+
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("wifiVIFStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
+
         ApElementConfiguration apElementConfig = (ApElementConfiguration) apNode.getDetails();
 
         ProfileContainer profileContainer = new ProfileContainer(profileServiceInterface.getProfileWithChildren(apNode.getProfileId()));
@@ -1215,20 +1205,19 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("wifiRadioStatusDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
         if (ce == null) {
             LOG.debug("wifiRadioStatusDbTableUpdate::Cannot get Equipment for AP {}", apId);
             return;
         }
 
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("wifiRadioStatusDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
+
         ApElementConfiguration apElementConfiguration = ((ApElementConfiguration) ce.getDetails());
 
         boolean configStateMismatch = false;
@@ -1441,18 +1430,16 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("wifiInetStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
+        if (ce == null) {
+            LOG.debug("wifiInetStateDbTableUpdate Cannot get customer Equipment for {}", apId);
             return;
         }
 
-        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
-
-        if (ce == null) {
-            LOG.debug("wifiInetStateDbTableUpdate Cannot get customer Equipment for {}", apId);
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("wifiInetStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
             return;
         }
 
@@ -1571,14 +1558,7 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("wifiAssociatedClientsDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
 
         if (ce == null) {
@@ -1586,6 +1566,12 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("wifiAssociatedClientsDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
+
         if ((wifiAssociatedClients == null) || wifiAssociatedClients.isEmpty()) {
             return;
         }
@@ -1624,8 +1610,8 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                 clientSession.setMacAddress(clientInstance.getMacAddress());
                 clientSession.setLocationId(ce.getLocationId());
                 clientSession.setDetails(new ClientSessionDetails());
-                clientSession.getDetails().setSessionId(clientInstance.getMacAddress().getAddressAsLong());
-                clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(clientInstance.getMacAddress().getAddressAsLong()));
+                clientSession.getDetails().setSessionId(Long.toUnsignedString(clientInstance.getMacAddress().getAddressAsLong()));
+                clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString(clientInstance.getMacAddress().getAddressAsLong())));
                 clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
                 clientSession.getDetails().setAssociationState(AssociationState._802_11_Associated);
                 clientSession.getDetails().setAssocTimestamp(timestamp);
@@ -1667,19 +1653,17 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.info("awlanNodeDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
         if (ce == null) {
             LOG.info("awlanNodeDbTableUpdate::Cannot find AP {}", apId);
             return;
         }
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.info("awlanNodeDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
 
         int upgradeStatusFromAp = node.getUpgradeStatus();
         EquipmentUpgradeState fwUpgradeState = null;
@@ -1894,32 +1878,26 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
 
         LOG.info("wifiVIFStateDbTableDelete for AP {} rows {}", apId, vifStateTables);
         OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
-
         if (ovsdbSession == null) {
             LOG.debug("wifiVIFStateDbTableDelete::Cannot get Session for AP {}", apId);
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
+        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
+        int customerId = ce.getCustomerId();
         if ((customerId < 0) || (equipmentId < 0)) {
             LOG.debug("wifiVIFStateDbTableDelete::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
             return;
         }
 
         Status activeBssidsStatus = statusServiceInterface.getOrNull(customerId, equipmentId, StatusDataType.ACTIVE_BSSIDS);
-
         if (activeBssidsStatus == null) {
             return; // nothing to delete
-
         }
-
         ActiveBSSIDs statusDetails = (ActiveBSSIDs) activeBssidsStatus.getDetails();
-
         List<ActiveBSSID> bssidList = statusDetails.getActiveBSSIDs();
         List<ActiveBSSID> toBeDeleted = new ArrayList<>();
-        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
         List<ClientSession> clientSessionsForCustomerAndEquipment = new ArrayList<>();
         if (ce != null) {
             PaginationResponse<ClientSession> clientSessions = clientServiceInterface.getSessionsForCustomer(customerId, ImmutableSet.of(equipmentId),
@@ -1962,15 +1940,13 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             LOG.debug("wifiAssociatedClientsDbTableDelete::Cannot get Session for AP {}", apId);
             return;
         }
-
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
+        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
+        int customerId = ce.getCustomerId();
         if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("wifiAssociatedClientsDbTableDelete::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            LOG.debug("wifiVIFStateDbTableDelete::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
             return;
         }
-
         Client client = clientServiceInterface.getOrNull(customerId, MacAddress.valueOf(deletedClientMac));
         ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId, MacAddress.valueOf(deletedClientMac));
 
@@ -2004,20 +1980,17 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("updateDhcpIpClientFingerprints::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
-
+        int customerId = ce.getCustomerId();
         if (ce == null) {
             LOG.debug("updateDhcpIpClientFingerprints::Cannot get Equipment for AP {}", apId);
             return;
         }
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("updateDhcpIpClientFingerprints::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
 
         long locationId = ce.getLocationId();
 
@@ -2311,20 +2284,17 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
             return;
         }
 
-        int customerId = ovsdbSession.getCustomerId();
         long equipmentId = ovsdbSession.getEquipmentId();
-
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.debug("clearEquipmentStatus::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
-
         Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
-
         if (ce == null) {
             LOG.debug("clearEquipmentStatus Cannot get customer Equipment for {}", apId);
             return;
         }
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.debug("clearEquipmentStatus::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
 
         List<Status> statusList = new ArrayList<>();
 
@@ -2391,22 +2361,26 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
     @Override
     public void apcStateDbTableUpdate(Map<String, String> apcStateAttributes, String apId, RowUpdateOperation rowUpdateOperation) {
         LOG.info("apcStateDbTableUpdate {} operations on AP {} with values {} ", rowUpdateOperation, apId, apcStateAttributes);
+
         OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
         if (ovsdbSession == null) {
             LOG.info("apcStateDbTableUpdate::Cannot get Session for AP {}", apId);
             return;
         }
-        int customerId = ovsdbSession.getCustomerId();
+
         long equipmentId = ovsdbSession.getEquipmentId();
-        if ((customerId < 0) || (equipmentId < 0)) {
-            LOG.info("apcStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
-            return;
-        }
         Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
         if (ce == null) {
             LOG.info("apcStateDbTableUpdate::Cannot get Equipment for AP {}", apId);
             return;
         }
+
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.info("apcStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
+
         InetAddress drIpAddr = null;
         InetAddress bdrIpAddr = null;
         try {
@@ -2441,4 +2415,59 @@ public class OpensyncExternalIntegrationCloud implements OpensyncExternalIntegra
                         RealTimeEventType.APC_Election_event, customerId, ce.getLocationId(), equipmentId, System.currentTimeMillis());
         mqttMessageProcessor.publishSystemEventFromTableStateMonitor(electionEvent);
     }
+
+    @Override
+    public void nodeStateDbTableUpdate(List<Map<String, String>> nodeStateAttributes, String apId) {
+        LOG.debug("start nodeStateDbTableUpdate for {}", apId);
+        if (LOG.isTraceEnabled())
+            LOG.trace("nodeStateDbTableUpdate tableAttributes {}", nodeStateAttributes);
+
+        OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
+        if (ovsdbSession == null) {
+            LOG.warn("nodeStateDbTableUpdate::Cannot get Session for AP {}", apId);
+            return;
+        }
+
+        long equipmentId = ovsdbSession.getEquipmentId();
+        Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
+        if (ce == null) {
+            LOG.warn("nodeStateDbTableUpdate::Cannot get Equipment for AP {}", apId);
+            return;
+        }
+
+        int customerId = ce.getCustomerId();
+        if ((customerId < 0) || (equipmentId < 0)) {
+            LOG.warn("nodeStateDbTableUpdate::Cannot get valid CustomerId {} or EquipmentId {} for AP {}", customerId, equipmentId, apId);
+            return;
+        }
+
+        Status eqAdminStatus = statusServiceInterface.getOrNull(customerId, equipmentId, StatusDataType.EQUIPMENT_ADMIN);
+
+        LedStatus ledStatus = null;
+        for (Map<String, String> nsa : nodeStateAttributes) {
+            if (nsa.get("module").equals("led")) {
+                if (nsa.get("key").equals("led_blink") && nsa.get("value").equals("on")) {
+                    ledStatus = LedStatus.led_blink;
+                } else if (nsa.get("key").equals("led_off") && nsa.get("value").equals("off")) {
+                    ledStatus = LedStatus.led_off;
+                } else {
+                    ledStatus = LedStatus.UNKNOWN;
+                }
+            }
+        }
+
+        if (ledStatus != null) {
+            if (eqAdminStatus != null) {
+                if (((EquipmentAdminStatusData) eqAdminStatus.getDetails()).getLedStatus() == null
+                        || !((EquipmentAdminStatusData) eqAdminStatus.getDetails()).getLedStatus().equals(ledStatus)) {
+                    ((EquipmentAdminStatusData) eqAdminStatus.getDetails()).setLedStatus(ledStatus);
+                    eqAdminStatus = statusServiceInterface.update(eqAdminStatus);
+                    LOG.debug("nodeStateDbTableUpdate updated status {}", eqAdminStatus);
+                }
+            }
+        }
+
+        LOG.debug("finished nodeStateDbTableUpdate for {}", apId);
+
+    }
 }

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

@@ -25,6 +25,7 @@ import org.springframework.security.web.bind.annotation.AuthenticationPrincipal;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 import com.telecominfraproject.wlan.core.client.PingClient;
@@ -50,6 +51,7 @@ import com.telecominfraproject.wlan.equipmentgateway.models.CEGWRebootRequest;
 import com.telecominfraproject.wlan.equipmentgateway.models.CEGWRouteCheck;
 import com.telecominfraproject.wlan.equipmentgateway.models.CEGWStartDebugEngine;
 import com.telecominfraproject.wlan.equipmentgateway.models.CEGWStopDebugEngine;
+import com.telecominfraproject.wlan.equipmentgateway.models.CEGatewayCommand;
 import com.telecominfraproject.wlan.equipmentgateway.models.EquipmentCommand;
 import com.telecominfraproject.wlan.equipmentgateway.models.EquipmentCommandResponse;
 import com.telecominfraproject.wlan.equipmentgateway.models.GatewayDefaults;
@@ -207,6 +209,9 @@ public class OpensyncCloudGatewayController {
                     case CellSizeAttributesRequest:
                         ret.add(sendCellSizeRequest(session, (CEGWCellSizeAttributesRequest) command));
                         break;
+                    case MostRecentStatsTimestamp:
+                        ret.add(sendGetMostRecentStatsTimestampRequest(command, inventoryId));
+                        break;
                     default:
                         LOG.warn("[{}] Failed to deliver command {}, unsupported command type", inventoryId, command);
                         ret.add(new EquipmentCommandResponse(CEGWCommandResultCode.UnsupportedCommand,
@@ -217,6 +222,8 @@ public class OpensyncCloudGatewayController {
 
             }
 
+
+
         });
 
         return ret;
@@ -242,6 +249,33 @@ public class OpensyncCloudGatewayController {
     public GatewayDefaults retrieveGatewayDefaults() {
         return new GatewayDefaults();
     }
+    
+    private EquipmentCommandResponse sendGetMostRecentStatsTimestampRequest(CEGWBaseCommand command, String inventoryId) {
+        Long ts = lastReceivedStatsTimestamp(inventoryId);
+        if (ts == null) {
+            return new EquipmentCommandResponse(CEGWCommandResultCode.NoRouteToCE,
+                    null, command,
+                    registeredGateway == null ? null : registeredGateway.getHostname(),
+                            registeredGateway == null ? -1 : registeredGateway.getPort());
+        } else {
+            return new EquipmentCommandResponse(CEGWCommandResultCode.Success,
+                    ts.toString(), command,
+                    registeredGateway == null ? null : registeredGateway.getHostname(),
+                            registeredGateway == null ? -1 : registeredGateway.getPort());
+        }
+    }
+    
+    @RequestMapping(value = "/lastReceivedStatsTimestamp", method = RequestMethod.GET)
+    public Long lastReceivedStatsTimestamp(@RequestParam String apId) {
+        Long ret = null;
+        if (ovsdbSessionMapInterface.getSession(apId) != null) {
+            ret = ovsdbSessionMapInterface.getSession(apId).getMostRecentStatsTimestamp();
+            LOG.debug("lastReceivedStatsTimestamp for apId {} {}",apId,ret);
+        } else {
+            LOG.warn("lastReceivedStatsTimestamp found no session for {}, cannot get timestamp",apId);
+        }
+        return ret;
+    }
 
     /**
      * Verify a route to customer equipment
@@ -254,7 +288,7 @@ public class OpensyncCloudGatewayController {
         if (null != command.getRoutingId()) {
             if (!command.getRoutingId().equals(session.getRoutingId())) {
 
-                LOG.info("[C:{} E:{} R:{}] Stale routing entry ({}) detected", session.getCustomerId(), command.getInventoryId(), session.getRoutingId(),
+                LOG.info("[E:{} R:{}] Stale routing entry ({}) detected", command.getInventoryId(), session.getRoutingId(),
                         command.getRoutingId());
 
                 return new EquipmentCommandResponse(CEGWCommandResultCode.NoRouteToCE, "Inactive Route Identifer", command,

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

@@ -5,6 +5,7 @@ import java.net.Inet4Address;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.text.DecimalFormat;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -13,18 +14,17 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
+import com.google.common.base.Objects;
 import com.google.protobuf.ByteString;
-import com.google.protobuf.Descriptors;
-import com.google.protobuf.InvalidProtocolBufferException;
-import com.google.protobuf.util.JsonFormat;
-import com.google.protobuf.util.JsonFormat.TypeRegistry;
 import com.telecominfraproject.wlan.alarm.AlarmServiceInterface;
 import com.telecominfraproject.wlan.alarm.models.Alarm;
 import com.telecominfraproject.wlan.alarm.models.AlarmCode;
@@ -52,6 +52,7 @@ import com.telecominfraproject.wlan.equipment.EquipmentServiceInterface;
 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.external.integration.controller.OpensyncCloudGatewayController;
 import com.telecominfraproject.wlan.opensync.util.OvsdbToWlanCloudTypeMappingUtility;
 import com.telecominfraproject.wlan.profile.ProfileServiceInterface;
 import com.telecominfraproject.wlan.profile.models.Profile;
@@ -76,7 +77,6 @@ import com.telecominfraproject.wlan.servicemetric.models.ServiceMetric;
 import com.telecominfraproject.wlan.servicemetric.neighbourscan.models.NeighbourReport;
 import com.telecominfraproject.wlan.servicemetric.neighbourscan.models.NeighbourScanReports;
 import com.telecominfraproject.wlan.status.StatusServiceInterface;
-import com.telecominfraproject.wlan.status.equipment.models.EquipmentAdminStatusData;
 import com.telecominfraproject.wlan.status.equipment.report.models.ActiveBSSID;
 import com.telecominfraproject.wlan.status.equipment.report.models.ActiveBSSIDs;
 import com.telecominfraproject.wlan.status.equipment.report.models.EquipmentCapacityDetails;
@@ -117,14 +117,10 @@ import sts.OpensyncStats.Survey;
 import sts.OpensyncStats.Survey.SurveySample;
 import sts.OpensyncStats.SurveyType;
 import sts.OpensyncStats.VLANMetrics;
-import traffic.NetworkMetadata;
-import traffic.NetworkMetadata.FlowReport;
-import wc.stats.IpDnsTelemetry;
-import wc.stats.IpDnsTelemetry.WCStatsReport;
 
 @org.springframework.context.annotation.Profile("opensync_cloud_config")
 @Component
-public class MqttStatsPublisher {
+public class MqttStatsPublisher implements StatsPublisherInterface {
 
     private static final Logger LOG = LoggerFactory.getLogger(MqttStatsPublisher.class);
 
@@ -144,7 +140,9 @@ public class MqttStatsPublisher {
     private RealtimeEventPublisher realtimeEventPublisher;
     @Autowired
     private AlarmServiceInterface alarmServiceInterface;
-
+    @Autowired
+    private OpensyncCloudGatewayController gatewayController;
+    
     @Value("${tip.wlan.mqttStatsPublisher.temperatureThresholdInC:80}")
     private int temperatureThresholdInC;
 
@@ -154,38 +152,67 @@ public class MqttStatsPublisher {
     @Value("${tip.wlan.mqttStatsPublisher.memoryUtilThresholdPct:70}")
     private int memoryUtilThresholdPct;
 
+    @Value("${tip.wlan.mqttStatsPublisher.reportProcessingThresholdSec:30}")
+    public int reportProcessingThresholdSec;
+
+    @Value("${tip.wlan.mqttStatsPublisher.statsTimeDriftThresholdSec:300}")
+    public int statsTimeDriftThresholdSec;
+
+    @Override
+    @Async
     public void processMqttMessage(String topic, Report report) {
-        LOG.info("Received report on topic {} for ap {}", topic, report.getNodeID());
-        int customerId = extractCustomerIdFromTopic(topic);
+        long startTime = System.nanoTime();
         String apId = extractApIdFromTopic(topic);
-        long equipmentId = extractEquipmentIdFromTopic(topic);
-        if ((equipmentId <= 0) || (customerId <= 0)) {
-            LOG.warn("Cannot determine equipment ids from topic {} - customerId {} equipmentId {}", topic, customerId, equipmentId);
-            return;
-        }
-        Equipment ce = equipmentServiceInterface.getOrNull(equipmentId);
+        LOG.info("Received report on topic {} for ap {}", topic, report.getNodeID());
+        Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
         if (ce == null) {
-            LOG.warn("Cannot read equipment {}", apId);
+            LOG.warn("Cannot get equipment for inventoryId {}. Ignore mqtt message for topic {}", apId, topic);
             return;
         }
 
+        int customerId = ce.getCustomerId();
+        long equipmentId = ce.getId();
         long locationId = ce.getLocationId();
         long profileId = ce.getProfileId();
-
+        
+        // update timestamp for active customer equipment
+        gatewayController.updateActiveCustomer(customerId);
         List<ServiceMetric> metricRecordList = new ArrayList<>();
 
         try {
+            long clientMetricsStart = System.nanoTime();
             populateApClientMetrics(metricRecordList, report, customerId, equipmentId, locationId);
+            long clientMetricsStop = System.nanoTime();
+            LOG.debug("Elapsed time for constructing Client metrics record was {} milliseconds for topic {}",
+                    TimeUnit.MILLISECONDS.convert(clientMetricsStop - clientMetricsStart, TimeUnit.NANOSECONDS), topic);
+
+            long nodeMetricsStart = System.nanoTime();
             populateApNodeMetrics(metricRecordList, report, customerId, equipmentId, locationId);
+            long nodeMetricsStop = System.nanoTime();
+            LOG.debug("Elapsed time for constructing ApNode metrics record was {} milliseconds for topic {}",
+                    TimeUnit.MILLISECONDS.convert(nodeMetricsStop - nodeMetricsStart, TimeUnit.NANOSECONDS), topic);
+
+            long neighbourScanStart = System.nanoTime();
             populateNeighbourScanReports(metricRecordList, report, customerId, equipmentId, locationId);
+            long neighbourScanStop = System.nanoTime();
+            LOG.debug("Elapsed time for constructing Neighbour metrics record was {} milliseconds for topic {}",
+                    TimeUnit.MILLISECONDS.convert(neighbourScanStop - neighbourScanStart, TimeUnit.NANOSECONDS), topic);
+
+            long channelInfoStart = System.nanoTime();
             populateChannelInfoReports(metricRecordList, report, customerId, equipmentId, locationId, profileId);
+            long channelInfoStop = System.nanoTime();
+            LOG.debug("Elapsed time for constructing Channel metrics record was {} milliseconds for topic {}",
+                    TimeUnit.MILLISECONDS.convert(channelInfoStop - channelInfoStart, TimeUnit.NANOSECONDS), topic);
+
+            long ssidStart = System.nanoTime();
             populateApSsidMetrics(metricRecordList, report, customerId, equipmentId, apId, locationId);
+            long ssidStop = System.nanoTime();
+            LOG.debug("Elapsed time for constructing ApSsid metrics record was {} milliseconds for topic {}",
+                    TimeUnit.MILLISECONDS.convert(ssidStop - ssidStart, TimeUnit.NANOSECONDS), topic);
 
             if (!metricRecordList.isEmpty()) {
-
                 long serviceMetricTimestamp = System.currentTimeMillis();
-                LOG.debug("Current timestamp for service metrics is {}", serviceMetricTimestamp);
-                metricRecordList.stream().forEach(smr -> {
+                metricRecordList.stream().forEach(smr -> {              
                     smr.setCreatedTimestamp(serviceMetricTimestamp);
                     if (smr.getLocationId() == 0)
                         smr.setLocationId(locationId);
@@ -193,89 +220,99 @@ public class MqttStatsPublisher {
                         smr.setCustomerId(customerId);
                     if (smr.getEquipmentId() == 0L)
                         smr.setEquipmentId(equipmentId);
+                    
+                    long sourceTimestamp = smr.getDetails().getSourceTimestampMs();
+                    long diffMillis = serviceMetricTimestamp - sourceTimestamp;
+                    long thresholdMillis = statsTimeDriftThresholdSec * 1000L;                    
+                    if (diffMillis > thresholdMillis) {
+                        double diffSec = diffMillis / 1000D;
+                        LOG.warn("AP {} stats report is {} seconds behind cloud. ServiceMetric {} sourceTimestampMs {} createdTimestampMs {}.", apId,diffSec,  smr.getDataType(), sourceTimestamp, serviceMetricTimestamp);
+                    } 
+                    
                 });
-                metricRecordList.stream().forEach(smr -> {
-                    LOG.debug("ServiceMetric {}", smr);
-                });
+                long publishStart = System.nanoTime();
                 cloudEventDispatcherInterface.publishMetrics(metricRecordList);
+                long publishStop = System.nanoTime();
+                LOG.debug("Elapsed publishing time for metrics records from AP {} is {} milliseconds", apId,
+                        TimeUnit.MILLISECONDS.convert(publishStop - publishStart, TimeUnit.NANOSECONDS));
             }
+
+            long mqttEventsStart = System.nanoTime();
             publishEvents(report, customerId, equipmentId, apId, locationId);
+            long mqttEventsStop = System.nanoTime();
+            LOG.debug("Elapsed publishing time for mqtt events from AP {} is {} milliseconds", apId,
+                    TimeUnit.MILLISECONDS.convert(mqttEventsStop - mqttEventsStart, TimeUnit.NANOSECONDS));
+
+            long endTime = System.nanoTime();
+            long elapsedTimeMillis = TimeUnit.MILLISECONDS.convert(endTime - startTime, TimeUnit.NANOSECONDS);
+            long elapsedTimeSeconds = TimeUnit.SECONDS.convert(endTime - startTime, TimeUnit.NANOSECONDS);
+
+            if (elapsedTimeSeconds > reportProcessingThresholdSec) {
+                LOG.warn("Processing threshold exceeded for stats messages from AP {}. Elapsed processing time {} seconds. Report: {}", apId,
+                        elapsedTimeSeconds, report);
+            } else {
+                if (elapsedTimeSeconds < 1) {
+                    LOG.debug("Total elapsed processing time {} milliseconds for stats messages from AP {}", elapsedTimeMillis, apId);
+                } else {
+                    LOG.debug("Total elapsed processing time {} seconds for stats messages from AP {}", elapsedTimeSeconds, apId);
+                }
+            }
         } catch (Exception e) {
             LOG.error("Exception when processing stats messages from AP", e);
         }
 
     }
 
+    @Override
     public void publishSystemEventFromTableStateMonitor(SystemEvent event) {
         LOG.info("Publishing SystemEvent received by TableStateMonitor {}", event);
+        gatewayController.updateActiveCustomer(event.getCustomerId());
         cloudEventDispatcherInterface.publishEvent(event);
     }
 
     void publishEvents(Report report, int customerId, long equipmentId, String apId, long locationId) {
 
+        // asynchronous
         realtimeEventPublisher.publishSipCallEvents(customerId, equipmentId, locationId, report.getVideoVoiceReportList());
 
         for (EventReport eventReport : report.getEventReportList()) {
 
             for (sts.OpensyncStats.EventReport.ClientSession apEventClientSession : eventReport.getClientSessionList()) {
 
-                LOG.debug("Processing EventReport::ClientSession {}", apEventClientSession);
-
+                LOG.debug("Processing EventReport::ClientSession for AP {}", apId);
+                // for the following MQTT events, the client/client session is first updated, then the real time event
+                // is published, asynchronously.
                 if (apEventClientSession.hasClientAuthEvent()) {
                     processClientAuthEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientAuthSystemEvent(customerId, equipmentId, locationId, apEventClientSession.getClientAuthEvent());
                 }
-
                 if (apEventClientSession.hasClientAssocEvent()) {
                     processClientAssocEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientAssocEvent(customerId, equipmentId, locationId, apEventClientSession.getClientAssocEvent());
                 }
-
                 if (apEventClientSession.hasClientFirstDataEvent()) {
                     processClientFirstDataEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientFirstDataEvent(customerId, equipmentId, locationId, apEventClientSession.getClientFirstDataEvent());
-
                 }
-
                 if (apEventClientSession.hasClientIdEvent()) {
                     processClientIdEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientIdEvent(customerId, equipmentId, locationId, apEventClientSession.getClientIdEvent());
-
                 }
-
                 if (apEventClientSession.hasClientIpEvent()) {
                     processClientIpEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientIpEvent(customerId, equipmentId, locationId, apEventClientSession.getClientIpEvent());
-
                 }
-
                 if (apEventClientSession.hasClientConnectEvent()) {
                     processClientConnectEvent(customerId, equipmentId, locationId, eventReport, apEventClientSession);
-                    realtimeEventPublisher.publishClientConnectSuccessEvent(customerId, equipmentId, locationId, apEventClientSession.getClientConnectEvent());
-
                 }
-
                 if (apEventClientSession.hasClientDisconnectEvent()) {
                     processClientDisconnectEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientDisconnectEvent(customerId, equipmentId, locationId, apEventClientSession.getClientDisconnectEvent());
                 }
-
                 if (apEventClientSession.hasClientTimeoutEvent()) {
                     processClientTimeoutEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientTimeoutEvent(customerId, equipmentId, locationId, apEventClientSession.getClientTimeoutEvent());
-
                 }
-
                 if (apEventClientSession.hasClientFailureEvent()) {
                     processClientFailureEvent(customerId, equipmentId, locationId, apEventClientSession);
-                    realtimeEventPublisher.publishClientFailureEvent(customerId, equipmentId, locationId, apEventClientSession.getClientFailureEvent());
-
                 }
-
             }
-
+            // asynchronous
             realtimeEventPublisher.publishChannelHopEvents(customerId, equipmentId, locationId, eventReport);
-
+            // asynchronous
             realtimeEventPublisher.publishDhcpTransactionEvents(customerId, equipmentId, locationId, eventReport.getDhcpTransactionList());
 
         }
@@ -302,20 +339,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         clientSession.getDetails().setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(apClientEvent.getBand()));
         clientSession.getDetails().setSsid(apClientEvent.getSsid());
 
@@ -383,12 +420,9 @@ public class MqttStatsPublisher {
             if (ipAddress != null) {
                 try {
                     InetAddress inetAddress = InetAddress.getByAddress(ipAddress.toByteArray());
-                    LOG.info("ipAddr {} ", inetAddress);
                     if (inetAddress instanceof Inet4Address) {
-                        LOG.info("IPv4 address {}", inetAddress);
                         clientSession.getDetails().setIpAddress(inetAddress);
                     } else if (inetAddress instanceof Inet6Address) {
-                        LOG.info("IPv6 address {}", inetAddress);
                         clientSession.getDetails().setIpAddress(inetAddress);
                     } else {
                         LOG.error("Invalid IP Address {}", ipAddress);
@@ -406,6 +440,7 @@ public class MqttStatsPublisher {
         }
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientConnectSuccessEvent(customerId, equipmentId, locationId, apEventClientSession.getClientConnectEvent());
 
     }
 
@@ -428,20 +463,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         clientSession.getDetails().setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(apClientEvent.getBand()));
         clientSession.getDetails().setSsid(apClientEvent.getSsid());
         if (apClientEvent.hasDevType()) {
@@ -482,6 +517,7 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setAssocTimestamp(apClientEvent.getTimestampMs());
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientDisconnectEvent(customerId, equipmentId, locationId, apEventClientSession.getClientDisconnectEvent());
 
     }
 
@@ -506,20 +542,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         clientSession.getDetails().setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(apClientEvent.getBand()));
         clientSession.getDetails().setSsid(apClientEvent.getSsid());
         if (apClientEvent.hasAuthStatus()) {
@@ -529,6 +565,9 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setAssociationState(AssociationState._802_11_Authenticated);
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+
+        realtimeEventPublisher.publishClientAuthSystemEvent(customerId, equipmentId, locationId, apEventClientSession.getClientAuthEvent());
+
     }
 
     protected void processClientAssocEvent(int customerId, long equipmentId, long locationId,
@@ -554,20 +593,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         clientSession.getDetails().setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(apClientEvent.getBand()));
         clientSession.getDetails().setSsid(apClientEvent.getSsid());
         if (apClientEvent.hasStatus()) {
@@ -596,6 +635,8 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setAssociationState(AssociationState._802_11_Associated);
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientAssocEvent(customerId, equipmentId, locationId, apEventClientSession.getClientAssocEvent());
+
     }
 
     protected void processClientFailureEvent(int customerId, long equipmentId, long locationId,
@@ -617,20 +658,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         clientSession.getDetails().setSsid(apClientEvent.getSsid());
 
         ClientFailureDetails clientFailureDetails = new ClientFailureDetails();
@@ -643,7 +684,7 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setLastFailureDetails(clientFailureDetails);
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
-
+        realtimeEventPublisher.publishClientFailureEvent(customerId, equipmentId, locationId, apEventClientSession.getClientFailureEvent());
     }
 
     protected void processClientFirstDataEvent(int customerId, long equipmentId, long locationId,
@@ -665,20 +706,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
 
         if (apClientEvent.hasFdataRxUpTsInUs()) {
             clientSession.getDetails().setFirstDataRcvdTimestamp(apClientEvent.getFdataRxUpTsInUs());
@@ -690,6 +731,8 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
 
+        realtimeEventPublisher.publishClientFirstDataEvent(customerId, equipmentId, locationId, apEventClientSession.getClientFirstDataEvent());
+
     }
 
     protected void processClientIdEvent(int customerId, long equipmentId, long locationId, sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
@@ -715,25 +758,26 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getCltMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         if (apClientEvent.hasCltId()) {
             clientSession.getDetails().setHostname(apClientEvent.getCltId());
         }
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientIdEvent(customerId, equipmentId, locationId, apEventClientSession.getClientIdEvent());
     }
 
     protected void processClientIpEvent(int customerId, long equipmentId, long locationId, sts.OpensyncStats.EventReport.ClientSession apEventClientSession) {
@@ -756,30 +800,28 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         if (apClientEvent.hasIpAddr()) {
             ByteString ipAddress = apClientEvent.getIpAddr();
             if (ipAddress != null) {
                 try {
                     InetAddress inetAddress = InetAddress.getByAddress(ipAddress.toByteArray());
                     if (inetAddress instanceof Inet4Address) {
-                        LOG.info("IPv4 address {}", inetAddress);
                         clientSession.getDetails().setIpAddress(inetAddress);
                     } else if (inetAddress instanceof Inet6Address) {
-                        LOG.info("IPv6 address {}", inetAddress);
                         clientSession.getDetails().setIpAddress(inetAddress);
                     } else {
                         LOG.error("Invalid IP Address {}", ipAddress);
@@ -792,6 +834,7 @@ public class MqttStatsPublisher {
         }
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientIpEvent(customerId, equipmentId, locationId, apEventClientSession.getClientIpEvent());
     }
 
     protected void processClientTimeoutEvent(int customerId, long equipmentId, long locationId,
@@ -814,20 +857,20 @@ public class MqttStatsPublisher {
             clientSession.setMacAddress(MacAddress.valueOf(apClientEvent.getStaMac()));
             clientSession.setLocationId(locationId);
             clientSession.setDetails(new ClientSessionDetails());
-            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(apEventClientSession.getSessionId()));
+            clientSession.getDetails().setDhcpDetails(new ClientDhcpDetails(Long.toUnsignedString( apEventClientSession.getSessionId())));
             clientSession.getDetails().setMetricDetails(new ClientSessionMetricDetails());
         }
         if (clientSession.getDetails().getPriorEquipmentId() == null) {
             clientSession.getDetails().setPriorEquipmentId(clientSession.getEquipmentId());
         }
         if (clientSession.getDetails().getPriorSessionId() == null) {
-            if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId())
+            if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId())))
                 clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        if (clientSession.getDetails().getSessionId() != apEventClientSession.getSessionId()) {
+        if (! Objects.equal( clientSession.getDetails().getSessionId(), Long.toUnsignedString( apEventClientSession.getSessionId()))) {
             clientSession.getDetails().setPriorSessionId(clientSession.getDetails().getSessionId());
         }
-        clientSession.getDetails().setSessionId(apEventClientSession.getSessionId());
+        clientSession.getDetails().setSessionId(Long.toUnsignedString( apEventClientSession.getSessionId()));
         if (apClientEvent.hasLastRcvUpTsInUs()) {
             clientSession.getDetails().setLastRxTimestamp(apClientEvent.getLastRcvUpTsInUs());
         }
@@ -838,6 +881,7 @@ public class MqttStatsPublisher {
         clientSession.getDetails().setAssociationState(AssociationState.AP_Timeout);
         clientSession.getDetails().setLastEventTimestamp(apClientEvent.getTimestampMs());
         clientSession = clientServiceInterface.updateSession(clientSession);
+        realtimeEventPublisher.publishClientTimeoutEvent(customerId, equipmentId, locationId, apEventClientSession.getClientTimeoutEvent());
     }
 
     void populateApNodeMetrics(List<ServiceMetric> metricRecordList, Report report, int customerId, long equipmentId, long locationId) {
@@ -846,7 +890,6 @@ public class MqttStatsPublisher {
         ServiceMetric smr = new ServiceMetric(customerId, equipmentId);
 
         smr.setLocationId(locationId);
-
         metricRecordList.add(smr);
         smr.setDetails(apNodeMetrics);
 
@@ -896,7 +939,7 @@ public class MqttStatsPublisher {
 
                 double usedMemory = deviceReport.getMemUtil().getMemUsed();
                 double totalMemory = deviceReport.getMemUtil().getMemTotal();
-                if (usedMemory/totalMemory * 100 > memoryUtilThresholdPct) {
+                if (usedMemory / totalMemory * 100 > memoryUtilThresholdPct) {
                     raiseDeviceThresholdAlarm(customerId, equipmentId, AlarmCode.MemoryUtilization, deviceReport.getTimestampMs());
                 } else {
                     // Clear any existing cpuUtilization alarms
@@ -1090,12 +1133,20 @@ public class MqttStatsPublisher {
                     checkIfOutOfBound("pctBusyTx", pctBusyTx, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
 
                     radioUtil.setAssocClientTx(pctBusyTx);
-                    int pctBusyRx = busyRx / totalDurationMs;
+                    int pctBusyRx = busySelf / totalDurationMs;
                     checkIfOutOfBound("pctBusyRx", pctBusyRx, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
                     radioUtil.setAssocClientRx(pctBusyRx);
 
-                    double pctIBSS = (busyTx + busySelf) / totalDurationMs;
+                    double pctIBSS = (busyTx + busySelf) / totalDurationMs;        
+                    if (pctIBSS > 100D || pctIBSS < 0D) {
+                        LOG.warn(
+                                "Calculated value for {} {} is out of bounds on totalDurationMs {} for survey.getBand {}. busyTx {} busyRx {} busy {} busySelf {} "
+                                        + " survey.getTimestampMs {}, survey.getSurveyListList {}",
+                                "pctIBSS", pctIBSS, totalDurationMs, survey.getBand(), busyTx, busyRx, busy, busySelf, survey.getTimestampMs(),
+                                survey.getSurveyListList());
+                    }                    
                     radioUtil.setIbss(pctIBSS);
+                    
                     int nonWifi = (busy - (busyTx + busyRx)) / totalDurationMs;
                     checkIfOutOfBound("nonWifi", nonWifi, survey, totalDurationMs, busyTx, busyRx, busy, busySelf);
                     radioUtil.setNonWifi(nonWifi);
@@ -1146,13 +1197,15 @@ public class MqttStatsPublisher {
         updateDeviceStatusRadioUtilizationReport(customerId, equipmentId, radioUtilizationReport);
     }
 
+    @Async
     void clearDeviceThresholdAlarm(int customerId, long equipmentId, AlarmCode alarmCode) {
         alarmServiceInterface.get(customerId, Set.of(equipmentId), Set.of(alarmCode)).stream().forEach(a -> {
             Alarm alarm = alarmServiceInterface.delete(customerId, equipmentId, a.getAlarmCode(), a.getCreatedTimestamp());
-            LOG.info("Cleared device threshold alarm {}", alarm);
+            LOG.debug("Cleared device threshold alarm {}", alarm);
         });
     }
 
+    @Async
     void raiseDeviceThresholdAlarm(int customerId, long equipmentId, AlarmCode alarmCode, long timestampMs) {
         // Raise an alarm for temperature
         Alarm alarm = new Alarm();
@@ -1172,9 +1225,9 @@ public class MqttStatsPublisher {
         if (alarms.isEmpty()) {
             alarm.setCreatedTimestamp(timestampMs);
             alarm = alarmServiceInterface.create(alarm);
-        } 
+        }
     }
-    
+
     private void checkIfOutOfBound(String checkedType, int checkedValue, Survey survey, int totalDurationMs, int busyTx, int busyRx, int busy, int busySelf) {
         if (checkedValue > 100 || checkedValue < 0) {
             LOG.warn(
@@ -1188,7 +1241,7 @@ public class MqttStatsPublisher {
     private void updateNetworkAdminStatusReport(int customerId, long equipmentId, ApNodeMetrics apNodeMetrics) {
         apNodeMetrics.getNetworkProbeMetrics().forEach(n -> {
 
-            LOG.info("Update NetworkAdminStatusReport for NetworkProbeMetrics {}", n.toString());
+            LOG.debug("Update NetworkAdminStatusReport for NetworkProbeMetrics {}", n.toString());
 
             Status networkAdminStatus = statusServiceInterface.getOrNull(customerId, equipmentId, StatusDataType.NETWORK_ADMIN);
 
@@ -1204,7 +1257,7 @@ public class MqttStatsPublisher {
             NetworkAdminStatusData statusData = (NetworkAdminStatusData) networkAdminStatus.getDetails();
 
             if (n.getDnsState() == null) {
-                LOG.debug("No DnsState present in networkProbeMetrics, DnsState and CloudLinkStatus set to 'normal");
+                LOG.trace("No DnsState present in networkProbeMetrics, DnsState and CloudLinkStatus set to 'normal");
                 statusData.setDnsStatus(StatusCode.normal);
                 statusData.setCloudLinkStatus(StatusCode.normal);
             } else {
@@ -1212,13 +1265,13 @@ public class MqttStatsPublisher {
                 statusData.setCloudLinkStatus(stateUpDownErrorToStatusCode(n.getDnsState()));
             }
             if (n.getDhcpState() == null) {
-                LOG.debug("No DhcpState present in networkProbeMetrics, set to 'normal");
+                LOG.trace("No DhcpState present in networkProbeMetrics, set to 'normal");
                 statusData.setDhcpStatus(StatusCode.normal);
             } else {
                 statusData.setDhcpStatus(stateUpDownErrorToStatusCode(n.getDhcpState()));
             }
             if (n.getRadiusState() == null) {
-                LOG.debug("No RadiusState present in networkProbeMetrics, set to 'normal");
+                LOG.trace("No RadiusState present in networkProbeMetrics, set to 'normal");
                 statusData.setRadiusStatus(StatusCode.normal);
             } else {
                 statusData.setRadiusStatus(stateUpDownErrorToStatusCode(n.getRadiusState()));
@@ -1253,9 +1306,7 @@ public class MqttStatsPublisher {
 
     void updateDeviceStatusRadioUtilizationReport(int customerId, long equipmentId, RadioUtilizationReport radioUtilizationReport) {
         LOG.info("Processing updateDeviceStatusRadioUtilizationReport for equipmentId {} with RadioUtilizationReport {}", equipmentId, radioUtilizationReport);
-
         Status radioUtilizationStatus = statusServiceInterface.getOrNull(customerId, equipmentId, StatusDataType.RADIO_UTILIZATION);
-
         if (radioUtilizationStatus == null) {
             LOG.debug("Create new radioUtilizationStatus");
             radioUtilizationStatus = new Status();
@@ -1263,9 +1314,7 @@ public class MqttStatsPublisher {
             radioUtilizationStatus.setEquipmentId(equipmentId);
             radioUtilizationStatus.setStatusDataType(StatusDataType.RADIO_UTILIZATION);
         }
-
         radioUtilizationStatus.setDetails(radioUtilizationReport);
-
         statusServiceInterface.update(radioUtilizationStatus);
     }
 
@@ -1273,32 +1322,23 @@ public class MqttStatsPublisher {
         List<NetworkProbeMetrics> networkProbeMetricsList = new ArrayList<>();
 
         for (NetworkProbe networkProbe : report.getNetworkProbeList()) {
-
             NetworkProbeMetrics networkProbeMetrics = new NetworkProbeMetrics();
-
             networkProbeMetrics.setSourceTimestampMs(networkProbe.getTimestampMs());
-
             List<DnsProbeMetric> dnsProbeResults = new ArrayList<>();
             if (networkProbe.hasDnsProbe()) {
-
                 DNSProbeMetric dnsProbeMetricFromAp = networkProbe.getDnsProbe();
-
                 DnsProbeMetric cloudDnsProbeMetric = new DnsProbeMetric();
-
                 if (dnsProbeMetricFromAp.hasLatency()) {
                     networkProbeMetrics.setDnsLatencyMs(dnsProbeMetricFromAp.getLatency());
                     cloudDnsProbeMetric.setDnsLatencyMs(dnsProbeMetricFromAp.getLatency());
                 }
-
                 if (dnsProbeMetricFromAp.hasState()) {
                     StateUpDownError dnsState =
                             OvsdbToWlanCloudTypeMappingUtility.getCloudMetricsStateFromOpensyncStatsStateUpDown(dnsProbeMetricFromAp.getState());
 
                     networkProbeMetrics.setDnsState(dnsState);
                     cloudDnsProbeMetric.setDnsState(dnsState);
-
                 }
-
                 if (dnsProbeMetricFromAp.hasServerIP()) {
                     InetAddress ipAddress;
                     try {
@@ -1308,9 +1348,7 @@ public class MqttStatsPublisher {
                         LOG.error("Could not get DNS Server IP from network_probe service_metrics_collection_config", e);
                     }
                 }
-
                 dnsProbeResults.add(cloudDnsProbeMetric);
-
             }
 
             networkProbeMetrics.setDnsProbeResults(dnsProbeResults);
@@ -1319,15 +1357,11 @@ public class MqttStatsPublisher {
                 if (radiusMetrics.hasLatency()) {
                     networkProbeMetrics.setRadiusLatencyInMs(radiusMetrics.getLatency());
                 }
-
                 if (radiusMetrics.hasRadiusState()) {
                     StateUpDownError radiusState =
                             OvsdbToWlanCloudTypeMappingUtility.getCloudMetricsStateFromOpensyncStatsStateUpDown(radiusMetrics.getRadiusState());
-
                     networkProbeMetrics.setRadiusState(radiusState);
-
                 }
-
             }
 
             if (networkProbe.hasVlanProbe()) {
@@ -1346,9 +1380,7 @@ public class MqttStatsPublisher {
                     networkProbeMetrics.setDhcpLatencyMs(vlanMetrics.getLatency());
                 }
             }
-
             networkProbeMetricsList.add(networkProbeMetrics);
-
         }
 
         apNodeMetrics.setNetworkProbeMetrics(networkProbeMetricsList);
@@ -1366,11 +1398,9 @@ public class MqttStatsPublisher {
         eqOsPerformance.setTotalAvailableMemoryKb(deviceReport.getMemUtil().getMemTotal());
         status.setDetails(eqOsPerformance);
         status = statusServiceInterface.update(status);
-        LOG.debug("updated status {}", status);
+        LOG.trace("updated status {}", status);
     }
 
-
-
     void populateApClientMetrics(List<ServiceMetric> metricRecordList, Report report, int customerId, long equipmentId, long locationId) {
         LOG.info("populateApClientMetrics for Customer {} Equipment {}", customerId, equipmentId);
 
@@ -1378,35 +1408,35 @@ public class MqttStatsPublisher {
             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.", cl);
+                    LOG.info("No mac address for Client {}, cannot set device mac address for client in ClientMetrics.", cl);
                     continue;
                 }
 
-                LOG.debug("Processing ClientReport from AP {}", cl.getMacAddress());
-
-                ServiceMetric smr = new ServiceMetric(customerId, equipmentId, MacAddress.valueOf(cl.getMacAddress()));
-                smr.setLocationId(locationId);
-                metricRecordList.add(smr);
-
-                smr.setClientMac(MacAddress.valueOf(cl.getMacAddress()).getAddressAsLong());
-
-                // clReport.getChannel();
-                ClientMetrics cMetrics = new ClientMetrics();
-                smr.setDetails(cMetrics);
-                cMetrics.setSourceTimestampMs(clReport.getTimestampMs());
-
-                Integer periodLengthSec = 60; // matches what's configured by
-                // OvsdbDao.configureStats(OvsdbClient)
-                cMetrics.setPeriodLengthSec(periodLengthSec);
-
-                cMetrics.setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(clReport.getBand()));
-
                 ClientSession session = clientServiceInterface.getSessionOrNull(customerId, equipmentId, MacAddress.valueOf(cl.getMacAddress()));
-                if (session != null) {
+
+                if (session == null) {
+                    LOG.info("No session for Client {}, ignore.");
+                    continue;
+                }
+
+                if (cl.hasStats()) {
+                    LOG.debug("Processing ClientReport from AP for client device {}", cl.getMacAddress());
+                    ServiceMetric smr = new ServiceMetric(customerId, equipmentId, MacAddress.valueOf(cl.getMacAddress()));
+                    smr.setLocationId(locationId);
+                    metricRecordList.add(smr);
+                    smr.setClientMac(MacAddress.valueOf(cl.getMacAddress()).getAddressAsLong());
+
+                    // clReport.getChannel();
+                    ClientMetrics cMetrics = new ClientMetrics();
+                    smr.setDetails(cMetrics);
+                    cMetrics.setSourceTimestampMs(clReport.getTimestampMs());
+                    Integer periodLengthSec = 60; // matches what's configured by
+                    // OvsdbDao.configureStats(OvsdbClient)
+                    cMetrics.setPeriodLengthSec(periodLengthSec);
+                    cMetrics.setRadioType(OvsdbToWlanCloudTypeMappingUtility.getRadioTypeFromOpensyncStatsRadioBandType(clReport.getBand()));
                     cMetrics.setSessionId(session.getDetails().getSessionId());
                     LOG.debug("populateApClientMetrics Session Id {}", session.getDetails().getSessionId());
-                }
-                if (cl.hasStats()) {
+
                     if (cl.getStats().hasRssi()) {
                         int rssi = cl.getStats().getRssi();
                         cMetrics.setRssi(rssi);
@@ -1459,9 +1489,10 @@ public class MqttStatsPublisher {
                     if (cl.getStats().hasTxRetries()) {
                         cMetrics.setNumTxDataRetries((int) cl.getStats().getTxRetries());
                     }
-                }
 
-                LOG.debug("ApClientMetrics Report {}", cMetrics);
+                    LOG.debug("ApClientMetrics Report {}", cMetrics);
+
+                }
 
             }
 
@@ -1529,98 +1560,12 @@ public class MqttStatsPublisher {
         }
     }
 
-    ClientSession handleClientSessionMetricsUpdate(int customerId, long equipmentId, long locationId, RadioType radioType, long timestamp,
-            sts.OpensyncStats.Client client) {
-        try
-
-        {
-            LOG.info("handleClientSessionUpdate for customerId {} equipmentId {} locationId {} client {} ", customerId, equipmentId, locationId,
-                    client.getMacAddress());
-
-            ClientSession clientSession = clientServiceInterface.getSessionOrNull(customerId, equipmentId, MacAddress.valueOf(client.getMacAddress()));
-
-            if (clientSession == null) {
-                LOG.warn("Cannot get client session for {}", client.getMacAddress());
-                return null;
-            }
-
-            ClientSessionDetails latestClientSessionDetails = clientSession.getDetails();
-            latestClientSessionDetails.setMetricDetails(calculateClientSessionMetricDetails(client, timestamp));
-
-            clientSession.getDetails().mergeSession(latestClientSessionDetails);
-            clientSession.setLastModifiedTimestamp(timestamp);
-
-            clientSession = clientServiceInterface.updateSession(clientSession);
-
-            LOG.debug("Updated client session {}", clientSession);
-
-            return clientSession;
-        } catch (Exception e) {
-            LOG.error("Error while attempting to create ClientSession and Info", e);
-        }
-        return null;
-    }
-
-    ClientSessionMetricDetails calculateClientSessionMetricDetails(sts.OpensyncStats.Client client, long timestamp) {
-
-        LOG.info("calculateClientSessionMetricDetails for Client {} at timestamp {}", client.getMacAddress(), timestamp);
-
-        ClientSessionMetricDetails metricDetails = new ClientSessionMetricDetails();
-
-        if (LOG.isDebugEnabled())
-            LOG.debug("Stats: {} DurationMs {}", client.getStats(), client.getDurationMs());
-        int rssi = client.getStats().getRssi();
-        metricDetails.setRssi(rssi);
-        metricDetails.setRxBytes(client.getStats().getRxBytes());
-        metricDetails.setTxBytes(client.getStats().getTxBytes());
-
-        // Frames : data chunk sent over data-link layer (Ethernet, ATM)
-        // Packets : data chunk sent over IP layer.
-        // in Wifi, these are the same size, so number of packets is equal to
-        // number of frames
-        metricDetails.setTotalTxPackets(client.getStats().getTxFrames());
-        metricDetails.setTotalRxPackets(client.getStats().getRxFrames());
-        metricDetails.setTxDataFrames((int) client.getStats().getTxFrames());
-        metricDetails.setRxDataFrames((int) client.getStats().getRxFrames());
-
-        metricDetails.setRxRateKbps((long) client.getStats().getRxRate());
-        metricDetails.setTxRateKbps((long) client.getStats().getTxRate());
-        if (LOG.isDebugEnabled())
-            LOG.debug("RxRateKbps {} TxRateKbps {}", metricDetails.getRxRateKbps(), metricDetails.getTxRateKbps());
-
-        // Throughput, do rate / duration
-        if (client.getDurationMs() > 1000) {
-            int durationSec = client.getDurationMs() / 1000;
-            // 1 Mbit = 125000 B
-
-            float rxBytesFv = Long.valueOf(client.getStats().getRxBytes()).floatValue();
-            float rxBytesToMb = rxBytesFv / 125000F;
-            float txBytesFv = Long.valueOf(client.getStats().getTxBytes()).floatValue();
-            float txBytesToMb = txBytesFv / 125000F;
-
-            if (LOG.isDebugEnabled())
-                LOG.debug("rxBytesToMb {} txBytesToMb {} ", rxBytesToMb, txBytesToMb);
-
-            metricDetails.setRxMbps(rxBytesToMb / durationSec);
-            metricDetails.setTxMbps(txBytesToMb / durationSec);
-            if (LOG.isDebugEnabled())
-                LOG.debug("RxMbps {} TxMbps {} ", metricDetails.getRxMbps(), metricDetails.getTxMbps());
-
-        } else {
-            LOG.warn("Cannot calculate tx/rx throughput for Client {} based on duration of {} Ms", client.getMacAddress(), client.getDurationMs());
-        }
-        metricDetails.setLastMetricTimestamp(timestamp);
-
-        return metricDetails;
-    }
-
     void populateApSsidMetrics(List<ServiceMetric> metricRecordList, Report report, int customerId, long equipmentId, String apId, long locationId) {
 
         LOG.info("populateApSsidMetrics for Customer {} Equipment {}", customerId, equipmentId);
         ServiceMetric smr = new ServiceMetric(customerId, equipmentId);
         smr.setLocationId(locationId);
         ApSsidMetrics apSsidMetrics = new ApSsidMetrics();
-
         smr.setDetails(apSsidMetrics);
         metricRecordList.add(smr);
 
@@ -1662,7 +1607,8 @@ public class MqttStatsPublisher {
                     }
                 }
             }
-            LOG.debug("Client Report Date is {}", new Date(clientReport.getTimestampMs()));
+            if (LOG.isTraceEnabled())
+                LOG.trace("Client Report Date is {}", new Date(clientReport.getTimestampMs()));
             int numConnectedClients = 0;
             for (Client client : clientReport.getClientListList()) {
                 if (client.hasStats()) {
@@ -1873,10 +1819,9 @@ public class MqttStatsPublisher {
             return -1;
         }
 
-        OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
-
-        if (ovsdbSession != null) {
-            return ovsdbSession.getCustomerId();
+        Equipment ce = equipmentServiceInterface.getByInventoryIdOrNull(apId);
+        if (ce != null) {
+            return ce.getCustomerId();
         }
 
         return -1;

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

@@ -8,6 +8,7 @@ import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
 import com.telecominfraproject.wlan.client.models.events.realtime.ClientConnectSuccessEvent;
@@ -89,7 +90,8 @@ public class RealtimeEventPublisher {
     private ProfileServiceInterface  profileServiceInterface;
 
     private static final Logger LOG = LoggerFactory.getLogger(RealtimeEventPublisher.class);
-
+    
+    @Async
     void publishChannelHopEvents(int customerId, long equipmentId, long locationId, EventReport e) {
 
         LOG.info("publishChannelHopEvents for customerId {} equipmentId {}");
@@ -178,6 +180,7 @@ public class RealtimeEventPublisher {
         }
     }
 
+    @Async
     void publishClientConnectSuccessEvent(int customerId, long equipmentId, long locationId, ClientConnectEvent clientConnectEvent) {
 
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientConnectEvent, customerId,
@@ -187,7 +190,7 @@ public class RealtimeEventPublisher {
         clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
                 .getRadioTypeFromOpensyncStatsRadioBandType(clientConnectEvent.getBand()));
         clientEvent.setSsid(clientConnectEvent.getSsid());
-        clientEvent.setSessionId(clientConnectEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientConnectEvent.getSessionId()));
 
         if (clientConnectEvent.hasFbtUsed()) {
             clientEvent.setFbtUsed(clientConnectEvent.getFbtUsed());
@@ -266,7 +269,7 @@ public class RealtimeEventPublisher {
         cloudEventDispatcherInterface.publishEvent(clientEvent);
 
     }
-
+    @Async
     void publishClientDisconnectEvent(int customerId, long equipmentId, long locationId, ClientDisconnectEvent clientDisconnectEvent) {
 
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientDisconnectEvent, customerId,
@@ -275,7 +278,7 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientDisconnectEvent(
                 clientDisconnectEvent.getTimestampMs());
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientDisconnectEvent.getStaMac()));
-        clientEvent.setSessionId(clientDisconnectEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientDisconnectEvent.getSessionId()));
         clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
                 .getRadioTypeFromOpensyncStatsRadioBandType(clientDisconnectEvent.getBand()));
         clientEvent.setSsid(clientDisconnectEvent.getSsid());
@@ -316,13 +319,14 @@ public class RealtimeEventPublisher {
         cloudEventDispatcherInterface.publishEvent(clientEvent);
 
     }
-
+    
+   @Async
     void publishClientAuthSystemEvent(int customerId, long equipmentId, long locationId, ClientAuthEvent clientAuthEvent) {
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientAuthEvent, customerId, equipmentId);
 
         com.telecominfraproject.wlan.client.models.events.realtime.ClientAuthEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAuthEvent(
                 clientAuthEvent.getTimestampMs());
-        clientEvent.setSessionId(clientAuthEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientAuthEvent.getSessionId()));
         clientEvent.setSsid(clientAuthEvent.getSsid());
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientAuthEvent.getStaMac()));
         clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
@@ -340,14 +344,15 @@ public class RealtimeEventPublisher {
         cloudEventDispatcherInterface.publishEvent(clientEvent);
 
     }
-
+   
+   @Async
     void publishClientAssocEvent(int customerId, long equipmentId, long locationId, ClientAssocEvent clientAssocEvent) {
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientAssocEvent, customerId, equipmentId);
 
         com.telecominfraproject.wlan.client.models.events.realtime.ClientAssocEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientAssocEvent(
                 clientAssocEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientAssocEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientAssocEvent.getSessionId()));
         clientEvent.setSsid(clientAssocEvent.getSsid());
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientAssocEvent.getStaMac()));
         clientEvent.setRadioType(OvsdbToWlanCloudTypeMappingUtility
@@ -390,7 +395,7 @@ public class RealtimeEventPublisher {
         cloudEventDispatcherInterface.publishEvent(clientEvent);
 
     }
-
+   @Async
     void publishClientFailureEvent(int customerId, long equipmentId, long locationId, ClientFailureEvent clientFailureEvent) {
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientFailureEvent, customerId,
                 equipmentId);
@@ -398,7 +403,7 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientFailureEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFailureEvent(
                 clientFailureEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientFailureEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientFailureEvent.getSessionId()));
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientFailureEvent.getStaMac()));
         clientEvent.setSsid(clientFailureEvent.getSsid());
 
@@ -417,7 +422,7 @@ public class RealtimeEventPublisher {
         LOG.info("publishing client event {} to cloud", clientEvent);
         cloudEventDispatcherInterface.publishEvent(clientEvent);
     }
-
+   @Async
     void publishClientFirstDataEvent(int customerId, long equipmentId, long locationId, ClientFirstDataEvent clientFirstDataEvent) {
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientFirstDataEvent, customerId,
                 equipmentId);
@@ -425,7 +430,7 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientFirstDataEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientFirstDataEvent(
                 clientFirstDataEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientFirstDataEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientFirstDataEvent.getSessionId()));
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientFirstDataEvent.getStaMac()));
 
         if (clientFirstDataEvent.hasFdataTxUpTsInUs()) {
@@ -444,7 +449,7 @@ public class RealtimeEventPublisher {
         cloudEventDispatcherInterface.publishEvent(clientEvent);
 
     }
-
+   @Async
     void publishClientIdEvent(int customerId, long equipmentId, long locationId, ClientIdEvent clientIdEvent) {
 
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientIdEvent, customerId, equipmentId);
@@ -452,7 +457,7 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientIdEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIdEvent(
                 clientIdEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientIdEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientIdEvent.getSessionId()));
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientIdEvent.getCltMac()));
         if (clientIdEvent.hasCltId()) {
             clientEvent.setUserId(clientIdEvent.getCltId());
@@ -465,7 +470,7 @@ public class RealtimeEventPublisher {
         LOG.info("publishing client event {} to cloud", clientEvent);
         cloudEventDispatcherInterface.publishEvent(clientEvent);
     }
-
+   @Async
     void publishClientIpEvent(int customerId, long equipmentId, long locationId, ClientIpEvent clientIpEvent) {
 
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientIpEvent, customerId, equipmentId);
@@ -473,11 +478,16 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientIpAddressEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientIpAddressEvent(
                 clientIpEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientIpEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientIpEvent.getSessionId()));
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientIpEvent.getStaMac()));
         if (clientIpEvent.hasIpAddr()) {
-            clientEvent.setIpAddr(clientIpEvent.getIpAddr().toByteArray());
+            try {
+                clientEvent.setIpAddr(InetAddress.getByAddress(clientIpEvent.getIpAddr().toByteArray()));
+            } catch (UnknownHostException e1) {
+                LOG.error("Invalid Client IP Address for equipmentId {}, clientIpEvent {}", equipmentId, clientIpEvent);
+            }
         }
+
         clientEvent.setEventTimestamp(clientIpEvent.getTimestampMs());
         clientEvent.setCustomerId(customerId);
         clientEvent.setEquipmentId(equipmentId);
@@ -486,7 +496,7 @@ public class RealtimeEventPublisher {
         LOG.info("publishing client event {} to cloud", clientEvent);
         cloudEventDispatcherInterface.publishEvent(clientEvent);
     }
-
+   @Async
     void publishClientTimeoutEvent(int customerId, long equipmentId, long locationId, ClientTimeoutEvent clientTimeoutEvent) {
 
         LOG.info("Received ClientEvent {} for customerId {} equipmentId {}", clientTimeoutEvent, customerId,
@@ -495,7 +505,7 @@ public class RealtimeEventPublisher {
         com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent clientEvent = new com.telecominfraproject.wlan.client.models.events.realtime.ClientTimeoutEvent(
                 clientTimeoutEvent.getTimestampMs());
 
-        clientEvent.setSessionId(clientTimeoutEvent.getSessionId());
+        clientEvent.setSessionId(Long.toUnsignedString( clientTimeoutEvent.getSessionId()));
         clientEvent.setClientMacAddress(MacAddress.valueOf(clientTimeoutEvent.getStaMac()));
         if (clientTimeoutEvent.hasRCode()) {
             clientEvent.setTimeoutReason(clientTimeoutEvent.getRCode().equals(CTReasonType.CTR_IDLE_TOO_LONG)
@@ -518,6 +528,7 @@ public class RealtimeEventPublisher {
 
     }
 
+   @Async
     void publishDhcpTransactionEvents(int customerId, long equipmentId, long locationId, List<DhcpTransaction> dhcpTransactionList) {
         LOG.info("Publish Dhcp Transaction Events for customer {} equipmentId {}", customerId, equipmentId);
         List<SystemEvent> dhcpEventsList = new ArrayList<>();
@@ -798,6 +809,7 @@ public class RealtimeEventPublisher {
         return cloudDhcpEvent;
     }
 
+    @Async
     void publishSipCallEvents(int customerId, long equipmentId, long locationId, List<VideoVoiceReport> sipCallReportList) {
         // only in case it is not there, we will just use the time when we
         // received the report/event
@@ -849,7 +861,7 @@ public class RealtimeEventPublisher {
             cloudSipCallReportEvent.setEventType(RealTimeEventType.SipCallReport);
 
             cloudSipCallReportEvent.setSipCallId(callReport.getWifiSessionId());
-            cloudSipCallReportEvent.setAssociationId(callReport.getSessionId());
+            cloudSipCallReportEvent.setAssociationId(Long.toUnsignedString( callReport.getSessionId()));
 
             if (callReport.hasReason()) {
                 cloudSipCallReportEvent.setReportReason(getCallReportReason(callReport.getReason()));
@@ -915,11 +927,11 @@ public class RealtimeEventPublisher {
             }
 
             if (apCallStart.hasSessionId()) {
-                cloudSipCallStartEvent.setAssociationId(apCallStart.getSessionId());
+                cloudSipCallStartEvent.setAssociationId(Long.toUnsignedString( apCallStart.getSessionId()));
             }
 
             if (apCallStart.hasWifiSessionId()) {
-                cloudSipCallStartEvent.setAssociationId(apCallStart.getWifiSessionId());
+                cloudSipCallStartEvent.setAssociationId(Long.toUnsignedString( apCallStart.getWifiSessionId()));
             }
 
             if (apCallStart.getCodecsCount() > 0) {
@@ -979,7 +991,7 @@ public class RealtimeEventPublisher {
 
             if (apCallStop.hasSessionId()) {
 
-                cloudSipCallStopEvent.setAssociationId(apCallStop.getSessionId());
+                cloudSipCallStopEvent.setAssociationId(Long.toUnsignedString(  apCallStop.getSessionId()));
 
             }
 
@@ -1043,11 +1055,11 @@ public class RealtimeEventPublisher {
             }
 
             if (apStreamVideoServer.hasSessionId()) {
-                rtsStartEvent.setSessionId(apStreamVideoServer.getSessionId());
+                rtsStartEvent.setSessionId(Long.toUnsignedString( apStreamVideoServer.getSessionId()));
             }
 
             if (apStreamVideoServer.hasVideoSessionId()) {
-                rtsStartEvent.setVideoSessionId(apStreamVideoServer.getVideoSessionId());
+                rtsStartEvent.setVideoSessionId(Long.toUnsignedString(  apStreamVideoServer.getVideoSessionId()));
             }
 
             eventsList.add(rtsStartEvent);
@@ -1167,7 +1179,7 @@ public class RealtimeEventPublisher {
             }
 
             if (apStreamVideoSessionStart.hasSessionId()) {
-                rtsStartSessionEvent.setSessionId(apStreamVideoSessionStart.getSessionId());
+                rtsStartSessionEvent.setSessionId(Long.toUnsignedString( apStreamVideoSessionStart.getSessionId()));
 
             }
 
@@ -1178,7 +1190,7 @@ public class RealtimeEventPublisher {
             }
 
             if (apStreamVideoSessionStart.hasVideoSessionId()) {
-                rtsStartSessionEvent.setVideoSessionId(apStreamVideoSessionStart.getVideoSessionId());
+                rtsStartSessionEvent.setVideoSessionId(Long.toUnsignedString( apStreamVideoSessionStart.getVideoSessionId()));
             }
             eventsList.add(rtsStartSessionEvent);
         }
@@ -1208,7 +1220,7 @@ public class RealtimeEventPublisher {
             }
 
             if (apStreamVideoStop.hasSessionId()) {
-                rtsStopEvent.setSessionId(apStreamVideoStop.getSessionId());
+                rtsStopEvent.setSessionId(Long.toUnsignedString( apStreamVideoStop.getSessionId()));
             }
 
             if (apStreamVideoStop.hasStreamingVideoType()) {
@@ -1223,7 +1235,7 @@ public class RealtimeEventPublisher {
             }
 
             if (apStreamVideoStop.hasVideoSessionId()) {
-                rtsStopEvent.setVideoSessionId(apStreamVideoStop.getVideoSessionId());
+                rtsStopEvent.setVideoSessionId(Long.toUnsignedString(  apStreamVideoStop.getVideoSessionId()));
             }
 
             eventsList.add(rtsStopEvent);

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

@@ -1,3 +1,4 @@
+
 package com.telecominfraproject.wlan.opensync.external.integration;
 
 import static org.junit.Assert.assertNotNull;
@@ -49,6 +50,8 @@ import com.telecominfraproject.wlan.core.model.entity.CountryCode;
 import com.telecominfraproject.wlan.core.model.equipment.EquipmentType;
 import com.telecominfraproject.wlan.core.model.equipment.MacAddress;
 import com.telecominfraproject.wlan.core.model.equipment.RadioType;
+import com.telecominfraproject.wlan.core.model.pagination.PaginationContext;
+import com.telecominfraproject.wlan.core.model.pagination.PaginationResponse;
 import com.telecominfraproject.wlan.customer.models.Customer;
 import com.telecominfraproject.wlan.customer.models.CustomerDetails;
 import com.telecominfraproject.wlan.customer.models.EquipmentAutoProvisioningSettings;
@@ -95,10 +98,9 @@ import sts.OpensyncStats.RadioBandType;
 import sts.OpensyncStats.Report;
 
 @RunWith(SpringRunner.class)
-@ActiveProfiles(profiles = { "integration_test", })
+@ActiveProfiles(profiles = {"integration_test",})
 @SpringBootTest(webEnvironment = WebEnvironment.NONE, classes = OpensyncExternalIntegrationCloudTest.class)
-@Import(value = { AlarmServiceInterface.class, OpensyncExternalIntegrationCloud.class,
-        OpensyncExternalIntegrationCloudTest.Config.class,
+@Import(value = {AlarmServiceInterface.class, OpensyncExternalIntegrationCloud.class, OpensyncExternalIntegrationCloudTest.Config.class,
 
 })
 public class OpensyncExternalIntegrationCloudTest {
@@ -133,7 +135,6 @@ public class OpensyncExternalIntegrationCloudTest {
     @Autowired
     OpensyncExternalIntegrationCloud opensyncExternalIntegrationCloud;
 
-
     MockitoSession mockito;
 
     @Configuration
@@ -194,25 +195,21 @@ public class OpensyncExternalIntegrationCloudTest {
         ssidProfile.setDetails(SsidConfiguration.createWithDefaults());
 
         apProfile.setChildProfileIds(ImmutableSet.of(ssidProfile.getId()));
-        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class))).thenAnswer(i -> i.getArguments()[0]);
         // .thenReturn(ssidProfile);
-        Mockito.when(profileServiceInterface.update(ArgumentMatchers.any(Profile.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(profileServiceInterface.update(ArgumentMatchers.any(Profile.class))).thenAnswer(i -> i.getArguments()[0]);
 
         Status fwStatus = new Status();
         fwStatus.setDetails(new EquipmentUpgradeStatusData());
-        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.eq(StatusDataType.FIRMWARE))).thenReturn(fwStatus);
+        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.eq(StatusDataType.FIRMWARE)))
+                .thenReturn(fwStatus);
 
-        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting())
-                .thenReturn(new CustomerFirmwareTrackSettings());
+        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting()).thenReturn(new CustomerFirmwareTrackSettings());
         CustomerFirmwareTrackRecord fwTrackRecord = new CustomerFirmwareTrackRecord();
         fwTrackRecord.setSettings(new CustomerFirmwareTrackSettings());
         fwTrackRecord.setTrackRecordId(3);
         fwTrackRecord.setCustomerId(2);
-        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt()))
-                .thenReturn(fwTrackRecord);
+        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt())).thenReturn(fwTrackRecord);
 
         Equipment equipment = new Equipment();
         equipment.setCustomerId(2);
@@ -224,22 +221,16 @@ public class OpensyncExternalIntegrationCloudTest {
         equipment.setDetails(ApElementConfiguration.createWithDefaults());
 
         Mockito.when(equipmentServiceInterface.get(1L)).thenReturn(equipment);
-        Mockito.when(
-                equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
-                .thenReturn(equipment);
+        Mockito.when(equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122"))).thenReturn(equipment);
 
-        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
-        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
 
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
-
         Mockito.when(ovsdbSessionMapInterface.getSession("Test_Client_21P10C68818122")).thenReturn(session);
 
         opensyncExternalIntegrationCloud.apConnected("Test_Client_21P10C68818122", createConnectNodeInfo());
 
-
         Mockito.verify(firmwareServiceInterface).getDefaultCustomerTrackSetting();
         Mockito.verify(locationServiceInterface).get(8L);
 
@@ -267,7 +258,7 @@ public class OpensyncExternalIntegrationCloudTest {
         customer.setDetails(customerDetails);
 
         Mockito.when(customerServiceInterface.getOrNull(ArgumentMatchers.anyInt())).thenReturn(customer);
-        
+
         Profile rfProfile = new Profile();
         rfProfile.setId(1);
         rfProfile.setName("testRfProfile");
@@ -282,21 +273,20 @@ public class OpensyncExternalIntegrationCloudTest {
         Profile ssidProfile = new Profile();
         ssidProfile.setId(2);
         ssidProfile.setDetails(SsidConfiguration.createWithDefaults());
-        
+
         Set<Long> childProfileIds = new HashSet<>();
-        childProfileIds.add(rfProfile.getId()); 
-        childProfileIds.add(ssidProfile.getId());   
+        childProfileIds.add(rfProfile.getId());
+        childProfileIds.add(ssidProfile.getId());
         apProfile.setChildProfileIds(childProfileIds);
 
-        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class))).thenReturn(apProfile)
-                .thenReturn(ssidProfile).thenReturn(rfProfile);
+        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class))).thenReturn(apProfile).thenReturn(ssidProfile).thenReturn(rfProfile);
         Mockito.when(profileServiceInterface.update(ArgumentMatchers.any(Profile.class))).thenReturn(apProfile);
-        
+
         List<Profile> profileList = new ArrayList<>();
         profileList.add(apProfile);
         profileList.add(rfProfile);
         profileList.add(ssidProfile);
-        
+
         Mockito.when(profileServiceInterface.getProfileWithChildren(ArgumentMatchers.anyLong())).thenReturn(profileList);
 
         Equipment equipment = new Equipment();
@@ -310,27 +300,21 @@ public class OpensyncExternalIntegrationCloudTest {
 
         Mockito.when(equipmentServiceInterface.get(1L)).thenReturn(equipment);
 
-        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
-        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
-        Mockito.when(
-                equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
-                .thenReturn(null);
+        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122"))).thenReturn(null);
 
         Status fwStatus = new Status();
         fwStatus.setDetails(new EquipmentUpgradeStatusData());
-        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.eq(StatusDataType.FIRMWARE))).thenReturn(fwStatus);
+        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.eq(StatusDataType.FIRMWARE)))
+                .thenReturn(fwStatus);
 
-        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting())
-                .thenReturn(new CustomerFirmwareTrackSettings());
+        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting()).thenReturn(new CustomerFirmwareTrackSettings());
         CustomerFirmwareTrackRecord fwTrackRecord = new CustomerFirmwareTrackRecord();
         fwTrackRecord.setSettings(new CustomerFirmwareTrackSettings());
         fwTrackRecord.setTrackRecordId(3);
         fwTrackRecord.setCustomerId(2);
-        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt()))
-                .thenReturn(fwTrackRecord);
+        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt())).thenReturn(fwTrackRecord);
 
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
 
@@ -342,7 +326,7 @@ public class OpensyncExternalIntegrationCloudTest {
         Mockito.verify(customerServiceInterface).getOrNull(ArgumentMatchers.anyInt());
         Mockito.verify(equipmentServiceInterface).getByInventoryIdOrNull("Test_Client_21P10C68818122");
         Mockito.verify(firmwareServiceInterface).getDefaultCustomerTrackSetting();
-        Mockito.verify(locationServiceInterface,  Mockito.times(2)).get(ArgumentMatchers.anyLong());
+        Mockito.verify(locationServiceInterface, Mockito.times(2)).get(ArgumentMatchers.anyLong());
 
     }
 
@@ -355,11 +339,9 @@ public class OpensyncExternalIntegrationCloudTest {
         ssidProfile.setDetails(SsidConfiguration.createWithDefaults());
 
         apProfile.setChildProfileIds(ImmutableSet.of(ssidProfile.getId()));
-        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(profileServiceInterface.create(ArgumentMatchers.any(Profile.class))).thenAnswer(i -> i.getArguments()[0]);
         // .thenReturn(ssidProfile);
-        Mockito.when(profileServiceInterface.update(ArgumentMatchers.any(Profile.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(profileServiceInterface.update(ArgumentMatchers.any(Profile.class))).thenAnswer(i -> i.getArguments()[0]);
 
         Location location = new Location();
         location.setId(8L);
@@ -381,26 +363,20 @@ public class OpensyncExternalIntegrationCloudTest {
         equipment.setDetails(ApElementConfiguration.createWithDefaults());
         Status fwStatus = new Status();
         fwStatus.setDetails(new EquipmentUpgradeStatusData());
-        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.eq(StatusDataType.FIRMWARE))).thenReturn(fwStatus);
+        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.eq(StatusDataType.FIRMWARE)))
+                .thenReturn(fwStatus);
 
-        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting())
-                .thenReturn(new CustomerFirmwareTrackSettings());
+        Mockito.when(firmwareServiceInterface.getDefaultCustomerTrackSetting()).thenReturn(new CustomerFirmwareTrackSettings());
         CustomerFirmwareTrackRecord fwTrackRecord = new CustomerFirmwareTrackRecord();
         fwTrackRecord.setSettings(new CustomerFirmwareTrackSettings());
         fwTrackRecord.setTrackRecordId(3);
         fwTrackRecord.setCustomerId(2);
-        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt()))
-                .thenReturn(fwTrackRecord);
+        Mockito.when(firmwareServiceInterface.getCustomerFirmwareTrackRecord(ArgumentMatchers.anyInt())).thenReturn(fwTrackRecord);
         Mockito.when(equipmentServiceInterface.get(1L)).thenReturn(equipment);
 
-        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
-        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class)))
-                .thenAnswer(i -> i.getArguments()[0]);
-        Mockito.when(
-                equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
-                .thenReturn(equipment);
+        Mockito.when(equipmentServiceInterface.create(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.update(ArgumentMatchers.any(Equipment.class))).thenAnswer(i -> i.getArguments()[0]);
+        Mockito.when(equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122"))).thenReturn(equipment);
 
         Customer customer = new Customer();
         customer.setId(2);
@@ -446,8 +422,7 @@ public class OpensyncExternalIntegrationCloudTest {
         ssidProfile.setDetails(SsidConfiguration.createWithDefaults());
 
         List<Profile> profileWithChildren = ImmutableList.of(apProfile, ssidProfile);
-        Mockito.when(profileServiceInterface.getProfileWithChildren(ArgumentMatchers.anyLong()))
-                .thenReturn(profileWithChildren);
+        Mockito.when(profileServiceInterface.getProfileWithChildren(ArgumentMatchers.anyLong())).thenReturn(profileWithChildren);
 
         assertNotNull(opensyncExternalIntegrationCloud.getApConfig("Test_Client_21P10C68818122"));
 
@@ -473,8 +448,7 @@ public class OpensyncExternalIntegrationCloudTest {
     @Test
     public void testProcessMqttMessageStringReport() {
 
-        Report report = Report.newBuilder().setNodeID("21P10C68818122")
-                .addAllClients(getOpensyncStatsClientReportsList())
+        Report report = Report.newBuilder().setNodeID("21P10C68818122").addAllClients(getOpensyncStatsClientReportsList())
                 .addAllEventReport(getOpensyncStatsEventReportsList()).build();
 
         String topic = "/ap/Test_Client_21P10C68818122/opensync";
@@ -487,8 +461,8 @@ public class OpensyncExternalIntegrationCloudTest {
         activeBssidsDetails.setActiveBSSIDs(getActiveBssidList());
         bssidStatus.setDetails(activeBssidsDetails);
 
-        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.eq(StatusDataType.ACTIVE_BSSIDS))).thenReturn(bssidStatus);
+        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.eq(StatusDataType.ACTIVE_BSSIDS)))
+                .thenReturn(bssidStatus);
 
         Mockito.when(statusServiceInterface.update(ArgumentMatchers.any(Status.class))).thenReturn(bssidStatus);
         com.telecominfraproject.wlan.client.models.Client clientInstance = new com.telecominfraproject.wlan.client.models.Client();
@@ -497,12 +471,10 @@ public class OpensyncExternalIntegrationCloudTest {
         com.telecominfraproject.wlan.client.models.Client clientInstance2 = new com.telecominfraproject.wlan.client.models.Client();
         clientInstance2.setMacAddress(MacAddress.valueOf("C0:9A:D0:76:A9:69"));
         clientInstance2.setDetails(new ClientInfoDetails());
-        Mockito.when(
-                clientServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.any(MacAddress.class)))
-                .thenReturn(clientInstance).thenReturn(clientInstance2);
-        Mockito.when(clientServiceInterface
-                .update(ArgumentMatchers.any(com.telecominfraproject.wlan.client.models.Client.class)))
-                .thenReturn(clientInstance).thenReturn(clientInstance2);
+        Mockito.when(clientServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.any(MacAddress.class))).thenReturn(clientInstance)
+                .thenReturn(clientInstance2);
+        Mockito.when(clientServiceInterface.update(ArgumentMatchers.any(com.telecominfraproject.wlan.client.models.Client.class))).thenReturn(clientInstance)
+                .thenReturn(clientInstance2);
 
         ClientSession clientSession = new ClientSession();
         clientSession.setMacAddress(MacAddress.valueOf("7C:AB:60:E6:EA:4D"));
@@ -510,16 +482,10 @@ public class OpensyncExternalIntegrationCloudTest {
         ClientSession clientSession2 = new ClientSession();
         clientSession2.setMacAddress(MacAddress.valueOf("C0:9A:D0:76:A9:69"));
         clientSession2.setDetails(new ClientSessionDetails());
-        Mockito.when(clientServiceInterface.getSessionOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.any(MacAddress.class))).thenReturn(clientSession).thenReturn(clientSession2);
-
-        Mockito.when(clientServiceInterface.updateSession(ArgumentMatchers.any(ClientSession.class)))
+        Mockito.when(clientServiceInterface.getSessionOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.any(MacAddress.class)))
                 .thenReturn(clientSession).thenReturn(clientSession2);
 
-        opensyncExternalIntegrationCloud.processMqttMessage(topic, report);
-
-        Mockito.verify(opensyncExternalIntegrationMqttProcessor, Mockito.times(1)).processMqttMessage(topic, report);
-
+        Mockito.when(clientServiceInterface.updateSession(ArgumentMatchers.any(ClientSession.class))).thenReturn(clientSession).thenReturn(clientSession2);
     }
 
     @Ignore
@@ -587,10 +553,10 @@ public class OpensyncExternalIntegrationCloudTest {
 
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
         Mockito.when(session.getEquipmentId()).thenReturn(1L);
-        Mockito.when(session.getCustomerId()).thenReturn(2);
 
         Equipment equipment = new Equipment();
-        equipment.setCustomerId(1);
+        equipment.setCustomerId(2);
+        equipment.setId(1L);
         equipment.setEquipmentType(EquipmentType.AP);
         equipment.setInventoryId(apId);
         equipment.setProfileId(1);
@@ -599,7 +565,7 @@ public class OpensyncExternalIntegrationCloudTest {
         Mockito.when(equipmentServiceInterface.getOrNull(1L)).thenReturn(equipment);
         Mockito.when(equipmentServiceInterface.get(equipment.getId())).thenReturn(equipment);
         Mockito.when(equipmentServiceInterface.update(equipment)).thenReturn(equipment);
-        
+
         Profile rfProfile = new Profile();
         rfProfile.setName("testRfProfile");
         rfProfile.setId(2);
@@ -607,30 +573,28 @@ public class OpensyncExternalIntegrationCloudTest {
         rfProfile.setProfileType(ProfileType.rf);
         Set<Long> childProfileIds = new HashSet<>();
         childProfileIds.add(rfProfile.getId());
-        
+
         Profile apProfile = new Profile();
         apProfile.setName("testApProfile");
         apProfile.setId(1);
         apProfile.setProfileType(ProfileType.equipment_ap);
         apProfile.setChildProfileIds(childProfileIds);
-        
+
         List<Profile> profileList = new ArrayList<>();
         profileList.add(apProfile);
         profileList.add(rfProfile);
-        
+
         Mockito.when(profileServiceInterface.getProfileWithChildren(ArgumentMatchers.anyLong())).thenReturn(profileList);
 
         Mockito.when(ovsdbSessionMapInterface.getSession(apId)).thenReturn(session);
 
-        opensyncExternalIntegrationCloud.wifiVIFStateDbTableUpdate(ImmutableList.of(vifState1, vifState2, vifState3),
-                apId);
+        opensyncExternalIntegrationCloud.wifiVIFStateDbTableUpdate(ImmutableList.of(vifState1, vifState2, vifState3), apId);
 
-        Mockito.verify(session).getCustomerId();
         Mockito.verify(session).getEquipmentId();
         Mockito.verify(ovsdbSessionMapInterface).getSession(apId);
         Mockito.verify(equipmentServiceInterface).getOrNull(1L);
-
         Mockito.verify(statusServiceInterface).getOrNull(2, 1L, StatusDataType.CLIENT_DETAILS);
+
         Mockito.verify(statusServiceInterface).getOrNull(2, 1L, StatusDataType.ACTIVE_BSSIDS);
         Mockito.verify(statusServiceInterface).update(clientStatus);
 
@@ -662,8 +626,7 @@ public class OpensyncExternalIntegrationCloudTest {
         radioState3.setChannel(149);
         radioState3.setVifStates(ImmutableSet.of(new Uuid(UUID.randomUUID())));
         radioState3.setFreqBand(RadioType.is5GHzL);
-        radioState3.setAllowedChannels(
-                ImmutableSet.of(00, 104, 108, 112, 116, 132, 136, 140, 144, 149, 153, 157, 161, 165));
+        radioState3.setAllowedChannels(ImmutableSet.of(00, 104, 108, 112, 116, 132, 136, 140, 144, 149, 153, 157, 161, 165));
         radioState3.setTxPower(32);
         radioState3.setEnabled(true);
         radioState3.setCountry("CA");
@@ -686,7 +649,7 @@ public class OpensyncExternalIntegrationCloudTest {
         protocolStatus.setStatusDataType(StatusDataType.PROTOCOL);
 
         Mockito.when(statusServiceInterface.getOrNull(2, 1L, StatusDataType.PROTOCOL)).thenReturn(protocolStatus);
-        
+
         Status channelStatus = new Status();
         channelStatus.setCustomerId(2);
         channelStatus.setEquipmentId(1L);
@@ -724,20 +687,18 @@ public class OpensyncExternalIntegrationCloudTest {
 
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
         Mockito.when(session.getEquipmentId()).thenReturn(1L);
-        Mockito.when(session.getCustomerId()).thenReturn(2);
 
         Mockito.when(ovsdbSessionMapInterface.getSession(apId)).thenReturn(session);
 
-        opensyncExternalIntegrationCloud
-                .wifiRadioStatusDbTableUpdate(ImmutableList.of(radioState1, radioState2, radioState3), apId);
+        opensyncExternalIntegrationCloud.wifiRadioStatusDbTableUpdate(ImmutableList.of(radioState1, radioState2, radioState3), apId);
 
-        Mockito.verify(session).getCustomerId();
         Mockito.verify(session).getEquipmentId();
 
         Mockito.verify(ovsdbSessionMapInterface).getSession(apId);
         Mockito.verify(equipmentServiceInterface, Mockito.times(1)).getByInventoryIdOrNull(apId);
+        Mockito.verify(statusServiceInterface, Mockito.times(1)).getOrNull(1, 1L, StatusDataType.RADIO_CHANNEL);
 
-        Mockito.verify(statusServiceInterface, Mockito.times(3)).getOrNull(2, 1L, StatusDataType.PROTOCOL);
+        Mockito.verify(statusServiceInterface, Mockito.times(3)).getOrNull(1, 1L, StatusDataType.PROTOCOL);
         Mockito.verify(statusServiceInterface, Mockito.never()).update(bssidStatus);
 
     }
@@ -759,6 +720,23 @@ public class OpensyncExternalIntegrationCloudTest {
 
     @Test
     public void testWifiVIFStateDbTableDelete() {
+        String apId = "Test_Client_21P10C68818122";
+
+        Equipment equipment = new Equipment();
+        equipment.setCustomerId(2);
+        equipment.setId(1L);
+        equipment.setEquipmentType(EquipmentType.AP);
+        equipment.setInventoryId(apId);
+        equipment.setProfileId(1);
+        equipment.setDetails(ApElementConfiguration.createWithDefaults());
+
+        Mockito.when(equipmentServiceInterface.getByInventoryIdOrNull(apId)).thenReturn(equipment);
+        Mockito.when(equipmentServiceInterface.getOrNull(1L)).thenReturn(equipment);
+
+        OvsdbSession session = new OvsdbSession();
+        session.setApId(apId);
+        session.setEquipmentId(1L);
+        Mockito.when(ovsdbSessionMapInterface.getSession(apId)).thenReturn(session);
 
         Status bssidStatus = new Status();
         bssidStatus.setStatusDataType(StatusDataType.ACTIVE_BSSIDS);
@@ -768,16 +746,36 @@ public class OpensyncExternalIntegrationCloudTest {
         activeBssidsDetails.setActiveBSSIDs(getActiveBssidList());
         bssidStatus.setDetails(activeBssidsDetails);
 
-        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(),
-                ArgumentMatchers.eq(StatusDataType.ACTIVE_BSSIDS))).thenReturn(bssidStatus);
+        Mockito.when(statusServiceInterface.getOrNull(ArgumentMatchers.anyInt(), ArgumentMatchers.anyLong(), ArgumentMatchers.eq(StatusDataType.ACTIVE_BSSIDS)))
+                .thenReturn(bssidStatus);
+        PaginationResponse<ClientSession> pr = Mockito.mock(PaginationResponse.class,Mockito.RETURNS_MOCKS);
+        Mockito.when(clientServiceInterface.getSessionsForCustomer(Mockito.anyInt(), Mockito.anySet(),
+                Mockito.anySet(), Mockito.isNull(), Mockito.isNull(),  Mockito.any())).thenReturn(pr);
 
-        opensyncExternalIntegrationCloud.wifiVIFStateDbTableDelete(ImmutableList.of(new OpensyncAPVIFState()), "apId");
+
+        opensyncExternalIntegrationCloud.wifiVIFStateDbTableDelete(ImmutableList.of(new OpensyncAPVIFState()), "Test_Client_21P10C68818122");
     }
 
     @Test
     public void testWifiAssociatedClientsDbTableDelete() {
+        String apId = "Test_Client_21P10C68818122";
 
-        opensyncExternalIntegrationCloud.wifiAssociatedClientsDbTableDelete("7C:AB:60:E6:EA:4D", "apId");
+        Equipment equipment = new Equipment();
+        equipment.setCustomerId(2);
+        equipment.setId(1L);
+        equipment.setEquipmentType(EquipmentType.AP);
+        equipment.setInventoryId(apId);
+        equipment.setProfileId(1);
+        equipment.setDetails(ApElementConfiguration.createWithDefaults());
+
+        Mockito.when(equipmentServiceInterface.getByInventoryIdOrNull(apId)).thenReturn(equipment);
+        Mockito.when(equipmentServiceInterface.getOrNull(1L)).thenReturn(equipment);
+
+        OvsdbSession session = new OvsdbSession();
+        session.setApId(apId);
+        session.setEquipmentId(1L);
+        Mockito.when(ovsdbSessionMapInterface.getSession(apId)).thenReturn(session);
+        opensyncExternalIntegrationCloud.wifiAssociatedClientsDbTableDelete("7C:AB:60:E6:EA:4D", "Test_Client_21P10C68818122");
     }
 
     // Helper methods
@@ -805,11 +803,9 @@ public class OpensyncExternalIntegrationCloudTest {
         return bssidList;
     }
 
-
     private List<EventReport> getOpensyncStatsEventReportsList() {
 
-        sts.OpensyncStats.EventReport.ClientAssocEvent.Builder clientAssocBuilder = EventReport.ClientAssocEvent
-                .getDefaultInstance().toBuilder();
+        sts.OpensyncStats.EventReport.ClientAssocEvent.Builder clientAssocBuilder = EventReport.ClientAssocEvent.getDefaultInstance().toBuilder();
         clientAssocBuilder.setAssocType(AssocType.ASSOC);
         clientAssocBuilder.setBand(RadioBandType.BAND5GU);
         clientAssocBuilder.setRssi(-65);
@@ -837,29 +833,26 @@ public class OpensyncExternalIntegrationCloudTest {
 
         eventReportList.add(eventReportBuilder.build());
 
-
         return eventReportList;
 
-
     }
 
     private List<ClientReport> getOpensyncStatsClientReportsList() {
         int rssi = Long.valueOf(4294967239L).intValue();
 
-        Client.Stats clientStats = Client.Stats.getDefaultInstance().toBuilder().setRssi(rssi).setRxBytes(225554786)
-                .setRxRate(24000.0).setTxBytes(1208133026).setTxRate(433300.0).setRssi(758722570).setRxFrames(10000)
-                .setTxFrames(10000).setTxRate(24000.0).build();
-        Client client2g = Client.getDefaultInstance().toBuilder().setMacAddress("7C:AB:60:E6:EA:4D").setSsid("ssid-1")
-                .setConnected(true).setDurationMs(59977).setStats(clientStats).build();
-        Client client5gu = Client.getDefaultInstance().toBuilder().setMacAddress("C0:9A:D0:76:A9:69").setSsid("ssid-3")
-                .setConnected(true).setDurationMs(298127).setStats(clientStats).build();
+        Client.Stats clientStats = Client.Stats.getDefaultInstance().toBuilder().setRssi(rssi).setRxBytes(225554786).setRxRate(24000.0).setTxBytes(1208133026)
+                .setTxRate(433300.0).setRssi(758722570).setRxFrames(10000).setTxFrames(10000).setTxRate(24000.0).build();
+        Client client2g = Client.getDefaultInstance().toBuilder().setMacAddress("7C:AB:60:E6:EA:4D").setSsid("ssid-1").setConnected(true).setDurationMs(59977)
+                .setStats(clientStats).build();
+        Client client5gu = Client.getDefaultInstance().toBuilder().setMacAddress("C0:9A:D0:76:A9:69").setSsid("ssid-3").setConnected(true).setDurationMs(298127)
+                .setStats(clientStats).build();
 
-        ClientReport clientReport2g = ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND2G)
-                .setChannel(6).addAllClientList(ImmutableList.of(client2g)).build();
-        ClientReport clientReport5gl = ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND5GL)
-                .setChannel(36).addAllClientList(new ArrayList<Client>()).build();
-        ClientReport clientReport5gu = ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND5GU)
-                .setChannel(157).addAllClientList(ImmutableList.of(client5gu)).build();
+        ClientReport clientReport2g =
+                ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND2G).setChannel(6).addAllClientList(ImmutableList.of(client2g)).build();
+        ClientReport clientReport5gl =
+                ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND5GL).setChannel(36).addAllClientList(new ArrayList<Client>()).build();
+        ClientReport clientReport5gu = ClientReport.getDefaultInstance().toBuilder().setBand(RadioBandType.BAND5GU).setChannel(157)
+                .addAllClientList(ImmutableList.of(client5gu)).build();
 
         List<ClientReport> clients = new ArrayList<>();
         clients.add(clientReport2g);
@@ -888,8 +881,7 @@ public class OpensyncExternalIntegrationCloudTest {
         connectNodeInfo.skuNumber = "connectus.ai_21P10C68818122";
         connectNodeInfo.redirectorAddr = "ssl:opensync.zone1.art2wave.com:6643";
         connectNodeInfo.platformVersion = "OPENWRT_EA8300";
-        connectNodeInfo.wifiRadioStates = ImmutableMap.of("2.4G", "home-ap-24", "5GL", "home-ap-l50", "5GU",
-                "home-ap-u50");
+        connectNodeInfo.wifiRadioStates = ImmutableMap.of("2.4G", "home-ap-24", "5GL", "home-ap-l50", "5GU", "home-ap-u50");
 
         Map<String, String> versionMatrix = new HashMap<>();
         versionMatrix.put("DATE", "Thu Jul 16 18:52:06 UTC 2020");

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

@@ -116,7 +116,7 @@ public class MqttStatsPublisherTest {
     static class Config {
 
         @Bean
-        public MqttStatsPublisher mqttStatsPublisher() {
+        public StatsPublisherInterface mqttStatsPublisher() {
             return new MqttStatsPublisher();
         }
 
@@ -125,7 +125,7 @@ public class MqttStatsPublisherTest {
     @Before
     public void setUp() throws Exception {
 
-        mockito = Mockito.mockitoSession().initMocks(this).strictness(Strictness.STRICT_STUBS).startMocking();
+        mockito = Mockito.mockitoSession().initMocks(this).strictness(Strictness.LENIENT).startMocking();
 
     }
 
@@ -148,8 +148,12 @@ public class MqttStatsPublisherTest {
     public void testExtractCustomerIdFromTopic() {
         String topic = "/ap/Test_Client_21P10C68818122/opensync";
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
-        Mockito.when(session.getCustomerId()).thenReturn(2);
+        Equipment ce = Mockito.mock(Equipment.class);
+        Mockito.when(ce.getCustomerId()).thenReturn(2);
 
+        Mockito.when(
+                equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
+                .thenReturn(ce);
         Mockito.when(ovsdbSessionMapInterface.getSession("Test_Client_21P10C68818122")).thenReturn(session);
 
         assertEquals(2, opensyncExternalIntegrationMqttProcessor.extractCustomerIdFromTopic(topic));
@@ -173,8 +177,7 @@ public class MqttStatsPublisherTest {
         Equipment equipment = new Equipment();
 
         equipment.setDetails(ApElementConfiguration.createWithDefaults());
-
-        equipment.setId(1L);
+        equipment.setId(1L); equipment.setCustomerId(2);
 
         Mockito.when(
                 equipmentServiceInterface.getByInventoryIdOrNull(ArgumentMatchers.eq("Test_Client_21P10C68818122")))
@@ -193,7 +196,6 @@ public class MqttStatsPublisherTest {
 
         OvsdbSession session = Mockito.mock(OvsdbSession.class);
         Mockito.when(session.getEquipmentId()).thenReturn(1L);
-        Mockito.when(session.getCustomerId()).thenReturn(2);
 
         Mockito.when(ovsdbSessionMapInterface.getSession("Test_Client_21P10C68818122")).thenReturn(session);
 

opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationInterface.java

@@ -36,8 +36,6 @@ public interface OpensyncExternalIntegrationInterface {
 
     void wifiInetStateDbTableDelete(List<OpensyncAPInetState> inetStateTables, String apId);
 
-    void processMqttMessage(String topic, Report report);
-
     void wifiAssociatedClientsDbTableUpdate(List<OpensyncWifiAssociatedClients> wifiAssociatedClients, String apId);
 
     void wifiAssociatedClientsDbTableDelete(String deletedClientMac, String apId);
@@ -52,6 +50,8 @@ public interface OpensyncExternalIntegrationInterface {
 
     void apcStateDbTableUpdate(Map<String, String> apcStateAttributes, String apId,
             RowUpdateOperation rowUpdateOperation);
+    
+    void nodeStateDbTableUpdate(List<Map<String, String>> nodeStateAttributes, String apId);
 
     void clearEquipmentStatus(String apId);
 }

opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OvsdbSession.java

@@ -1,5 +1,7 @@
 package com.telecominfraproject.wlan.opensync.external.integration;
 
+import java.util.Objects;
+
 import com.vmware.ovsdb.service.OvsdbClient;
 
 public class OvsdbSession {
@@ -7,7 +9,7 @@ public class OvsdbSession {
     private String apId;
     private long routingId;
     private long equipmentId;
-    private int customerId;
+    private long mostRecentStatsTimestamp;
     
     public OvsdbClient getOvsdbClient() {
         return ovsdbClient;
@@ -33,12 +35,34 @@ public class OvsdbSession {
     public void setEquipmentId(long equipmentId) {
         this.equipmentId = equipmentId;
     }
-    public int getCustomerId() {
-        return customerId;
+    public long getMostRecentStatsTimestamp() {
+        return mostRecentStatsTimestamp;
     }
-    public void setCustomerId(int customerId) {
-        this.customerId = customerId;
+    public void setMostRecentStatsTimestamp(long mostRecentStatsTimestamp) {
+        this.mostRecentStatsTimestamp = mostRecentStatsTimestamp;
+    }
+    @Override
+    public int hashCode() {
+        return Objects.hash(apId, equipmentId, mostRecentStatsTimestamp, ovsdbClient, routingId);
+    }
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        OvsdbSession other = (OvsdbSession) obj;
+        return Objects.equals(apId, other.apId) && equipmentId == other.equipmentId && mostRecentStatsTimestamp == other.mostRecentStatsTimestamp
+                && Objects.equals(ovsdbClient, other.ovsdbClient) && routingId == other.routingId;
     }
     
+    @Override
+    public String toString() {
+        return "OvsdbSession [ovsdbClient=" + ovsdbClient + ", apId=" + apId + ", routingId=" + routingId + ", equipmentId=" + equipmentId
+                + ", mostRecentStatsTimestamp=" + mostRecentStatsTimestamp + "]";
+    }
+
     
 }

opensync-ext-interface/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/utils/StatsPublisherInterface.java

@@ -0,0 +1,14 @@
+
+package com.telecominfraproject.wlan.opensync.external.integration.utils;
+
+import com.telecominfraproject.wlan.systemevent.models.SystemEvent;
+
+import sts.OpensyncStats.Report;
+
+public interface StatsPublisherInterface {
+
+    void processMqttMessage(String topic, Report report);
+
+    void publishSystemEventFromTableStateMonitor(SystemEvent event);
+
+}

opensync-ext-static/src/main/java/com/telecominfraproject/wlan/opensync/external/integration/OpensyncExternalIntegrationSimple.java

@@ -207,4 +207,9 @@ public class OpensyncExternalIntegrationSimple implements OpensyncExternalIntegr
 
     }
 
+    @Override
+    public void nodeStateDbTableUpdate(List<Map<String, String>> nodeStateAttributes, String apId) {
+        LOG.info("nodeStateDbTableUpdate for AP {}", apId);
+    }
+
 }

opensync-gateway-cloud-docker/src/main/docker/app/opensync/logback.xml

@@ -30,6 +30,22 @@
     </triggeringPolicy>
   </appender>
   
+    <appender name="mqttMsgTracer" class="ch.qos.logback.core.rolling.RollingFileAppender">
+    <file>/app/logs/mqttMsgTracer.log</file>
+    <append>true</append>
+    <encoder>
+      <pattern>%date %msg%n</pattern>
+    </encoder>
+    <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+      <fileNamePattern>/app/logs/mqttMsgTracer.%i.log.gz</fileNamePattern>
+      <minIndex>1</minIndex>
+      <maxIndex>3</maxIndex>
+    </rollingPolicy>
+    <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+      <maxFileSize>20MB</maxFileSize>
+    </triggeringPolicy>
+  </appender>
+  
   <appender name="logfile" class="ch.qos.logback.core.rolling.RollingFileAppender">
     <file>/app/logs/opensyncgw.log</file>
     <append>true</append>
@@ -67,7 +83,7 @@
   <logger name="com.vmware.ovsdb.service.OvsdbConnectionInfo" level="OFF"/>
   <logger name="com.vmware.ovsdb.netty.OvsdbConnectionHandler" level="ERROR"/>
 
-  <logger name="MQTT_DATA" level="DEBUG" additivity="false">
+  <logger name="MQTT_DATA" level="INFO" additivity="false">
     <appender-ref ref="mqttDataFile"/>
   </logger>
 

opensync-gateway-cloud-process/src/main/resources/logback.xml

@@ -63,7 +63,7 @@
   <logger name="com.vmware.ovsdb.service.OvsdbConnectionInfo" level="OFF"/>
   <logger name="com.vmware.ovsdb.netty.OvsdbConnectionHandler" level="ERROR"/>
   
-  <logger name="MQTT_DATA" level="DEBUG"/>
+  <logger name="MQTT_DATA" level="INFO"/>
 
   <!--   
   <logger name="org.springframework.security.web.authentication.preauth" level="DEBUG"/>

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/mqtt/OpensyncMqttClient.java

@@ -25,9 +25,8 @@ import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.stereotype.Component;
 
 import com.google.protobuf.Descriptors;
-import com.google.protobuf.MessageOrBuilder;
+import com.google.protobuf.TypeRegistry;
 import com.google.protobuf.util.JsonFormat;
-import com.google.protobuf.util.JsonFormat.TypeRegistry;
 import com.netflix.servo.DefaultMonitorRegistry;
 import com.netflix.servo.monitor.BasicCounter;
 import com.netflix.servo.monitor.BasicTimer;
@@ -37,7 +36,9 @@ import com.netflix.servo.monitor.Stopwatch;
 import com.netflix.servo.monitor.Timer;
 import com.netflix.servo.tag.TagList;
 import com.telecominfraproject.wlan.cloudmetrics.CloudMetricsTags;
-import com.telecominfraproject.wlan.opensync.external.integration.OpensyncExternalIntegrationInterface;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSession;
+import com.telecominfraproject.wlan.opensync.external.integration.OvsdbSessionMapInterface;
+import com.telecominfraproject.wlan.opensync.external.integration.utils.StatsPublisherInterface;
 import com.telecominfraproject.wlan.opensync.util.ZlibUtil;
 
 import sts.OpensyncStats;
@@ -62,7 +63,10 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
     private final Timer timerMessageProcess = new BasicTimer(MonitorConfig.builder("osgw-mqtt-messageProcessTimer").withTags(tags).build());
 
     @Autowired
-    private OpensyncExternalIntegrationInterface extIntegrationInterface;
+    private StatsPublisherInterface statsPublisher;
+    
+    @Autowired
+    private OvsdbSessionMapInterface ovsdbSessionMapInterface;
 
     // dtop: use anonymous constructor to ensure that the following code always
     // get executed,
@@ -112,7 +116,7 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
                 while (keepReconnecting) {
                     BlockingConnection blockingConnection = null;
                     try {
-                        Thread.sleep(5000);
+                        // Thread.sleep(5000);
 
                         // Create a new MQTT connection to the broker.
                         /*
@@ -145,51 +149,13 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
                         mqtt.setClientId("opensync_mqtt");
                         mqtt.setUserName(username);
                         mqtt.setPassword(password);
-                        mqtt.setTracer(new Tracer() {
-                            @Override
-                            public void onReceive(MQTTFrame frame) {
-                                switch (frame.messageType()) {
-                                    case PINGREQ.TYPE:
-                                    case PINGRESP.TYPE:
-                                        // PINGs don't want to fill log
-                                        LOG.trace("MQTT Client Received: {}", frame);
-                                        break;
-                                    default:
-                                        LOG.debug("MQTT Client Received: {}", frame);
-                                }
-                            }
-
-                            @Override
-                            public void onSend(MQTTFrame frame) {
-                                switch (frame.messageType()) {
-                                    case PINGREQ.TYPE:
-                                    case PINGRESP.TYPE:
-                                        // PINGs don't want to fill log
-                                        LOG.trace("MQTT Client Sent: {}", frame);
-                                        break;
-                                    default:
-                                        LOG.debug("MQTT Client Sent: {}", frame);
-                                }
-                            }
-
-                            @Override
-                            public void debug(String message, Object... args) {
-                                LOG.debug("MQTT Client debugging messages: {}", String.format(message, args));
-                            }
-                        });
-
                         blockingConnection = mqtt.blockingConnection();
                         blockingConnection.connect();
 
-                        LOG.info("Connected to MQTT broker at {}", mqtt.getHost());
+                        LOG.debug("Connected to MQTT broker at {}", mqtt.getHost());
 
-                        // Subscribe to topics:
-                        //
-                        // new Topic("mqtt/example/publish", QoS.AT_LEAST_ONCE),
-                        // new Topic("#", QoS.AT_LEAST_ONCE),
-                        // new Topic("test/#", QoS.EXACTLY_ONCE),
-                        // new Topic("foo/+/bar", QoS.AT_LEAST_ONCE)
-                        Topic[] topics = {new Topic("#", QoS.AT_LEAST_ONCE),};
+                        // NB. setting to AT_MOST_ONCE to match the APs message level
+                        Topic[] topics = {new Topic("/ap/#", QoS.AT_MOST_ONCE),};
 
                         blockingConnection.subscribe(topics);
                         LOG.info("Subscribed to mqtt topics {}", Arrays.asList(topics));
@@ -204,40 +170,45 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
 
                         // main loop - receive messages
                         while (true) {
+                            LOG.debug("{} awaiting mqtt message from broker",mqttClientThread.getName());
                             Message mqttMsg = blockingConnection.receive();
 
                             if (mqttMsg == null) {
+                                if (LOG.isTraceEnabled())
+                                    LOG.trace("NULL message received for blocking connection");
                                 continue;
                             }
-
-                            LOG.debug("MQTT Topic {}", mqttMsg.getTopic());
-
-                            byte payload[] = mqttMsg.getPayload();
-
-                            messagesReceived.increment();
-                            messageBytesReceived.increment(payload.length);
                             Stopwatch stopwatchTimerMessageProcess = timerMessageProcess.start();
-
-                            LOG.trace("received message on topic {} size {}", mqttMsg.getTopic(), payload.length);
-
-                            if (payload[0] == 0x78) {
-                                // looks like zlib-compressed data, let's
-                                // decompress
-                                // it before deserializing
-                                payload = ZlibUtil.decompress(payload);
-                            }
-
-                            // attempt to parse the message as protobuf
-                            MessageOrBuilder encodedMsg = null;
                             try {
+                                byte payload[] = mqttMsg.getPayload();
+                                messagesReceived.increment();
+                                messageBytesReceived.increment(payload.length);
+
+                                if (payload[0] == 0x78) {
+                                    // looks like zlib-compressed data, let's
+                                    // decompress
+                                    // it before deserializing
+                                    payload = ZlibUtil.decompress(payload);
+                                }
                                 // Only supported protobuf on the TIP opensync APs is Report
-                                encodedMsg = Report.parseFrom(payload);
+                                Report statsReport = Report.parseFrom(payload);
                                 mqttMsg.ack();
-                                MQTT_LOG.info("topic = {}\nReport = {}", mqttMsg.getTopic(), jsonPrinter.print(encodedMsg));
-                                extIntegrationInterface.processMqttMessage(mqttMsg.getTopic(), (Report) encodedMsg);
+                                String apId = extractApIdFromTopic(mqttMsg.getTopic());
+                                if (apId != null) {
+                                    OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(extractApIdFromTopic(mqttMsg.getTopic()));
+                                    if (ovsdbSession != null) {
+                                        ovsdbSession.setMostRecentStatsTimestamp(System.currentTimeMillis());
+                                        LOG.debug("Last metrics received from AP updated to {}",ovsdbSession.toString());
+                                    } else {
+                                        LOG.debug("No ovsdb session exists for this AP {}",apId);
+                                    }
+                                }
+                                MQTT_LOG.info("Topic {}\n{}", mqttMsg.getTopic(), jsonPrinter.print(statsReport));
+                                statsPublisher.processMqttMessage(mqttMsg.getTopic(), statsReport);                           
+                                LOG.debug("Dispatched report for topic {} to backend for processing", mqttMsg.getTopic());
+
                             } catch (Exception e) {
-                                String msgStr = new String(mqttMsg.getPayload(), utf8);
-                                LOG.warn("Could not process message topic = {}\nmessage = {}", mqttMsg.getTopic(), msgStr);
+                                LOG.error("Exception processing topic for message {}",mqttMsg, e);
                             } finally {
                                 stopwatchTimerMessageProcess.stop();
                             }
@@ -273,4 +244,26 @@ public class OpensyncMqttClient implements ApplicationListener<ContextClosedEven
             mqttClientThread.interrupt();
         }
     }
+    
+    /**
+     * @param topic
+     * @return apId extracted from the topic name, or null if it cannot be
+     *         extracted
+     */
+    static String extractApIdFromTopic(String topic) {
+        // Topic is formatted as
+        // "/ap/"+clientCn+"_"+ret.serialNumber+"/opensync"
+        if (topic == null) {
+            return null;
+        }
+
+        String[] parts = topic.split("/");
+        if (parts.length < 3) {
+            return null;
+        }
+
+        // apId is the third element in the topic
+        return parts[2];
+    }
+    
 }

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/TipWlanOvsdbClient.java

@@ -28,6 +28,7 @@ import com.vmware.ovsdb.callback.ConnectionCallback;
 import com.vmware.ovsdb.exception.OvsdbClientException;
 import com.vmware.ovsdb.protocol.methods.*;
 import com.vmware.ovsdb.protocol.operation.notation.Row;
+import com.vmware.ovsdb.protocol.operation.notation.Value;
 import com.vmware.ovsdb.service.OvsdbClient;
 import com.vmware.ovsdb.service.OvsdbPassiveConnectionListener;
 import io.netty.handler.ssl.SslContext;
@@ -259,6 +260,7 @@ public class TipWlanOvsdbClient implements OvsdbClientInterface {
             if (ovsdbDao.getDeviceStatsReportingInterval(ovsdbClient) != collectionIntervalSecDeviceStats) {
                 ovsdbDao.updateDeviceStatsReportingInterval(ovsdbClient, collectionIntervalSecDeviceStats);
             }
+            ovsdbDao.enableNetworkProbeForSyntheticClient(ovsdbClient);
             ovsdbDao.updateEventReportingInterval(ovsdbClient, collectionIntervalSecEvent);
 
         } else {
@@ -287,7 +289,7 @@ public class TipWlanOvsdbClient implements OvsdbClientInterface {
 
         return ovsdbDao.changeRedirectorAddress(ovsdbSession.getOvsdbClient(), apId, newRedirectorAddress);
     }
-    
+
     public String processBlinkRequest(String apId, boolean blinkAllLEDs) {
         OvsdbSession ovsdbSession = ovsdbSessionMapInterface.getSession(apId);
         if (ovsdbSession == null) {
@@ -716,20 +718,35 @@ public class TipWlanOvsdbClient implements OvsdbClientInterface {
         CompletableFuture<TableUpdates> nsCf = ovsdbClient.monitor(OvsdbDao.ovsdbName, OvsdbDao.nodeStateTable + "_" + key,
                 new MonitorRequests(ImmutableMap.of(OvsdbDao.nodeStateTable, new MonitorRequest(new MonitorSelect(true, true, true, true)))), tableUpdates -> {
                     LOG.info(OvsdbDao.nodeStateTable + "_" + key + " monitor callback received {}");
-                    tableUpdates.getTableUpdates().forEach((key1, value) -> {
-                        LOG.info("TableUpdate for {}", key1);
-                        value.getRowUpdates().values().forEach(r -> {
-                            if (r.getOld() != null) {
-                                LOG.info("Node_State old row {}", r.getOld().getColumns());
-                            }
-                            if (r.getNew() != null) {
-                                LOG.info("Node_State new row {}", r.getNew().getColumns());
-                            }
-                        });
-
-                    });
+                    extIntegrationInterface.nodeStateDbTableUpdate(processNodeStateTableUpdate(key, tableUpdates), key);
+                    
                 });
-        nsCf.join();
+        extIntegrationInterface.nodeStateDbTableUpdate(processNodeStateTableUpdate(key, nsCf.join()), key);
+    }
+
+    private List<Map<String,String>> processNodeStateTableUpdate(String key, TableUpdates tableUpdates) {
+        List<Map<String, String>> updates = new ArrayList<>();
+        tableUpdates.getTableUpdates().forEach((key1, value) -> {
+            LOG.info("TableUpdate for {}", key1);
+            value.getRowUpdates().values().forEach(r -> {
+                if (r.getOld() != null) {
+                    LOG.info("Node_State old row {}", r.getOld().getColumns());
+                }
+                if (r.getNew() != null) {
+                    LOG.info("Node_State new row {}", r.getNew().getColumns());
+                    Map<String, Value> columns = r.getNew().getColumns();
+                    Map<String, String> update = new HashMap<>();
+                    update.put("key", columns.get("key").toString());
+                    update.put("module", columns.get("module").toString());
+                    update.put("persist", columns.get("persist").toString());
+                    update.put("value", columns.get("value").toString());
+                    updates.add(update);
+                }
+            });
+
+        });
+        
+        return updates;
     }
 
     private void monitorAPCStateTable(OvsdbClient ovsdbClient, String key) throws OvsdbClientException {

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/dao/OvsdbDao.java

@@ -9,6 +9,7 @@ import org.springframework.stereotype.Component;
 
 import com.telecominfraproject.wlan.core.model.equipment.MacAddress;
 import com.telecominfraproject.wlan.core.model.equipment.RadioType;
+import com.telecominfraproject.wlan.equipment.models.ApElementConfiguration;
 import com.telecominfraproject.wlan.equipment.models.CellSizeAttributes;
 import com.telecominfraproject.wlan.opensync.external.integration.models.ConnectNodeInfo;
 import com.telecominfraproject.wlan.opensync.external.integration.models.OpensyncAPConfig;
@@ -86,7 +87,10 @@ public class OvsdbDao extends OvsdbDaoBase {
     }
     
     public void configureNode(OvsdbClient ovsdbClient, OpensyncAPConfig opensyncAPConfig) {
-        configureNtpServer(ovsdbClient,opensyncAPConfig);configureSyslog(ovsdbClient, opensyncAPConfig);
+        configureNtpServer(ovsdbClient, opensyncAPConfig);
+        configureSyslog(ovsdbClient, opensyncAPConfig);
+        processBlinkRequest(ovsdbClient, opensyncAPConfig.getCustomerEquipment().getInventoryId(),
+                ((ApElementConfiguration) opensyncAPConfig.getCustomerEquipment().getDetails()).isBlinkAllLEDs());
     }
     
     void configureNtpServer(OvsdbClient ovsdbClient, OpensyncAPConfig opensyncAPConfig) {

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/dao/OvsdbRadioConfig.java

@@ -14,6 +14,7 @@ import com.vmware.ovsdb.protocol.operation.Operation;
 import com.vmware.ovsdb.protocol.operation.Update;
 import com.vmware.ovsdb.protocol.operation.notation.*;
 import com.vmware.ovsdb.protocol.operation.result.OperationResult;
+import com.vmware.ovsdb.protocol.schema.DatabaseSchema;
 import com.vmware.ovsdb.service.OvsdbClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -39,88 +40,100 @@ public class OvsdbRadioConfig extends OvsdbDaoBase {
         Map<String, WifiRadioConfigInfo> provisionedRadioConfigs = ovsdbGet.getProvisionedWifiRadioConfigs(ovsdbClient);
         Map<String, WifiVifConfigInfo> vifConfigs = ovsdbGet.getProvisionedWifiVifConfigs(ovsdbClient);
         List<Operation> operations = new ArrayList<>();
-        for (RadioType radioType : apElementConfiguration.getRadioMap().keySet()) {
-            Map<String, String> hwConfig = new HashMap<>();
-            ElementRadioConfiguration elementRadioConfig = apElementConfiguration.getRadioMap().get(radioType);
-            RfElementConfiguration rfElementConfig = rfConfig.getRfConfig(radioType);
-            if (elementRadioConfig == null || rfElementConfig == null) {
-                continue;
-            }
-            boolean autoChannelSelection = rfElementConfig.getAutoChannelSelection();
-            int channel = elementRadioConfig.getActiveChannel(autoChannelSelection);
-            LOG.debug("configureWifiRadios autoChannelSelection {} activeChannel {} getChannelNumber {} ", autoChannelSelection, channel,
-                    elementRadioConfig.getChannelNumber());
-            ChannelBandwidth bandwidth = rfElementConfig.getChannelBandwidth();
-            String ht_mode = getBandwidth(bandwidth);
-            RadioConfiguration radioConfig = apElementConfiguration.getAdvancedRadioMap().get(radioType);
-            int beaconInterval = rfElementConfig.getBeaconInterval();
-            boolean enabled = radioConfig.getRadioAdminState().equals(StateSetting.enabled);
-            boolean autoCellSizeSelection = rfElementConfig.getAutoCellSizeSelection();
-            int txPower = 0;
-            if (autoCellSizeSelection) {
-                if (elementRadioConfig.getEirpTxPower() != null) {
-                    txPower = elementRadioConfig.getEirpTxPower().getValue();
-                }
-            } else {
-                txPower = rfElementConfig.getEirpTxPower();
-            }
-            String hwMode = getHwMode(rfElementConfig);
-            String freqBand = getHwConfigAndFreq(radioType, hwConfig);
-            String radioName = null;
-            for (String key : provisionedRadioConfigs.keySet()) {
-                if (provisionedRadioConfigs.get(key).freqBand.equals(freqBand)) {
-                    radioName = key;
-                    break;
-                }
-            }
-            if (radioName == null)
-                continue;
-            String ifName = null; // for vifConfigs
-            if (radioName.equals(radio0)) {
-                ifName = defaultRadio0;
-            } else if (radioName.equals(radio1)) {
-                ifName = defaultRadio1;
-            } else if (radioName.equals(radio2)) {
-                ifName = defaultRadio2;
-            }
-            if (ifName == null)
-                continue;
-            Set<Uuid> vifUuidsForRadio = new HashSet<>();
-            for (String key : vifConfigs.keySet()) {
-                if (key.contains(ifName))
-                    vifUuidsForRadio.add(vifConfigs.get(key).uuid);
-            }
-            int mimoMode = MimoMode.none.getId();
-            if (rfElementConfig.getMimoMode() != null) {
-                mimoMode = rfElementConfig.getMimoMode().getId();
-            }
-            int maxNumClients = 0;
-            if (rfElementConfig.getMaxNumClients() != null) {
-                maxNumClients = rfElementConfig.getMaxNumClients();
-            }
-            try {
-                configureWifiRadios(freqBand, channel, hwConfig, country.toUpperCase(), beaconInterval, enabled, hwMode, ht_mode, txPower, mimoMode,
-                        vifUuidsForRadio, operations, maxNumClients);
-            } catch (OvsdbClientException e) {
-                LOG.error("ConfigureWifiRadios failed with OvsdbClient exception.", e);
-                throw new RuntimeException(e);
-            } catch (TimeoutException e) {
-                LOG.error("ConfigureWifiRadios failed with Timeout.", e);
-                throw new RuntimeException(e);
-            } catch (ExecutionException e) {
-                LOG.error("ConfigureWifiRadios excecution failed.", e);
-                throw new RuntimeException(e);
-            } catch (InterruptedException e) {
-                LOG.error("ConfigureWifiRadios interrupted.", e);
-                throw new RuntimeException(e);
-            }
-        }
         try {
+            CompletableFuture<DatabaseSchema> cfDatabaseSchema = ovsdbClient.getSchema(ovsdbName);
+            DatabaseSchema databaseSchema = cfDatabaseSchema.get();
+            Set<String> columnNames = databaseSchema.getTables().get(wifiRadioConfigDbTable).getColumns().keySet();
+ 
+            for (RadioType radioType : apElementConfiguration.getRadioMap().keySet()) {
+                Map<String, String> hwConfig = new HashMap<>();
+                ElementRadioConfiguration elementRadioConfig = apElementConfiguration.getRadioMap().get(radioType);
+                RfElementConfiguration rfElementConfig = rfConfig.getRfConfig(radioType);
+                if (elementRadioConfig == null || rfElementConfig == null) {
+                    continue;
+                }
+                boolean autoChannelSelection = rfElementConfig.getAutoChannelSelection();
+                int channel = elementRadioConfig.getActiveChannel(autoChannelSelection);
+                ChannelBandwidth bandwidth = rfElementConfig.getChannelBandwidth();
+                String ht_mode = getBandwidth(bandwidth);
+                RadioConfiguration radioConfig = apElementConfiguration.getAdvancedRadioMap().get(radioType);
+                int beaconInterval = rfElementConfig.getBeaconInterval();
+                boolean enabled = radioConfig.getRadioAdminState().equals(StateSetting.enabled);
+                boolean autoCellSizeSelection = rfElementConfig.getAutoCellSizeSelection();
+                int txPower = 0;
+                if (autoCellSizeSelection) {
+                    if (elementRadioConfig.getEirpTxPower() != null) {
+                        txPower = elementRadioConfig.getEirpTxPower().getValue();
+                    }
+                } else {
+                    txPower = rfElementConfig.getEirpTxPower();
+                }
+                String hwMode = getHwMode(rfElementConfig);
+                String freqBand = getHwConfigAndFreq(radioType, hwConfig);
+                String radioName = null;
+                for (String key : provisionedRadioConfigs.keySet()) {
+                    if (provisionedRadioConfigs.get(key).freqBand.equals(freqBand)) {
+                        radioName = key;
+                        break;
+                    }
+                }
+                LOG.debug(
+                        "configureWifiRadios for AP {} autoChannelSelection={} activeChannel={} getChannelNumber={} radioName={} freqBand={} hwMode={} txPower={} enabled={} adminStateSetting={} autoCellSizeSelection={}",
+                        opensyncAPConfig.getCustomerEquipment().getInventoryId(), autoChannelSelection, channel, elementRadioConfig.getChannelNumber(), radioName, freqBand, hwMode, txPower,
+                        enabled, radioConfig.getRadioAdminState(), autoCellSizeSelection);
+                if (radioName == null)
+                    continue;
+                String ifName = null; // for vifConfigs
+                if (radioName.equals(radio0)) {
+                    ifName = defaultRadio0;
+                } else if (radioName.equals(radio1)) {
+                    ifName = defaultRadio1;
+                } else if (radioName.equals(radio2)) {
+                    ifName = defaultRadio2;
+                }
+                if (ifName == null)
+                    continue;
+                Set<Uuid> vifUuidsForRadio = new HashSet<>();
+                for (String key : vifConfigs.keySet()) {
+                    if (key.contains(ifName))
+                        vifUuidsForRadio.add(vifConfigs.get(key).uuid);
+                }
+                int mimoMode = MimoMode.none.getId();
+                if (rfElementConfig.getMimoMode() != null) {
+                    mimoMode = rfElementConfig.getMimoMode().getId();
+                }
+                int maxNumClients = 0;
+                if (rfElementConfig.getMaxNumClients() != null) {
+                    maxNumClients = rfElementConfig.getMaxNumClients();
+                }
+                try {
+                    configureWifiRadios(freqBand, channel, hwConfig, country.toUpperCase(), beaconInterval, enabled, hwMode, ht_mode, txPower, mimoMode,
+                            vifUuidsForRadio, operations, maxNumClients,columnNames);
+                } catch (OvsdbClientException e) {
+                    LOG.error("ConfigureWifiRadios failed with OvsdbClient exception.", e);
+                    throw new RuntimeException(e);
+                } catch (TimeoutException e) {
+                    LOG.error("ConfigureWifiRadios failed with Timeout.", e);
+                    throw new RuntimeException(e);
+                } catch (ExecutionException e) {
+                    LOG.error("ConfigureWifiRadios excecution failed.", e);
+                    throw new RuntimeException(e);
+                } catch (InterruptedException e) {
+                    LOG.error("ConfigureWifiRadios interrupted.", e);
+                    throw new RuntimeException(e);
+                }
+            }
+
             CompletableFuture<OperationResult[]> fResult = ovsdbClient.transact(ovsdbName, operations);
             OperationResult[] result = fResult.get(ovsdbTimeoutSec, TimeUnit.SECONDS);
             for (OperationResult res : result) {
-                LOG.debug("Op Result {}", res);
+                LOG.trace("configureWifiRadios Op Result {}", res);
             }
+            
+            ovsdbGet.getProvisionedWifiRadioConfigs(ovsdbClient).entrySet().forEach(r -> {
+                LOG.debug("configureWifiRadios for AP {} Wifi_Radio_Config for if_name:{}\n{}",  opensyncAPConfig.getCustomerEquipment().getInventoryId(), r.getKey(),r.getValue());
+            });
+
         } catch (OvsdbClientException | ExecutionException | InterruptedException | TimeoutException e) {
             LOG.error("configureWifiRadios error", e);
             throw new RuntimeException(e);
@@ -227,7 +240,7 @@ public class OvsdbRadioConfig extends OvsdbDaoBase {
     }
 
     void configureWifiRadios(String freqBand, int channel, Map<String, String> hwConfig, String country, int beaconInterval, boolean enabled, String hwMode,
-            String ht_mode, int txPower, int mimoMode, Set<Uuid> vifUuidsForRadio, List<Operation> operations, int maxNumClients)
+            String ht_mode, int txPower, int mimoMode, Set<Uuid> vifUuidsForRadio, List<Operation> operations, int maxNumClients, Set<String> tableColumns)
             throws OvsdbClientException, TimeoutException, ExecutionException, InterruptedException {
         Map<String, Value> updateColumns = new HashMap<>();
         List<Condition> conditions = new ArrayList<>();
@@ -253,7 +266,7 @@ public class OvsdbRadioConfig extends OvsdbDaoBase {
             updateColumns.put("hw_mode", new Atom<>(hwMode));
         }
         configureCustomOptionsMap(maxNumClients, updateColumns);
-        setTxAndRxChainmask(mimoMode, updateColumns);
+        setTxAndRxChainmask(mimoMode, updateColumns,tableColumns);
         if (vifUuidsForRadio.size() > 0) {
             com.vmware.ovsdb.protocol.operation.notation.Set vifConfigUuids = com.vmware.ovsdb.protocol.operation.notation.Set.of(vifUuidsForRadio);
             updateColumns.put("vif_configs", vifConfigUuids);
@@ -270,7 +283,7 @@ public class OvsdbRadioConfig extends OvsdbDaoBase {
         updateColumns.put("custom_options", customOptionsMap);
     }
 
-    void setTxAndRxChainmask(int mimoMode, Map<String, Value> updateColumns) {
+    void setTxAndRxChainmask(int mimoMode, Map<String, Value> updateColumns, Set<String> tableColumns) {
         /*
          * Chainmask is a bitmask, so map mimo mode values accordingly
          * Note values 0, 1 remain unchanged
@@ -314,7 +327,9 @@ public class OvsdbRadioConfig extends OvsdbDaoBase {
                 break;
         }
         updateColumns.put("tx_chainmask", new Atom<>(mimoMode));
-        updateColumns.put("rx_chainmask", new Atom<>(mimoMode));
+        if (tableColumns.contains("rx_chainmask")) {
+            updateColumns.put("rx_chainmask", new Atom<>(mimoMode));
+        }
     }
 
 }

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/dao/OvsdbRrmConfig.java

@@ -5,6 +5,7 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -41,6 +42,7 @@ import com.vmware.ovsdb.protocol.operation.result.ErrorResult;
 import com.vmware.ovsdb.protocol.operation.result.InsertResult;
 import com.vmware.ovsdb.protocol.operation.result.OperationResult;
 import com.vmware.ovsdb.protocol.operation.result.UpdateResult;
+import com.vmware.ovsdb.protocol.schema.DatabaseSchema;
 import com.vmware.ovsdb.service.OvsdbClient;
 
 @Component
@@ -49,6 +51,7 @@ public class OvsdbRrmConfig extends OvsdbDaoBase {
 
         ApElementConfiguration apElementConfig = (ApElementConfiguration) opensyncApConfig.getCustomerEquipment().getDetails();
         RfConfiguration rfConfig = (RfConfiguration) opensyncApConfig.getRfProfile().getDetails();
+
         for (RadioType radioType : apElementConfig.getRadioMap().keySet()) {
             String freqBand = null;
             if (radioType == RadioType.is2dot4GHz) {
@@ -138,60 +141,76 @@ public class OvsdbRrmConfig extends OvsdbDaoBase {
             int noiseFloorThresholdInDB, int noiseFloorThresholdTimeInSeconds, int nonWifiThresholdInPercentage, int nonWifiThresholdTimeInSeconds)
             throws OvsdbClientException, TimeoutException, ExecutionException, InterruptedException {
 
+        CompletableFuture<DatabaseSchema> cfDatabaseSchema = ovsdbClient.getSchema(ovsdbName);
+        DatabaseSchema databaseSchema = cfDatabaseSchema.get();
+        Set<String> columnNames = databaseSchema.getTables().get(wifiRrmConfigDbTable).getColumns().keySet();
+
         List<Operation> operations = new ArrayList<>();
         Map<String, Value> updateColumns = new HashMap<>();
 
-        updateColumns.put("freq_band", new Atom<>(freqBand));
-        updateColumns.put("backup_channel", new Atom<>(backupChannel));
-
-        if (multicastRate == null || multicastRate == MulticastRate.auto) {
-            updateColumns.put("mcast_rate", new Atom<>(0));
-        } else {
-            updateColumns.put("mcast_rate", new Atom<>(multicastRate.getId()));
+        if (columnNames.contains("freq_band")) updateColumns.put("freq_band", new Atom<>(freqBand));
+        if (columnNames.contains("backup_channel")) updateColumns.put("backup_channel", new Atom<>(backupChannel));
+        if (columnNames.contains("mcast_rate")) {
+            if (multicastRate == null || multicastRate == MulticastRate.auto) {
+                updateColumns.put("mcast_rate", new Atom<>(0));
+            } else {
+                updateColumns.put("mcast_rate", new Atom<>(multicastRate.getId()));
+            }
         }
 
-        if (probeResponseThreshold == null) {
-            updateColumns.put("probe_resp_threshold", new com.vmware.ovsdb.protocol.operation.notation.Set());
-        } else {
-            updateColumns.put("probe_resp_threshold", new Atom<>(probeResponseThreshold.intValue()));
+        if (columnNames.contains("probe_resp_threshold")) {
+            if (probeResponseThreshold == null) {
+                updateColumns.put("probe_resp_threshold", new com.vmware.ovsdb.protocol.operation.notation.Set());
+            } else {
+                updateColumns.put("probe_resp_threshold", new Atom<>(probeResponseThreshold.intValue()));
+            }
+        }
+        if (columnNames.contains("client_disconnect_threshold")) {
+            if (clientDisconnectThreshold == null) {
+                updateColumns.put("client_disconnect_threshold", new com.vmware.ovsdb.protocol.operation.notation.Set());
+            } else {
+                updateColumns.put("client_disconnect_threshold", new Atom<>(clientDisconnectThreshold.intValue()));
+            }
+        }
+        if (columnNames.contains("beacon_rate")) {
+            if (managementRate == null || managementRate == ManagementRate.auto) {
+                updateColumns.put("beacon_rate", new Atom<>(0));
+            } else {
+                updateColumns.put("beacon_rate", new Atom<>(managementRate.getId() * 10));
+            }
         }
 
-        if (clientDisconnectThreshold == null) {
-            updateColumns.put("client_disconnect_threshold", new com.vmware.ovsdb.protocol.operation.notation.Set());
-        } else {
-            updateColumns.put("client_disconnect_threshold", new Atom<>(clientDisconnectThreshold.intValue()));
-        }
-
-        if (managementRate == null || managementRate == ManagementRate.auto) {
-            updateColumns.put("beacon_rate", new Atom<>(0));
-        } else {
-            updateColumns.put("beacon_rate", new Atom<>(managementRate.getId() * 10));
-        }
-
-        if (bestApSettings == null) {
-            updateColumns.put("min_load", new com.vmware.ovsdb.protocol.operation.notation.Set());
-            updateColumns.put("snr_percentage_drop", new com.vmware.ovsdb.protocol.operation.notation.Set());
-        } else {
-            if (bestApSettings.getDropInSnrPercentage() == null) {
+        if (columnNames.contains("min_load") && columnNames.contains("snr_percentage_drop")) {
+            if (bestApSettings == null) {
+                updateColumns.put("min_load", new com.vmware.ovsdb.protocol.operation.notation.Set());
                 updateColumns.put("snr_percentage_drop", new com.vmware.ovsdb.protocol.operation.notation.Set());
             } else {
-                updateColumns.put("snr_percentage_drop", new Atom<>(bestApSettings.getDropInSnrPercentage()));
-            }
-            if (bestApSettings.getMinLoadFactor() == null) {
-                updateColumns.put("min_load", new com.vmware.ovsdb.protocol.operation.notation.Set());
-            } else {
-                updateColumns.put("min_load", new Atom<>(bestApSettings.getMinLoadFactor()));
+                if (bestApSettings.getDropInSnrPercentage() == null) {
+                    updateColumns.put("snr_percentage_drop", new com.vmware.ovsdb.protocol.operation.notation.Set());
+                } else {
+                    updateColumns.put("snr_percentage_drop", new Atom<>(bestApSettings.getDropInSnrPercentage()));
+                }
+                if (bestApSettings.getMinLoadFactor() == null) {
+                    updateColumns.put("min_load", new com.vmware.ovsdb.protocol.operation.notation.Set());
+                } else {
+                    updateColumns.put("min_load", new Atom<>(bestApSettings.getMinLoadFactor()));
+                }
             }
         }
 
-        if (obssHopMode != null) {
+        if (obssHopMode != null && columnNames.contains("obss_hop_mode"))
+
+        {
             updateColumns.put("obss_hop_mode", new Atom<>(obssHopMode.equals(OBSSHopMode.NON_WIFI) ? 1 : 2));
         }
-        updateColumns.put("noise_floor_thresh", new Atom<>(noiseFloorThresholdInDB));
-        updateColumns.put("noise_floor_time", new Atom<>(noiseFloorThresholdTimeInSeconds));
-        updateColumns.put("non_wifi_thresh", new Atom<>(nonWifiThresholdInPercentage));
-        updateColumns.put("non_wifi_time", new Atom<>(nonWifiThresholdTimeInSeconds));
-
+        if (columnNames.contains("noise_floor_thresh"))
+            updateColumns.put("noise_floor_thresh", new Atom<>(noiseFloorThresholdInDB));
+        if (columnNames.contains("noise_floor_time"))
+            updateColumns.put("noise_floor_time", new Atom<>(noiseFloorThresholdTimeInSeconds));
+        if (columnNames.contains("non_wifi_thresh"))
+            updateColumns.put("non_wifi_thresh", new Atom<>(nonWifiThresholdInPercentage));
+        if (columnNames.contains("non_wifi_time"))
+            updateColumns.put("non_wifi_time", new Atom<>(nonWifiThresholdTimeInSeconds));
 
         Row row = new Row(updateColumns);
         operations.add(new Insert(wifiRrmConfigDbTable, row));

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/dao/OvsdbSsidConfig.java

@@ -554,7 +554,7 @@ public class OvsdbSsidConfig extends OvsdbDaoBase {
                 }
                 boolean enableProxyArpForHotspot = false;
                 if (ssidConfig.getEnableProxyArpForHotspot() != null) {
-                    enableProxyArpForHotspot = false;
+                    enableProxyArpForHotspot = ssidConfig.getEnableProxyArpForHotspot();
                 }
                 try {
                     configureSingleSsid(ovsdbClient, ifName, ssidConfig.getSsid(), ssidBroadcast, security, vlanId, rrmEnabled, enable80211r, mobilityDomain,

opensync-gateway/src/main/java/com/telecominfraproject/wlan/opensync/ovsdb/dao/OvsdbStatsConfig.java

@@ -63,7 +63,6 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
 
             provisionWifiStatsConfigClient(radioConfigs, ovsdbGet.getProvisionedWifiStatsConfigs(ovsdbClient),
                     operations);
-
             if (!operations.isEmpty()) {
                 LOG.debug("Sending batch of operations : {} ", operations);
 
@@ -78,11 +77,7 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
                     }
                 }
             }
-
-            // TODO: when schema support is added, these should be part of the
-            // bulk provisioning operation above.
             provisionVideoVoiceStats(ovsdbClient);
-
         } catch (OvsdbClientException | TimeoutException | ExecutionException | InterruptedException e) {
             throw new RuntimeException(e);
         }
@@ -248,7 +243,6 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
                     rowColumns.put("report_type", new Atom<>("raw"));
                     rowColumns.put("sampling_interval", new Atom<>(10));
                     rowColumns.put("stats_type", new Atom<>("client"));
-                    rowColumns.put("survey_interval_ms", new Atom<>(65));
                     Row updateRow = new Row(rowColumns);
                     operations.add(new Insert(wifiStatsConfigDbTable, updateRow));
 
@@ -258,6 +252,7 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
 
     }
 
+
     void provisionWifiStatsConfigNeighbor(Map<String, Set<Integer>> allowedChannels,
             Map<String, WifiRadioConfigInfo> radioConfigs, Map<String, WifiStatsConfigInfo> provisionedWifiStatsConfigs,
             List<Operation> operations) {
@@ -280,6 +275,7 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
                     rowColumns.put("reporting_interval", new Atom<>(defaultOffChannelReportingIntervalSeconds));
                     rowColumns.put("stats_type", new Atom<>("neighbor"));
                     rowColumns.put("survey_type", new Atom<>("off-chan"));
+                    rowColumns.put("survey_interval_ms", new Atom<>(10));
 
                     Row updateRow = new Row(rowColumns);
                     operations.add(new Insert(wifiStatsConfigDbTable, updateRow));
@@ -299,6 +295,7 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
                     rowColumns.put("reporting_interval", new Atom<>(defaultReportingIntervalSeconds));
                     rowColumns.put("stats_type", new Atom<>("neighbor"));
                     rowColumns.put("survey_type", new Atom<>("on-chan"));
+                    rowColumns.put("survey_interval_ms", new Atom<>(0));
 
                     Row updateRow = new Row(rowColumns);
                     operations.add(new Insert(wifiStatsConfigDbTable, updateRow));
@@ -354,8 +351,8 @@ public class OvsdbStatsConfig extends OvsdbDaoBase {
                     rowColumns.put("report_type", new Atom<>("raw"));
                     rowColumns.put("stats_type", new Atom<>("survey"));
                     rowColumns.put("survey_type", new Atom<>("off-chan"));
-                    rowColumns.put("sampling_interval", new Atom<>(10));
-                    rowColumns.put("survey_interval_ms", new Atom<>(50));
+                    rowColumns.put("sampling_interval", new Atom<>(0));
+                    rowColumns.put("survey_interval_ms", new Atom<>(10));
                     rowColumns.put("threshold", thresholds);
                     Row updateRow = new Row(rowColumns);
                     operations.add(new Insert(wifiStatsConfigDbTable, updateRow));