implement persisting DeviceToClientMapRepresentation for [#381](#381)

Author: ck-c8y

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/core/client/AConnectorClient.java

@@ -321,7 +321,7 @@ public CompletableFuture<Void> submitDisconnect() {
     }
 
     public void submitHousekeeping() {
-        log.debug("{} - Starting housekeeping...", tenant);
+        log.debug("{} - Starting housekeeping for {} ...", tenant, connectorName);
         housekeepingExecutor.scheduleAtFixedRate(
                 this::runHousekeeping,
                 0,
@@ -750,9 +750,7 @@ private void performHousekeepingTasks() throws Exception {
         monitorSubscriptions();
     }
 
-    protected void connectorSpecificHousekeeping(String tenant) {
-       // Implement in specific connector if needed 
-    }
+    protected abstract void connectorSpecificHousekeeping(String tenant);
 
     private void logHousekeepingStatus(Instant now) {
         if (Duration.between(start, now).getSeconds() < 1800) {

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/http/HttpClient.java

@@ -231,6 +231,10 @@ public void onMessage(ConnectorMessage message) {
         dispatcher.onMessage(message);
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     @Override
     public List<Direction> supportedDirections() {
         return new ArrayList<>(Arrays.asList(Direction.INBOUND));

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/kafka/KafkaClient.java

@@ -392,6 +392,10 @@ public boolean isConfigValid(ConnectorConfiguration configuration) {
         return true;
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     @Override
     public void publishMEAO(ProcessingContext<?> context) {
         C8YRequest currentRequest = context.getCurrentRequest();

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/mqtt/MQTT3Client.java

@@ -563,6 +563,10 @@ public void monitorSubscriptions() {
         // nothing to do
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     @Override
     public List<Direction> supportedDirections() {
         return new ArrayList<>(Arrays.asList(Direction.INBOUND, Direction.OUTBOUND));

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/mqtt/MQTT5Client.java

@@ -563,6 +563,10 @@ public void monitorSubscriptions() {
         // nothing to do
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     @Override
     public List<Direction> supportedDirections() {
         return new ArrayList<>(Arrays.asList(Direction.INBOUND, Direction.OUTBOUND));

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/pulsar/MQTTServicePulsarClient.java

@@ -476,6 +476,7 @@ public void unsubscribe(String topic) throws Exception {
         log.info("{} - Unsubscription registered for topic: [{}]", tenant, topic);
     }
 
+
     @Override
     public void connectorSpecificHousekeeping(String tenant) {
         mappingService.sendDeviceToClientMap(tenant);

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/pulsar/PulsarConnectorClient.java

@@ -810,6 +810,10 @@ private void sendMessageWithQos(Producer<byte[]> producer, String payload, Qos q
         }
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     /**
      * Handles publish errors with connection recovery
      */

dynamic-mapper-service/src/main/java/dynamic/mapper/connector/webhook/WebHook.java

@@ -615,6 +615,10 @@ public JsonNode mergeNodes(JsonNode existingNode, JsonNode updateNode) {
         return result;
     }
 
+    @Override
+    public void connectorSpecificHousekeeping(String tenant) {
+    }
+
     // Helper method for deep merging of objects - unlike mergeNodes, this preserves
     // all fields
     private JsonNode deepMergeObjects(JsonNode existingObj, JsonNode updateObj) {

dynamic-mapper-service/src/main/java/dynamic/mapper/service/MappingService.java

@@ -228,17 +228,17 @@ public void sendMappingStatus(String tenant) {
     }
 
     public void sendDeviceToClientMap(String tenant) {
-        if (configurationRegistry.getAllClientMappings(tenant) == null
-                & configurationRegistry.getAllClientMappings(tenant).size() >= 0) {
+        Map<String, String> clientToDeviceMap = configurationRegistry.getAllClientMappings(tenant);
+        if (clientToDeviceMap != null
+                && clientToDeviceMap.size() >= 0) {
             subscriptionsService.runForTenant(tenant, () -> {
                 String deviceToClientMapId = configurationRegistry
                         .getDeviceToClientMapId(tenant);
-                Map<String, String> clientToDeviceMap = configurationRegistry.getAllClientMappings(tenant);
 
-                log.debug("{} - Sending Device To Client Map: {}", tenant, clientToDeviceMap.values().size());
-                
+                log.debug("{} - Sending Device To Client Map: {}", tenant, clientToDeviceMap.size());
+
                 Map<String, Object> fragment = new ConcurrentHashMap<String, Object>();
-                fragment.put(DeviceToClientMapRepresentation.DEVICE_TO_CLIENT_MAP_FRAGMENT, fragment);
+                fragment.put(DeviceToClientMapRepresentation.DEVICE_TO_CLIENT_MAP_FRAGMENT, clientToDeviceMap);
                 ManagedObjectRepresentation updateMor = new ManagedObjectRepresentation();
                 updateMor.setId(GId.asGId(deviceToClientMapId));
                 updateMor.setAttrs(fragment);