Merge pull request #160 from zyclonite/fix_subsmaphelper_cursor_5_0

Continous query cursor cleanup, localSubs clear on leave

Author: zyclonite

src/main/java/io/vertx/spi/cluster/ignite/impl/SubsMapHelper.java

@@ -28,6 +28,7 @@
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.cache.CacheEntryProcessor;
 import org.apache.ignite.cache.query.ContinuousQuery;
+import org.apache.ignite.cache.query.QueryCursor;
 import org.apache.ignite.cache.query.ScanQuery;
 
 import javax.cache.Cache;
@@ -50,14 +51,15 @@ public class SubsMapHelper {
   private final RegistrationListener registrationListener;
   private final ConcurrentMap<String, Set<RegistrationInfo>> localSubs = new ConcurrentHashMap<>();
   private final Throttling throttling;
+  private final QueryCursor<Cache.Entry<String, Set<IgniteRegistrationInfo>>> continuousQuery;
   private volatile boolean shutdown;
 
   public SubsMapHelper(Ignite ignite, RegistrationListener registrationListener, VertxInternal vertxInternal) {
     map = ignite.getOrCreateCache("__vertx.subs");
     this.registrationListener = registrationListener;
     throttling = new Throttling(vertxInternal, a -> getAndUpdate(a, vertxInternal));
     shutdown = false;
-    map.query(new ContinuousQuery<String, Set<IgniteRegistrationInfo>>()
+    continuousQuery = map.query(new ContinuousQuery<String, Set<IgniteRegistrationInfo>>()
             .setAutoUnsubscribe(true)
             .setTimeInterval(100L)
             .setPageSize(128)
@@ -163,13 +165,22 @@ public void removeAllForNode(String nodeId) {
 
   public void leave() {
     shutdown = true;
+    try {
+      continuousQuery.close();
+    } catch (Exception e) {
+      log.debug("Failed to close continuous query cursor: " + e.getMessage());
+    }
+    localSubs.clear();
   }
 
   private void fireRegistrationUpdateEvent(String address) {
     throttling.onEvent(address);
   }
 
   private Future<List<RegistrationInfo>> getAndUpdate(String address, VertxInternal vertxInternal) {
+    if (shutdown) {
+      return Future.succeededFuture();
+    }
     Promise<List<RegistrationInfo>> prom = Promise.promise();
     if (registrationListener.wantsUpdatesFor(address)) {
       prom.future().onSuccess(registrationInfos -> {
@@ -185,6 +196,9 @@ private Future<List<RegistrationInfo>> getAndUpdate(String address, VertxInterna
   }
 
   private void listen(final Iterable<CacheEntryEvent<? extends String, ? extends Set<IgniteRegistrationInfo>>> events, final VertxInternal vertxInternal) {
+    if (shutdown) {
+      return;
+    }
     vertxInternal.executeBlocking(() -> {
       StreamSupport.stream(events.spliterator(), false)
               .map(Cache.Entry::getKey)
@@ -259,4 +273,4 @@ public Void process(MutableEntry<String, Set<IgniteRegistrationInfo>> entry, Obj
       return null;
     }
   }
-}
+}