commit 4f60f98bd890288b60620d0df1b9f6cbaeb5100b
author Jonathan Hart <jono@onlab.us> Mon Oct 27 08:11:17 2014 -0700
committer Jonathan Hart <jono@onlab.us> Thu Oct 30 11:02:52 2014 -0700
tree 7d530d8cc67be9a0a2595e1586fa99d503fed9a3
parent 07df706db10d1431b50ebb1f446fc1e23ee83c78

Implemented multi-instance packet out.

We've defined a PacketStore abstraction through which outbound packets are
sent. The packet store has a simple implementation (basically a no-op) and a
distributed implementation on top of the cluster messaging service.

Change-Id: Ib32753314fe518ef1fd67c858db744b004539938

core/net/src/main/java/org/onlab/onos/net/packet/impl/PacketManager.java

diff --git a/core/net/src/main/java/org/onlab/onos/net/packet/impl/PacketManager.java b/core/net/src/main/java/org/onlab/onos/net/packet/impl/PacketManager.java
index b90d561..5bf0ac9 100644
--- a/core/net/src/main/java/org/onlab/onos/net/packet/impl/PacketManager.java
+++ b/core/net/src/main/java/org/onlab/onos/net/packet/impl/PacketManager.java
@@ -19,7 +19,7 @@
 import static org.slf4j.LoggerFactory.getLogger;
 
 import java.util.Map;
-import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -31,11 +31,14 @@
 import org.onlab.onos.net.device.DeviceService;
 import org.onlab.onos.net.packet.OutboundPacket;
 import org.onlab.onos.net.packet.PacketContext;
+import org.onlab.onos.net.packet.PacketEvent;
 import org.onlab.onos.net.packet.PacketProcessor;
 import org.onlab.onos.net.packet.PacketProvider;
 import org.onlab.onos.net.packet.PacketProviderRegistry;
 import org.onlab.onos.net.packet.PacketProviderService;
 import org.onlab.onos.net.packet.PacketService;
+import org.onlab.onos.net.packet.PacketStore;
+import org.onlab.onos.net.packet.PacketStoreDelegate;
 import org.onlab.onos.net.provider.AbstractProviderRegistry;
 import org.onlab.onos.net.provider.AbstractProviderService;
 import org.slf4j.Logger;
@@ -51,18 +54,25 @@
 
     private final Logger log = getLogger(getClass());
 
+    private final PacketStoreDelegate delegate = new InternalStoreDelegate();
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     private DeviceService deviceService;
 
-    private final Map<Integer, PacketProcessor> processors = new TreeMap<>();
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    private PacketStore store;
+
+    private final Map<Integer, PacketProcessor> processors = new ConcurrentHashMap<>();
 
     @Activate
     public void activate() {
+        store.setDelegate(delegate);
         log.info("Started");
     }
 
     @Deactivate
     public void deactivate() {
+        store.unsetDelegate(delegate);
         log.info("Stopped");
     }
 
@@ -81,6 +91,11 @@
     @Override
     public void emit(OutboundPacket packet) {
         checkNotNull(packet, "Packet cannot be null");
+
+        store.emit(packet);
+    }
+
+    private void localEmit(OutboundPacket packet) {
         final Device device = deviceService.getDevice(packet.sendThrough());
         final PacketProvider packetProvider = getProvider(device.providerId());
         if (packetProvider != null) {
@@ -110,4 +125,16 @@
         }
 
     }
+
+    /**
+     * Internal callback from the packet store.
+     */
+    private class InternalStoreDelegate
+    implements PacketStoreDelegate {
+        @Override
+        public void notify(PacketEvent event) {
+            localEmit(event.subject());
+        }
+    }
+
 }