KAA-823: Implement topic list hash algorithm. Get rid of sequence num…

…bers in configuration and notification features.
kaaproject · Jan 21, 2016 · 7727372 · 7727372
1 parent bf9f647
commit 7727372
Show file tree

Hide file tree

Showing 16 changed files with 294 additions and 246 deletions.
diff --git a/...ent-multi/client-java-core/src/main/java/org/kaaproject/kaa/client/AbstractKaaClient.java b/...ent-multi/client-java-core/src/main/java/org/kaaproject/kaa/client/AbstractKaaClient.java
@@ -78,7 +78,11 @@
 import org.kaaproject.kaa.client.exceptions.KaaClusterConnectionException;
 import org.kaaproject.kaa.client.exceptions.KaaException;
 import org.kaaproject.kaa.client.exceptions.KaaRuntimeException;
-import org.kaaproject.kaa.client.logging.*;
+import org.kaaproject.kaa.client.logging.AbstractLogCollector;
+import org.kaaproject.kaa.client.logging.DefaultLogCollector;
+import org.kaaproject.kaa.client.logging.LogDeliveryListener;
+import org.kaaproject.kaa.client.logging.LogStorage;
+import org.kaaproject.kaa.client.logging.LogUploadStrategy;
 import org.kaaproject.kaa.client.notification.DefaultNotificationManager;
 import org.kaaproject.kaa.client.notification.NotificationListener;
 import org.kaaproject.kaa.client.notification.NotificationTopicListListener;
@@ -405,7 +409,7 @@ public void addNotificationListener(NotificationListener listener) {
     }
 
     @Override
-    public void addNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException {
+    public void addNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException {
         this.notificationManager.addNotificationListener(topicId, listener);
     }
 
@@ -415,50 +419,50 @@ public void removeNotificationListener(NotificationListener listener) {
     }
 
     @Override
-    public void removeNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException {
+    public void removeNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException {
         this.notificationManager.removeNotificationListener(topicId, listener);
     }
 
     @Override
-    public void subscribeToTopic(String topicId) throws UnavailableTopicException {
+    public void subscribeToTopic(Long topicId) throws UnavailableTopicException {
         subscribeToTopic(topicId, FORCE_SYNC);
     }
 
     @Override
-    public void subscribeToTopic(String topicId, boolean forceSync) throws UnavailableTopicException {
+    public void subscribeToTopic(Long topicId, boolean forceSync) throws UnavailableTopicException {
         checkClientState(State.STARTED, "Kaa client isn't started");
         notificationManager.subscribeToTopic(topicId, forceSync);
     }
 
     @Override
-    public void subscribeToTopics(List<String> topicIds) throws UnavailableTopicException {
+    public void subscribeToTopics(List<Long> topicIds) throws UnavailableTopicException {
         subscribeToTopics(topicIds, FORCE_SYNC);
     }
 
     @Override
-    public void subscribeToTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException {
+    public void subscribeToTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException {
         checkClientState(State.STARTED, "Kaa client isn't started");
         notificationManager.subscribeToTopics(topicIds, forceSync);
     }
 
     @Override
-    public void unsubscribeFromTopic(String topicId) throws UnavailableTopicException {
+    public void unsubscribeFromTopic(Long topicId) throws UnavailableTopicException {
         unsubscribeFromTopic(topicId, FORCE_SYNC);
     }
 
     @Override
-    public void unsubscribeFromTopic(String topicId, boolean forceSync) throws UnavailableTopicException {
+    public void unsubscribeFromTopic(Long topicId, boolean forceSync) throws UnavailableTopicException {
         checkClientState(State.STARTED, "Kaa client isn't started");
         notificationManager.unsubscribeFromTopic(topicId, forceSync);
     }
 
     @Override
-    public void unsubscribeFromTopics(List<String> topicIds) throws UnavailableTopicException {
+    public void unsubscribeFromTopics(List<Long> topicIds) throws UnavailableTopicException {
         unsubscribeFromTopics(topicIds, FORCE_SYNC);
     }
 
     @Override
-    public void unsubscribeFromTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException {
+    public void unsubscribeFromTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException {
         checkClientState(State.STARTED, "Kaa client isn't started");
         this.notificationManager.unsubscribeFromTopics(topicIds, forceSync);
     }

diff --git a/...ient-multi/client-java-core/src/main/java/org/kaaproject/kaa/client/GenericKaaClient.java b/...ient-multi/client-java-core/src/main/java/org/kaaproject/kaa/client/GenericKaaClient.java
@@ -195,7 +195,7 @@ public interface GenericKaaClient {
      *
      * @see NotificationListener
      */
-    void addNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException;
+    void addNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -230,7 +230,7 @@ public interface GenericKaaClient {
      *
      * @see NotificationListener
      */
-    void removeNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException;
+    void removeNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -244,7 +244,7 @@ public interface GenericKaaClient {
      *             Throw if unknown topic id is provided or topic isn't
      *             optional.
      */
-    void subscribeToTopic(String topicId) throws UnavailableTopicException;
+    void subscribeToTopic(Long topicId) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -263,7 +263,7 @@ public interface GenericKaaClient {
      *
      * @see #syncTopicsList()
      */
-    void subscribeToTopic(String topicId, boolean forceSync) throws UnavailableTopicException;
+    void subscribeToTopic(Long topicId, boolean forceSync) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -278,7 +278,7 @@ public interface GenericKaaClient {
      *             Throw if unknown topic id is provided or topic isn't
      *             optional.
      */
-    void subscribeToTopics(List<String> topicIds) throws UnavailableTopicException;
+    void subscribeToTopics(List<Long> topicIds) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -298,7 +298,7 @@ public interface GenericKaaClient {
      *
      * @see #syncTopicsList()
      */
-    void subscribeToTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException;
+    void subscribeToTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -316,7 +316,7 @@ public interface GenericKaaClient {
      *             Throw if unknown topic id is provided or topic isn't
      *             optional.
      */
-    void unsubscribeFromTopic(String topicId) throws UnavailableTopicException;
+    void unsubscribeFromTopic(Long topicId) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -339,7 +339,7 @@ public interface GenericKaaClient {
      *
      * @see #syncTopicsList()
      */
-    void unsubscribeFromTopic(String topicId, boolean forceSync) throws UnavailableTopicException;
+    void unsubscribeFromTopic(Long topicId, boolean forceSync) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -358,7 +358,7 @@ public interface GenericKaaClient {
      *             Throw if unknown topic id is provided or topic isn't
      *             optional.
      */
-    void unsubscribeFromTopics(List<String> topicIds) throws UnavailableTopicException;
+    void unsubscribeFromTopics(List<Long> topicIds) throws UnavailableTopicException;
 
     /**
      * <p>
@@ -382,17 +382,17 @@ public interface GenericKaaClient {
      *
      * @see #syncTopicsList()
      */
-    void unsubscribeFromTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException;
+    void unsubscribeFromTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException;
 
     /**
      * <p>
      * Force sync of pending subscription changes with server.
      * </p>
      *
      * <p>
-     * Should be used after all {@link #subscribeToTopic(String, boolean)},
+     * Should be used after all {@link #subscribeToTopic(Long, boolean)},
      * {@link #subscribeToTopics(List, boolean)},
-     * {@link #unsubscribeFromTopic(String, boolean)},
+     * {@link #unsubscribeFromTopic(Long, boolean)},
      * {@link #unsubscribeFromTopics(List, boolean)} calls with parameter
      * {@code forceSync} set to {@code false}.
      * </p>

diff --git a/...java/org/kaaproject/kaa/client/channel/impl/transports/DefaultConfigurationTransport.java b/...java/org/kaaproject/kaa/client/channel/impl/transports/DefaultConfigurationTransport.java
@@ -84,7 +84,6 @@ public ConfigurationSyncRequest createConfigurationRequest() {
             if (hash != null) {
                 request.setConfigurationHash(ByteBuffer.wrap(hash.getData()));
             }
-            request.setAppStateSeqNumber(clientState.getConfigSeqNumber());
             request.setResyncOnly(resyncOnly);
             return request;
         }
@@ -94,7 +93,6 @@ public ConfigurationSyncRequest createConfigurationRequest() {
     @Override
     public void onConfigurationResponse(ConfigurationSyncResponse response) throws IOException {
         if (clientState != null && configProcessor != null) {
-            clientState.setConfigSeqNumber(response.getAppStateSeqNumber());
             ByteBuffer schemaBody = response.getConfSchemaBody();
             if (schemaBody != null && schemaProcessor != null) {
                 schemaProcessor.loadSchema(schemaBody);

diff --git a/.../java/org/kaaproject/kaa/client/channel/impl/transports/DefaultNotificationTransport.java b/.../java/org/kaaproject/kaa/client/channel/impl/transports/DefaultNotificationTransport.java
@@ -29,6 +29,7 @@
 
 import org.kaaproject.kaa.client.channel.NotificationTransport;
 import org.kaaproject.kaa.client.notification.NotificationProcessor;
+import org.kaaproject.kaa.client.notification.TopicListHashCalculator;
 import org.kaaproject.kaa.common.TransportType;
 import org.kaaproject.kaa.common.endpoint.gen.Notification;
 import org.kaaproject.kaa.common.endpoint.gen.NotificationSyncRequest;
@@ -47,14 +48,15 @@ public class DefaultNotificationTransport extends AbstractKaaTransport implement
     private NotificationProcessor processor;
     private final Set<String> acceptedUnicastNotificationIds = new HashSet<>();
     private final List<SubscriptionCommand> sentNotificationCommands = new LinkedList<SubscriptionCommand>();
+    private Integer topicListHash = TopicListHashCalculator.calculateTopicListHash(null);
 
     private List<TopicState> getTopicStates() {
         List<TopicState> states = null;
-        Map<String, Integer> nfSubscriptions = clientState.getNfSubscriptions();
+        Map<Long, Integer> nfSubscriptions = clientState.getNfSubscriptions();
         if(!nfSubscriptions.isEmpty()){
             states = new ArrayList<>();
             LOG.info("Topic States:");
-            for(Entry<String, Integer> nfSubscription : nfSubscriptions.entrySet()){
+            for(Entry<Long, Integer> nfSubscription : nfSubscriptions.entrySet()){
                 TopicState state = new TopicState(nfSubscription.getKey(), nfSubscription.getValue());
                 states.add(state);
                 LOG.info("{} : {}", state.getTopicId(), state.getSeqNumber());
@@ -68,7 +70,7 @@ private List<TopicState> getTopicStates() {
     public NotificationSyncRequest createEmptyNotificationRequest() {
         if (clientState != null) {
             NotificationSyncRequest request = new NotificationSyncRequest();
-            request.setAppStateSeqNumber(clientState.getNotificationSeqNumber());
+            request.setTopicListHash(topicListHash);
             request.setTopicStates(getTopicStates());
             return request;
         }
@@ -79,12 +81,13 @@ public NotificationSyncRequest createEmptyNotificationRequest() {
     public NotificationSyncRequest createNotificationRequest() {
         if (clientState != null) {
             NotificationSyncRequest request = new NotificationSyncRequest();
-            request.setAppStateSeqNumber(clientState.getNotificationSeqNumber());
+
             if(!acceptedUnicastNotificationIds.isEmpty()){
                 LOG.info("Accepted unicast Notifications: {}", acceptedUnicastNotificationIds.size());
                 request.setAcceptedUnicastNotifications(new ArrayList<>(acceptedUnicastNotificationIds));
             }
             request.setSubscriptionCommands(sentNotificationCommands);
+            request.setTopicListHash(topicListHash);
             request.setTopicStates(getTopicStates());
             return request;
         }
@@ -98,6 +101,8 @@ public void onNotificationResponse(NotificationSyncResponse response) throws IOE
                 acceptedUnicastNotificationIds.clear();
             }
             List<Topic> topics = response.getAvailableTopics();
+            topicListHash = TopicListHashCalculator.calculateTopicListHash(topics);
+
             if (topics != null) {
                 for (Topic topic : topics) {
                     clientState.addTopic(topic);
@@ -131,7 +136,6 @@ public void onNotificationResponse(NotificationSyncResponse response) throws IOE
                 processor.notificationReceived(newNotifications);
             }
             sentNotificationCommands.clear();
-            clientState.setNotificationSeqNumber(response.getAppStateSeqNumber());
 
             syncAck(response.getResponseStatus());
 

diff --git a/...core/src/main/java/org/kaaproject/kaa/client/notification/DefaultNotificationManager.java b/...core/src/main/java/org/kaaproject/kaa/client/notification/DefaultNotificationManager.java
@@ -47,12 +47,12 @@
 public class DefaultNotificationManager implements NotificationManager, NotificationProcessor {
     private static final Logger LOG = LoggerFactory.getLogger(DefaultNotificationManager.class);
 
-    private Map<String, Topic> topics = new HashMap<String, Topic>();
+    private Map<Long, Topic> topics = new HashMap<>();
 
     private final ExecutorContext executorContext;
     private final NotificationDeserializer deserializer;
     private final Set<NotificationListener> mandatoryListeners = new HashSet<NotificationListener>();
-    private final Map<String, List<NotificationListener>> optionalListeners = new HashMap<String, List<NotificationListener>>();
+    private final Map<Long, List<NotificationListener>> optionalListeners = new HashMap<>();
     private final Set<NotificationTopicListListener> topicsListeners = new HashSet<NotificationTopicListListener>();
 
     private final List<SubscriptionCommand> subscriptionInfo = new LinkedList<SubscriptionCommand>();
@@ -140,7 +140,7 @@ public List<Topic> getTopics() {
     }
 
     @Override
-    public void subscribeToTopic(String topicId, boolean forceSync) throws UnavailableTopicException {
+    public void subscribeToTopic(Long topicId, boolean forceSync) throws UnavailableTopicException {
         Topic topic = findTopicById(topicId);
         if (topic.getSubscriptionType() != SubscriptionType.OPTIONAL_SUBSCRIPTION) {
             LOG.warn("Failed to subscribe: topic '{}' isn't optional", topicId);
@@ -155,10 +155,10 @@ public void subscribeToTopic(String topicId, boolean forceSync) throws Unavailab
     }
 
     @Override
-    public void subscribeToTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException {
+    public void subscribeToTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException {
         List<SubscriptionCommand> subscriptionUpdate = new LinkedList<>();
 
-        for (String id : topicIds) {
+        for (Long id : topicIds) {
             Topic topic = findTopicById(id);
             if (topic.getSubscriptionType() != SubscriptionType.OPTIONAL_SUBSCRIPTION) {
                 LOG.warn("Failed to subscribe: topic '{}' isn't optional", id);
@@ -176,7 +176,7 @@ public void subscribeToTopics(List<String> topicIds, boolean forceSync) throws U
     }
 
     @Override
-    public void unsubscribeFromTopic(String topicId, boolean forceSync) throws UnavailableTopicException {
+    public void unsubscribeFromTopic(Long topicId, boolean forceSync) throws UnavailableTopicException {
         Topic topic = findTopicById(topicId);
         if (topic.getSubscriptionType() != SubscriptionType.OPTIONAL_SUBSCRIPTION) {
             LOG.warn("Failed to unsubscribe: topic '{}' isn't optional", topicId);
@@ -191,10 +191,10 @@ public void unsubscribeFromTopic(String topicId, boolean forceSync) throws Unava
     }
 
     @Override
-    public void unsubscribeFromTopics(List<String> topicIds, boolean forceSync) throws UnavailableTopicException {
+    public void unsubscribeFromTopics(List<Long> topicIds, boolean forceSync) throws UnavailableTopicException {
         List<SubscriptionCommand> subscriptionUpdate = new LinkedList<>();
 
-        for (String id : topicIds) {
+        for (Long id : topicIds) {
             Topic topic = findTopicById(id);
             if (topic.getSubscriptionType() != SubscriptionType.OPTIONAL_SUBSCRIPTION) {
                 LOG.warn("Failed to unsubscribe: topic '{}' isn't optional", id);
@@ -212,8 +212,8 @@ public void unsubscribeFromTopics(List<String> topicIds, boolean forceSync) thro
     }
 
     @Override
-    public void addNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException {
+    public void addNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException {
-        if (topicId == null || listener == null) {
+        if (listener == null) {
             LOG.warn("Failed to add listener: id={}, listener={}", topicId, listener);
             throw new IllegalArgumentException("Bad listener data");
         }
@@ -233,7 +233,7 @@ public void addNotificationListener(String topicId, NotificationListener listene
     }
 
     @Override
-    public void removeNotificationListener(String topicId, NotificationListener listener) throws UnavailableTopicException {
+    public void removeNotificationListener(Long topicId, NotificationListener listener) throws UnavailableTopicException {
         if (topicId == null || listener == null) {
             LOG.warn("Failed to remove listener: id={}, listener={}", topicId, listener);
             throw new IllegalArgumentException("Bad listener data");
@@ -257,7 +257,7 @@ public void sync() {
 
     @Override
     public void topicsListUpdated(final List<Topic> list) {
-        Map<String, Topic> newTopics = new HashMap<String, Topic>();
+        Map<Long, Topic> newTopics = new HashMap<>();
 
         synchronized (topics) {
             for (Topic topic : list) {
@@ -329,7 +329,7 @@ public void run() {
         }
     }
 
-    private void updateSubscriptionInfo(String id, SubscriptionCommandType type) {
+    private void updateSubscriptionInfo(Long id, SubscriptionCommandType type) {
         synchronized (subscriptionInfo) {
             subscriptionInfo.add(new SubscriptionCommand(id, type));
         }
@@ -341,7 +341,7 @@ private void updateSubscriptionInfo(List<SubscriptionCommand> subscriptionUpdate
         }
     }
 
-    private Topic findTopicById(String id) throws UnavailableTopicException {
+    private Topic findTopicById(Long id) throws UnavailableTopicException {
         synchronized (topics) {
             Topic topic = topics.get(id);
             if (topic == null) {

diff --git a/...-java-core/src/main/java/org/kaaproject/kaa/client/notification/NotificationListener.java b/...-java-core/src/main/java/org/kaaproject/kaa/client/notification/NotificationListener.java
@@ -37,6 +37,6 @@ public interface NotificationListener {
      * @param notification the notification object.
      *
      */
-    void onNotification(String topicId, Notification notification);
+    void onNotification(long topicId, Notification notification);
 
 }
diff --git a/...a-core/src/main/java/org/kaaproject/kaa/client/notification/NotificationListenerInfo.java b/...a-core/src/main/java/org/kaaproject/kaa/client/notification/NotificationListenerInfo.java
@@ -20,8 +20,8 @@
  * <p>Wrapper class for a topic subscription stuff.</p>
  *
  * <p><b>This class is deprecated</b>. Use instead:
- * {@link NotificationManager#subscribeToTopic(String, boolean)},
+ * {@link NotificationManager#subscribeToTopic(Long, boolean)},
- * {@link NotificationManager#addNotificationListener(String, NotificationListener)},
+ * {@link NotificationManager#addNotificationListener(Long, NotificationListener)},
  * {@link NotificationManager#subscribeToTopics(java.util.List, boolean)},
  * {@link NotificationManager#addTopicListListener(NotificationTopicListListener)}.</p>
  *