Reactive forwarding apps should explicitly ask for packets they want to receive.

Previously we relied on the table-miss entry set up for OVS switches, but
this doesn't work for non-OVS switches 1.3 switches.

Fixes ONOS-661.

Change-Id: Ibc0aee09f304eaf240691a4d5d2f8765a5f8cdd5

@@ -30,10 +30,14 @@ import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.onlab.packet.Ethernet;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
+import org.onosproject.net.Device;
 import org.onosproject.net.Host;
 import org.onosproject.net.HostId;
 import org.onosproject.net.Path;
 import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
 import org.onosproject.net.flow.DefaultFlowRule;
 import org.onosproject.net.flow.DefaultTrafficSelector;
 import org.onosproject.net.flow.DefaultTrafficTreatment;
@@ -58,6 +62,7 @@ public class ReactiveForwarding {
 
     private static final int TIMEOUT = 10;
     private static final int PRIORITY = 10;
+    private static final int PUNT_RULE_PRIORITY = 5;
 
     private final Logger log = getLogger(getClass());
 
@@ -74,6 +79,9 @@ public class ReactiveForwarding {
     protected FlowRuleService flowRuleService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected CoreService coreService;
 
     private ReactivePacketProcessor processor = new ReactivePacketProcessor();
@@ -91,6 +99,8 @@ public class ReactiveForwarding {
     @Activate
     public void activate(ComponentContext context) {
         appId = coreService.registerApplication("org.onosproject.fwd");
+        deviceService.addListener(new InternalDeviceListener());
+        pushRules();
         packetService.addProcessor(processor, PacketProcessor.ADVISOR_MAX + 2);
         readComponentConfiguration(context);
         log.info("Started with Application ID {}", appId.id());
@@ -293,6 +303,58 @@ public class ReactiveForwarding {
         }
     }
 
+    /**
+     * Pushes flow rules to all devices.
+     */
+    private void pushRules() {
+        for (Device device : deviceService.getDevices()) {
+            pushRules(device);
+        }
+    }
+
+    /**
+     * Pushes flow rules to the device to receive packets that need
+     * to be processed.
+     *
+     * @param device the device to push the rules to
+     */
+    private synchronized void pushRules(Device device) {
+        TrafficSelector.Builder sbuilder = DefaultTrafficSelector.builder();
+        TrafficTreatment.Builder tbuilder = DefaultTrafficTreatment.builder();
+
+        // Get all IPv4 packets
+        sbuilder.matchEthType(Ethernet.TYPE_IPV4);
+        tbuilder.punt();
+        FlowRule flowArp =
+            new DefaultFlowRule(device.id(),
+                                sbuilder.build(), tbuilder.build(),
+                                PUNT_RULE_PRIORITY, appId, 0, true);
+
+        flowRuleService.applyFlowRules(flowArp);
+    }
+
+    public class InternalDeviceListener implements DeviceListener {
+
+        @Override
+        public void event(DeviceEvent event) {
+            Device device = event.subject();
+            switch (event.type()) {
+                case DEVICE_ADDED:
+                    pushRules(device);
+                    break;
+                case DEVICE_AVAILABILITY_CHANGED:
+                case DEVICE_SUSPENDED:
+                case DEVICE_UPDATED:
+                case DEVICE_REMOVED:
+                case PORT_ADDED:
+                case PORT_UPDATED:
+                case PORT_REMOVED:
+                default:
+                    break;
+            }
+        }
+    }
+
 }
 
 

@@ -25,11 +25,18 @@ import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.onlab.packet.Ethernet;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
+import org.onosproject.net.Device;
 import org.onosproject.net.Host;
 import org.onosproject.net.HostId;
 import org.onosproject.net.PortNumber;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.flow.DefaultFlowRule;
 import org.onosproject.net.flow.DefaultTrafficSelector;
 import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.FlowRuleService;
 import org.onosproject.net.flow.TrafficSelector;
 import org.onosproject.net.flow.TrafficTreatment;
 import org.onosproject.net.host.HostService;
@@ -52,6 +59,8 @@ public class IntentReactiveForwarding {
 
     private final Logger log = getLogger(getClass());
 
+    private static final int PUNT_RULE_PRIORITY = 5;
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected CoreService coreService;
 
@@ -67,12 +76,20 @@ public class IntentReactiveForwarding {
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected HostService hostService;
 
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected FlowRuleService flowRuleService;
+
+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+    protected DeviceService deviceService;
+
     private ReactivePacketProcessor processor = new ReactivePacketProcessor();
     private ApplicationId appId;
 
     @Activate
     public void activate() {
         appId = coreService.registerApplication("org.onosproject.ifwd");
+        deviceService.addListener(new InternalDeviceListener());
+        pushRules();
         packetService.addProcessor(processor, PacketProcessor.ADVISOR_MAX + 2);
         log.info("Started");
     }
@@ -155,4 +172,56 @@ public class IntentReactiveForwarding {
         intentService.submit(intent);
     }
 
+    /**
+     * Pushes flow rules to all devices.
+     */
+    private void pushRules() {
+        for (Device device : deviceService.getDevices()) {
+            pushRules(device);
+        }
+    }
+
+    /**
+     * Pushes flow rules to the device to receive packets that need
+     * to be processed.
+     *
+     * @param device the device to push the rules to
+     */
+    private synchronized void pushRules(Device device) {
+        TrafficSelector.Builder sbuilder = DefaultTrafficSelector.builder();
+        TrafficTreatment.Builder tbuilder = DefaultTrafficTreatment.builder();
+
+        // Get all IPv4 packets
+        sbuilder.matchEthType(Ethernet.TYPE_IPV4);
+        tbuilder.punt();
+        FlowRule flowArp =
+            new DefaultFlowRule(device.id(),
+                                sbuilder.build(), tbuilder.build(),
+                                PUNT_RULE_PRIORITY, appId, 0, true);
+
+        flowRuleService.applyFlowRules(flowArp);
+    }
+
+    public class InternalDeviceListener implements DeviceListener {
+
+        @Override
+        public void event(DeviceEvent event) {
+            Device device = event.subject();
+            switch (event.type()) {
+                case DEVICE_ADDED:
+                    pushRules(device);
+                    break;
+                case DEVICE_AVAILABILITY_CHANGED:
+                case DEVICE_SUSPENDED:
+                case DEVICE_UPDATED:
+                case DEVICE_REMOVED:
+                case PORT_ADDED:
+                case PORT_UPDATED:
+                case PORT_REMOVED:
+                default:
+                    break;
+            }
+        }
+    }
+
 }