Minor: moved custom events processing in GridContinuousProcessor's methods.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/eef5afda
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/eef5afda
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/eef5afda

Branch: refs/heads/ignite-7016
Commit: eef5afdacc9de77175c4452be864ee77930dc57a
Parents: 5fa5ae7
Author: sboikov <sboi...@gridgain.com>
Authored: Wed Nov 29 11:34:23 2017 +0300
Committer: sboikov <sboi...@gridgain.com>
Committed: Wed Nov 29 11:34:23 2017 +0300

----------------------------------------------------------------------
 .../continuous/GridContinuousProcessor.java     | 132 ++++++++++++-------
 1 file changed, 85 insertions(+), 47 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/eef5afda/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
----------------------------------------------------------------------
diff --git 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
index fa52be2..571d654 100644
--- 
a/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
+++ 
b/modules/core/src/main/java/org/apache/ignite/internal/processors/continuous/GridContinuousProcessor.java
@@ -176,8 +176,10 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
                 @Override public void onCustomEvent(AffinityTopologyVersion 
topVer,
                     ClusterNode snd,
                     StartRoutineDiscoveryMessage msg) {
-                    if (!snd.id().equals(ctx.localNodeId()) && 
!ctx.isStopping())
-                        processStartRequest(snd, msg);
+                    if (ctx.isStopping())
+                        return;
+
+                    processStartRequest(snd, msg);
                 }
             });
 
@@ -186,39 +188,10 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
                 @Override public void onCustomEvent(AffinityTopologyVersion 
topVer,
                     ClusterNode snd,
                     StartRoutineAckDiscoveryMessage msg) {
-                    StartFuture fut = startFuts.remove(msg.routineId());
-
-                    if (fut != null) {
-                        if (msg.errs().isEmpty()) {
-                            LocalRoutineInfo routine = 
locInfos.get(msg.routineId());
-
-                            // Update partition counters.
-                            if (routine != null && 
routine.handler().isQuery()) {
-                                Map<UUID, Map<Integer, T2<Long, Long>>> 
cntrsPerNode = msg.updateCountersPerNode();
-                                Map<Integer, T2<Long, Long>> cntrs = 
msg.updateCounters();
-
-                                GridCacheAdapter<Object, Object> interCache =
-                                    
ctx.cache().internalCache(routine.handler().cacheName());
-
-                                GridCacheContext cctx = interCache != null ? 
interCache.context() : null;
-
-                                if (cctx != null && cntrsPerNode != null && 
!cctx.isLocal() && cctx.affinityNode())
-                                    cntrsPerNode.put(ctx.localNodeId(),
-                                        
toCountersMap(cctx.topology().localUpdateCounters(false)));
-
-                                routine.handler().updateCounters(topVer, 
cntrsPerNode, cntrs);
-                            }
-
-                            fut.onRemoteRegistered();
-                        }
-                        else {
-                            IgniteCheckedException firstEx = 
F.first(msg.errs().values());
-
-                            fut.onDone(firstEx);
+                    if (ctx.isStopping())
+                        return;
 
-                            stopRoutine(msg.routineId());
-                        }
-                    }
+                    processStartAckRequest(topVer, msg);
                 }
             });
 
@@ -227,16 +200,10 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
                 @Override public void onCustomEvent(AffinityTopologyVersion 
topVer,
                     ClusterNode snd,
                     StopRoutineDiscoveryMessage msg) {
-                    if (!snd.id().equals(ctx.localNodeId())) {
-                        UUID routineId = msg.routineId();
-
-                        unregisterRemote(routineId);
-                    }
+                    if (ctx.isStopping())
+                        return;
 
-                    for (Map<UUID, LocalRoutineInfo> clientInfo : 
clientInfos.values()) {
-                        if (clientInfo.remove(msg.routineId()) != null)
-                            break;
-                    }
+                    processStopRequest(snd, msg);
                 }
             });
 
@@ -245,10 +212,10 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
                 @Override public void onCustomEvent(AffinityTopologyVersion 
topVer,
                     ClusterNode snd,
                     StopRoutineAckDiscoveryMessage msg) {
-                    StopFuture fut = stopFuts.remove(msg.routineId());
+                    if (ctx.isStopping())
+                        return;
 
-                    if (fut != null)
-                        fut.onDone();
+                    processStopAckRequest(msg);
                 }
             });
 
@@ -459,7 +426,7 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
     /** {@inheritDoc} */
     @Override public void onJoiningNodeDataReceived(JoiningNodeDiscoveryData 
data) {
         if (log.isDebugEnabled()) {
-            log.info("onJoiningNodeDataReceived [joining=" + 
data.joiningNodeId() +
+            log.debug("onJoiningNodeDataReceived [joining=" + 
data.joiningNodeId() +
                     ", loc=" + ctx.localNodeId() +
                     ", data=" + data.joiningNodeData() +
                     ']');
@@ -976,11 +943,82 @@ public class GridContinuousProcessor extends 
GridProcessorAdapter {
     }
 
     /**
+     * @param msg Message.
+     */
+    private void processStopAckRequest(StopRoutineAckDiscoveryMessage msg) {
+        StopFuture fut = stopFuts.remove(msg.routineId());
+
+        if (fut != null)
+            fut.onDone();
+    }
+
+    /**
+     * @param snd Sender node.
+     * @param msg Message/
+     */
+    private void processStopRequest(ClusterNode snd, 
StopRoutineDiscoveryMessage msg) {
+        if (!snd.id().equals(ctx.localNodeId())) {
+            UUID routineId = msg.routineId();
+
+            unregisterRemote(routineId);
+        }
+
+        for (Map<UUID, LocalRoutineInfo> clientInfo : clientInfos.values()) {
+            if (clientInfo.remove(msg.routineId()) != null)
+                break;
+        }
+    }
+
+    /**
+     * @param topVer Topology version.
+     * @param msg Message.
+     */
+    private void processStartAckRequest(AffinityTopologyVersion topVer,
+        StartRoutineAckDiscoveryMessage msg) {
+        StartFuture fut = startFuts.remove(msg.routineId());
+
+        if (fut != null) {
+            if (msg.errs().isEmpty()) {
+                LocalRoutineInfo routine = locInfos.get(msg.routineId());
+
+                // Update partition counters.
+                if (routine != null && routine.handler().isQuery()) {
+                    Map<UUID, Map<Integer, T2<Long, Long>>> cntrsPerNode = 
msg.updateCountersPerNode();
+                    Map<Integer, T2<Long, Long>> cntrs = msg.updateCounters();
+
+                    GridCacheAdapter<Object, Object> interCache =
+                        
ctx.cache().internalCache(routine.handler().cacheName());
+
+                    GridCacheContext cctx = interCache != null ? 
interCache.context() : null;
+
+                    if (cctx != null && cntrsPerNode != null && 
!cctx.isLocal() && cctx.affinityNode())
+                        cntrsPerNode.put(ctx.localNodeId(),
+                            
toCountersMap(cctx.topology().localUpdateCounters(false)));
+
+                    routine.handler().updateCounters(topVer, cntrsPerNode, 
cntrs);
+                }
+
+                fut.onRemoteRegistered();
+            }
+            else {
+                IgniteCheckedException firstEx = F.first(msg.errs().values());
+
+                fut.onDone(firstEx);
+
+                stopRoutine(msg.routineId());
+            }
+        }
+    }
+
+    /**
      * @param node Sender.
      * @param req Start request.
      */
     private void processStartRequest(ClusterNode node, 
StartRoutineDiscoveryMessage req) {
         UUID routineId = req.routineId();
+        if (node.id().equals(ctx.localNodeId()))
+            return;
+
         StartRequestData data = req.startRequestData();
 
         GridContinuousHandler hnd = data.handler();

Reply via email to