removeListener on deactivate

Change-Id: I82b2cc7960bc19ebb1c6875e8cf3dd79d7421dd1

@@ -72,6 +72,10 @@ public class DistributedDeviceStore
     private IMap<byte[], byte[]> rawDevicePorts;
     private LoadingCache<DeviceId, Optional<Map<PortNumber, Port>>> devicePorts;
 
+    private String devicesListener;
+
+    private String portsListener;
+
     @Override
     @Activate
     public void activate() {
@@ -86,7 +90,7 @@ public class DistributedDeviceStore
                 = new OptionalCacheLoader<>(storeService, rawDevices);
         devices = new AbsentInvalidatingLoadingCache<>(newBuilder().build(deviceLoader));
         // refresh/populate cache based on notification from other instance
-        rawDevices.addEntryListener(new RemoteDeviceEventHandler(devices), includeValue);
+        devicesListener = rawDevices.addEntryListener(new RemoteDeviceEventHandler(devices), includeValue);
 
         // TODO cache availableDevices
         availableDevices = theInstance.getSet("availableDevices");
@@ -96,7 +100,7 @@ public class DistributedDeviceStore
                 = new OptionalCacheLoader<>(storeService, rawDevicePorts);
         devicePorts = new AbsentInvalidatingLoadingCache<>(newBuilder().build(devicePortLoader));
         // refresh/populate cache based on notification from other instance
-        rawDevicePorts.addEntryListener(new RemotePortEventHandler(devicePorts), includeValue);
+        portsListener = rawDevicePorts.addEntryListener(new RemotePortEventHandler(devicePorts), includeValue);
 
         loadDeviceCache();
         loadDevicePortsCache();
@@ -106,6 +110,8 @@ public class DistributedDeviceStore
 
     @Deactivate
     public void deactivate() {
+        rawDevicePorts.removeEntryListener(portsListener);
+        rawDevices.removeEntryListener(devicesListener);
         log.info("Stopped");
     }
 

@@ -58,6 +58,8 @@ public class DistributedLinkStore
     private final Multimap<DeviceId, Link> srcLinks = HashMultimap.create();
     private final Multimap<DeviceId, Link> dstLinks = HashMultimap.create();
 
+    private String linksListener;
+
     @Override
     @Activate
     public void activate() {
@@ -71,7 +73,7 @@ public class DistributedLinkStore
                 = new OptionalCacheLoader<>(storeService, rawLinks);
         links = new AbsentInvalidatingLoadingCache<>(newBuilder().build(linkLoader));
         // refresh/populate cache based on notification from other instance
-        rawLinks.addEntryListener(new RemoteLinkEventHandler(links), includeValue);
+        linksListener = rawLinks.addEntryListener(new RemoteLinkEventHandler(links), includeValue);
 
         loadLinkCache();
 
@@ -80,7 +82,7 @@ public class DistributedLinkStore
 
     @Deactivate
     public void deactivate() {
-        super.activate();
+        rawLinks.removeEntryListener(linksListener);
         log.info("Stopped");
     }