GUI -- Further work on refactoring Topology View server side code. Still WIP...

- added topology client heartbeat.
- modified AbstractListenerRegistry to allow for extension.

Change-Id: Ib8ea6ad4ba34f5732d062f1c9ef545f105eb167b

@@ -33,7 +33,7 @@ public class AbstractListenerRegistry<E extends Event, L extends EventListener<E
 
 
     private final Logger log = getLogger(getClass());
     private final Logger log = getLogger(getClass());
 
 
-    private final Set<L> listeners = new CopyOnWriteArraySet<>();
+    protected final Set<L> listeners = new CopyOnWriteArraySet<>();
     private volatile boolean shutdown = false;
     private volatile boolean shutdown = false;
 
 
     /**
     /**
@@ -93,5 +93,4 @@ public class AbstractListenerRegistry<E extends Event, L extends EventListener<E
         shutdown = true;
         shutdown = true;
     }
     }
 
 
-
 }
 }

@@ -41,6 +41,7 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.Map;
 
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static java.lang.System.currentTimeMillis;
 import static org.onosproject.ui.impl.topo.TopoUiEvent.Type.SUMMARY_UPDATE;
 import static org.onosproject.ui.impl.topo.TopoUiEvent.Type.SUMMARY_UPDATE;
 
 
 /**
 /**
@@ -51,6 +52,7 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
             implements OverlayService {
             implements OverlayService {
 
 
     private static final String TOPO_START = "topoStart";
     private static final String TOPO_START = "topoStart";
+    private static final String TOPO_HEARTBEAT = "topoHeartbeat";
     private static final String TOPO_STOP = "topoStop";
     private static final String TOPO_STOP = "topoStop";
     private static final String REQ_SUMMARY = "requestSummary";
     private static final String REQ_SUMMARY = "requestSummary";
     private static final String CANCEL_SUMMARY = "cancelSummary";
     private static final String CANCEL_SUMMARY = "cancelSummary";
@@ -60,7 +62,7 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
     protected ServiceDirectory directory;
     protected ServiceDirectory directory;
     protected TopoUiModelService modelService;
     protected TopoUiModelService modelService;
 
 
-    private TopoUiListener modelListener;
+    private ModelListener modelListener;
     private String version;
     private String version;
     private SummaryGenerator defaultSummaryGenerator;
     private SummaryGenerator defaultSummaryGenerator;
     private SummaryGenerator currentSummaryGenerator;
     private SummaryGenerator currentSummaryGenerator;
@@ -83,22 +85,17 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
 
 
     @Override
     @Override
     public void destroy() {
     public void destroy() {
-//        cancelAllMonitoring();
-//        stopListeningToModel();
+        cancelAllMonitoring();
+        stopListeningToModel();
         super.destroy();
         super.destroy();
     }
     }
 
 
 
 
-    private String getVersion() {
-        String ver = directory.get(CoreService.class).version().toString();
-        return ver.replace(".SNAPSHOT", "*").replaceFirst("~.*$", "");
-    }
-
-
     @Override
     @Override
     protected Collection<RequestHandler> createRequestHandlers() {
     protected Collection<RequestHandler> createRequestHandlers() {
         return ImmutableSet.of(
         return ImmutableSet.of(
                 new TopoStart(),
                 new TopoStart(),
+                new TopoHeartbeat(),
                 new TopoStop(),
                 new TopoStop(),
                 new ReqSummary(),
                 new ReqSummary(),
                 new CancelSummary()
                 new CancelSummary()
@@ -107,6 +104,27 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
     }
     }
 
 
     // =====================================================================
     // =====================================================================
+
+    private void cancelAllMonitoring() {
+        // TODO:
+    }
+
+    private void startListeningToModel() {
+        topoActive = true;
+        modelService.addListener(modelListener);
+    }
+
+    private void stopListeningToModel() {
+        topoActive = false;
+        modelService.removeListener(modelListener);
+    }
+
+    private String getVersion() {
+        String ver = directory.get(CoreService.class).version().toString();
+        return ver.replace(".SNAPSHOT", "*").replaceFirst("~.*$", "");
+    }
+
+    // =====================================================================
     // Overlay Service
     // Overlay Service
     // TODO: figure out how we are going to switch overlays in and out...
     // TODO: figure out how we are going to switch overlays in and out...
 
 
@@ -136,12 +154,21 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
 
 
         @Override
         @Override
         public void process(long sid, ObjectNode payload) {
         public void process(long sid, ObjectNode payload) {
-            topoActive = true;
-            modelService.addListener(modelListener);
+            startListeningToModel();
             sendMessages(modelService.getInitialState());
             sendMessages(modelService.getInitialState());
         }
         }
     }
     }
 
 
+    private final class TopoHeartbeat extends RequestHandler {
+        private TopoHeartbeat() {
+            super(TOPO_HEARTBEAT);
+        }
+        @Override
+        public void process(long sid, ObjectNode payload) {
+            modelListener.nudge();
+        }
+    }
+
     private final class TopoStop extends RequestHandler {
     private final class TopoStop extends RequestHandler {
         private TopoStop() {
         private TopoStop() {
             super(TOPO_STOP);
             super(TOPO_STOP);
@@ -149,8 +176,7 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
 
 
         @Override
         @Override
         public void process(long sid, ObjectNode payload) {
         public void process(long sid, ObjectNode payload) {
-            topoActive = false;
-            modelService.removeListener(modelListener);
+            stopListeningToModel();
         }
         }
     }
     }
 
 
@@ -227,6 +253,10 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
     // Our listener for model events so we can push changes out to the UI...
     // Our listener for model events so we can push changes out to the UI...
 
 
     private class ModelListener implements TopoUiListener {
     private class ModelListener implements TopoUiListener {
+        private static final long AWAKE_THRESHOLD_MS = 6000;
+
+        private long lastNudged = currentTimeMillis();
+
         @Override
         @Override
         public void event(TopoUiEvent event) {
         public void event(TopoUiEvent event) {
             log.debug("Handle Event: {}", event);
             log.debug("Handle Event: {}", event);
@@ -238,6 +268,15 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
             }
             }
             handler.handleEvent(event);
             handler.handleEvent(event);
         }
         }
+
+        @Override
+        public boolean isAwake() {
+            return currentTimeMillis() - lastNudged < AWAKE_THRESHOLD_MS;
+        }
+
+        public void nudge() {
+            lastNudged = currentTimeMillis();
+        }
     }
     }
 
 
 
 
@@ -271,5 +310,4 @@ public class AltTopoViewMessageHandler extends UiMessageHandler
         eventHandlerBinding.put(SUMMARY_UPDATE, summaryHandler);
         eventHandlerBinding.put(SUMMARY_UPDATE, summaryHandler);
         // NOTE: no need to bind pass-thru handlers
         // NOTE: no need to bind pass-thru handlers
     }
     }
-
 }
 }

@@ -99,6 +99,7 @@ public class TopologyViewMessageHandler extends TopologyViewMessageHandlerBase {
     private static final String SPRITE_LIST_REQ = "spriteListRequest";
     private static final String SPRITE_LIST_REQ = "spriteListRequest";
     private static final String SPRITE_DATA_REQ = "spriteDataRequest";
     private static final String SPRITE_DATA_REQ = "spriteDataRequest";
     private static final String TOPO_START = "topoStart";
     private static final String TOPO_START = "topoStart";
+    private static final String TOPO_HEARTBEAT = "topoHeartbeat";
     private static final String TOPO_STOP = "topoStop";
     private static final String TOPO_STOP = "topoStop";
 
 
 
 
@@ -170,6 +171,7 @@ public class TopologyViewMessageHandler extends TopologyViewMessageHandlerBase {
     protected Collection<RequestHandler> createRequestHandlers() {
     protected Collection<RequestHandler> createRequestHandlers() {
         return ImmutableSet.of(
         return ImmutableSet.of(
                 new TopoStart(),
                 new TopoStart(),
+                new TopoHeartbeat(),
                 new TopoStop(),
                 new TopoStop(),
                 new ReqSummary(),
                 new ReqSummary(),
                 new CancelSummary(),
                 new CancelSummary(),
@@ -211,6 +213,18 @@ public class TopologyViewMessageHandler extends TopologyViewMessageHandlerBase {
     }
     }
 
 
     @Deprecated
     @Deprecated
+    private final class TopoHeartbeat extends RequestHandler {
+        private TopoHeartbeat() {
+            super(TOPO_HEARTBEAT);
+        }
+
+        @Override
+        public void process(long sid, ObjectNode payload) {
+            // place holder for now
+        }
+    }
+
+    @Deprecated
     private final class TopoStop extends RequestHandler {
     private final class TopoStop extends RequestHandler {
         private TopoStop() {
         private TopoStop() {
             super(TOPO_STOP);
             super(TOPO_STOP);

+/*
+ * Copyright 2015 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.onosproject.ui.impl.topo;
+
+import org.onosproject.event.AbstractListenerRegistry;
+import org.slf4j.Logger;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * A listener registry that automatically prunes listeners that have not
+ * been sending heartbeat messages to assure us they are really listening.
+ */
+// package private
+class ModelListenerRegistry
+        extends AbstractListenerRegistry<TopoUiEvent, TopoUiListener> {
+
+    private final Logger log = getLogger(getClass());
+
+    private final Set<TopoUiListener> zombies = new HashSet<>();
+
+    @Override
+    public void process(TopoUiEvent event) {
+        zombies.clear();
+        for (TopoUiListener listener : listeners) {
+            try {
+                if (listener.isAwake()) {
+                    listener.event(event);
+                } else {
+                    zombies.add(listener);
+                }
+            } catch (Exception error) {
+                reportProblem(event, error);
+            }
+        }
+
+        // clean up zombie listeners
+        for (TopoUiListener z : zombies) {
+            log.debug("Removing zombie model listener: {}", z);
+            removeListener(z);
+        }
+    }
+}

@@ -22,6 +22,7 @@ import org.onosproject.ui.impl.topo.overlay.SummaryGenerator;
 /**
 /**
  * Provides the API for external agents to inject topology overlay behavior.
  * Provides the API for external agents to inject topology overlay behavior.
  */
  */
+// TODO: move to core-api module
 public interface OverlayService {
 public interface OverlayService {
 
 
     /**
     /**

@@ -20,6 +20,7 @@ package org.onosproject.ui.impl.topo;
 /**
 /**
  * Provides basic summary data for the topology.
  * Provides basic summary data for the topology.
  */
  */
+// TODO: review -- move to core-api module?
 public interface SummaryData {
 public interface SummaryData {
 
 
     /**
     /**

@@ -24,4 +24,10 @@ import org.onosproject.event.EventListener;
  */
  */
 public interface TopoUiListener extends EventListener<TopoUiEvent> {
 public interface TopoUiListener extends EventListener<TopoUiEvent> {
 
 
+    /**
+     * Returns true if the listener really is listening.
+     *
+     * @return true if awake
+     */
+    boolean isAwake();
 }
 }

@@ -27,7 +27,6 @@ import org.apache.felix.scr.annotations.Service;
 import org.onosproject.cluster.ClusterEvent;
 import org.onosproject.cluster.ClusterEvent;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.ControllerNode;
 import org.onosproject.cluster.ControllerNode;
-import org.onosproject.event.AbstractListenerRegistry;
 import org.onosproject.event.EventDeliveryService;
 import org.onosproject.event.EventDeliveryService;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.net.Device;
 import org.onosproject.net.Device;
@@ -103,9 +102,8 @@ public class TopoUiModelManager implements TopoUiModelService {
     protected EventDeliveryService eventDispatcher;
     protected EventDeliveryService eventDispatcher;
 
 
 
 
-    private AbstractListenerRegistry<TopoUiEvent, TopoUiListener>
-            listenerRegistry = new AbstractListenerRegistry<>();
-
+    private final ModelListenerRegistry listenerRegistry =
+            new ModelListenerRegistry();
 
 
     private final TopoMessageFactory messageFactory = new TopoMessageFactory();
     private final TopoMessageFactory messageFactory = new TopoMessageFactory();
     private final MetaDb metaDb = new MetaDb();
     private final MetaDb metaDb = new MetaDb();
@@ -138,17 +136,6 @@ public class TopoUiModelManager implements TopoUiModelService {
     }
     }
 
 
 
 
-    // TODO: figure out how to cull zombie listeners
-    // The problem is when one refreshes the GUI (topology view)
-    //  a new instance of AltTopoViewMessageHandler is created and added
-    //  as a listener, but we never got a TopoStop event, which is what
-    //  causes the listener (for an AltTopoViewMessageHandler instance) to
-    //  be removed.
-    // ==== Somehow need to tie this in to the GUI-disconnected event.
-    //  This probably requires client-generated heartbeat messages to
-    //  Keep the connection alive.
-
-
     @Override
     @Override
     public void addListener(TopoUiListener listener) {
     public void addListener(TopoUiListener listener) {
         listenerRegistry.addListener(listener);
         listenerRegistry.addListener(listener);
@@ -156,7 +143,12 @@ public class TopoUiModelManager implements TopoUiModelService {
 
 
     @Override
     @Override
     public void removeListener(TopoUiListener listener) {
     public void removeListener(TopoUiListener listener) {
-        listenerRegistry.removeListener(listener);
+        // we don't really care if the listener is not listed...
+        try {
+            listenerRegistry.removeListener(listener);
+        } catch (IllegalArgumentException e) {
+            log.debug("Oops, listener not registered: {}", listener);
+        }
     }
     }
 
 
     @Override
     @Override

@@ -27,11 +27,14 @@
     'use strict';
     'use strict';
 
 
     // injected refs
     // injected refs
-    var $log, wss, tps, tis, tfs, tss, tts, tspr;
+    var $log, $interval, wss, tps, tis, tfs, tss, tts, tspr;
 
 
     // internal state
     // internal state
     var handlerMap,
     var handlerMap,
-        openListener;
+        openListener,
+        heartbeatTimer;
+
+    var heartbeatPeriod = 5000; // 5 seconds
 
 
     // ==========================
     // ==========================
 
 
@@ -68,14 +71,31 @@
         wss.sendEvent('topoStart');
         wss.sendEvent('topoStart');
     }
     }
 
 
+    function cancelHeartbeat() {
+        if (heartbeatTimer) {
+            $interval.cancel(heartbeatTimer);
+        }
+        heartbeatTimer = null;
+    }
+
+    function scheduleHeartbeat() {
+        cancelHeartbeat();
+        heartbeatTimer = $interval(function () {
+            wss.sendEvent('topoHeartbeat');
+        }, heartbeatPeriod);
+    }
+
+
     angular.module('ovTopo')
     angular.module('ovTopo')
     .factory('TopoEventService',
     .factory('TopoEventService',
-        ['$log', '$location', 'WebSocketService',
+        ['$log', '$interval', 'WebSocketService',
             'TopoPanelService', 'TopoInstService', 'TopoForceService',
             'TopoPanelService', 'TopoInstService', 'TopoForceService',
             'TopoSelectService', 'TopoTrafficService', 'TopoSpriteService',
             'TopoSelectService', 'TopoTrafficService', 'TopoSpriteService',
 
 
-        function (_$log_, $loc, _wss_, _tps_, _tis_, _tfs_, _tss_, _tts_, _tspr_) {
+        function (_$log_,  _$interval_, _wss_,
+                  _tps_, _tis_, _tfs_, _tss_, _tts_, _tspr_) {
             $log = _$log_;
             $log = _$log_;
+            $interval = _$interval_;
             wss = _wss_;
             wss = _wss_;
             tps = _tps_;
             tps = _tps_;
             tis = _tis_;
             tis = _tis_;
@@ -90,10 +110,12 @@
                 openListener = wss.addOpenListener(wsOpen);
                 openListener = wss.addOpenListener(wsOpen);
                 wss.bindHandlers(handlerMap);
                 wss.bindHandlers(handlerMap);
                 wss.sendEvent('topoStart');
                 wss.sendEvent('topoStart');
+                scheduleHeartbeat();
                 $log.debug('topo comms started');
                 $log.debug('topo comms started');
             }
             }
 
 
             function stop() {
             function stop() {
+                cancelHeartbeat();
                 wss.sendEvent('topoStop');
                 wss.sendEvent('topoStop');
                 wss.unbindHandlers(handlerMap);
                 wss.unbindHandlers(handlerMap);
                 wss.removeOpenListener(openListener);
                 wss.removeOpenListener(openListener);