OLT ability to remove a subscriber

Change-Id: I5fee9dd8189ae374bf39b0a74da5bd33304a3346

apps/olt/src/main/java/org/onosproject/olt/Olt.java

@@ -15,6 +15,8 @@
  */
 package org.onosproject.olt;
 
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -52,6 +54,7 @@ import org.slf4j.Logger;
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -94,6 +97,11 @@ public class Olt
 
     private Map<DeviceId, AccessDeviceData> oltData = new ConcurrentHashMap<>();
 
+    private Map<ConnectPoint, Set<ForwardingObjective.Builder>> objectives =
+            Maps.newConcurrentMap();
+
+    private Map<ConnectPoint, VlanId> subscribers = Maps.newConcurrentMap();
+
     private InternalNetworkConfigListener configListener =
             new InternalNetworkConfigListener();
     private static final Class<AccessDeviceConfig> CONFIG_CLASS =
@@ -108,6 +116,7 @@ public class Olt
                 }
             };
 
+
     @Activate
     public void activate() {
         appId = coreService.registerApplication("org.onosproject.olt");
@@ -152,7 +161,68 @@ public class Olt
 
     @Override
     public void removeSubscriber(ConnectPoint port) {
-        throw new UnsupportedOperationException();
+        AccessDeviceData olt = oltData.get(port.deviceId());
+
+        if (olt == null) {
+            log.warn("No data found for OLT device {}", port.deviceId());
+            return;
+        }
+
+        unprovisionSubscriber(olt.deviceId(), olt.uplink(), port.port(), olt.vlan());
+
+    }
+
+    private void unprovisionSubscriber(DeviceId deviceId, PortNumber uplink,
+                                       PortNumber subscriberPort, VlanId deviceVlan) {
+
+        //FIXME: This method is slightly ugly but it'll do until we have a better
+        // way to remove flows from the flow store.
+
+        CompletableFuture<ObjectiveError> downFuture = new CompletableFuture();
+        CompletableFuture<ObjectiveError> upFuture = new CompletableFuture();
+
+        ConnectPoint cp = new ConnectPoint(deviceId, subscriberPort);
+
+        VlanId subscriberVlan = subscribers.remove(cp);
+
+        Set<ForwardingObjective.Builder> fwds = objectives.remove(cp);
+
+        if (fwds == null || fwds.size() != 2) {
+            log.warn("Unknown or incomplete subscriber at {}", cp);
+            return;
+        }
+
+
+        fwds.stream().forEach(
+                fwd -> flowObjectiveService.forward(deviceId,
+                                                 fwd.remove(new ObjectiveContext() {
+                                                     @Override
+                                                     public void onSuccess(Objective objective) {
+                                                         upFuture.complete(null);
+                                                     }
+
+                                                     @Override
+                                                     public void onError(Objective objective, ObjectiveError error) {
+                                                         upFuture.complete(error);
+                                                     }
+                                                 })));
+
+        upFuture.thenAcceptBothAsync(downFuture, (upStatus, downStatus) -> {
+            if (upStatus == null && downStatus == null) {
+                post(new AccessDeviceEvent(AccessDeviceEvent.Type.SUBSCRIBER_UNREGISTERED,
+                                           deviceId,
+                                           deviceVlan,
+                                           subscriberVlan));
+            } else if (downStatus != null) {
+                log.error("Subscriber with vlan {} on device {} " +
+                                  "on port {} failed downstream uninstallation: {}",
+                          subscriberVlan, deviceId, subscriberPort, downStatus);
+            } else if (upStatus != null) {
+                log.error("Subscriber with vlan {} on device {} " +
+                                  "on port {} failed upstream uninstallation: {}",
+                          subscriberVlan, deviceId, subscriberPort, upStatus);
+            }
+        }, oltInstallers);
 
     }
 
@@ -190,14 +260,30 @@ public class Olt
                 .build();
 
 
-        ForwardingObjective upFwd = DefaultForwardingObjective.builder()
+        ForwardingObjective.Builder upFwd = DefaultForwardingObjective.builder()
                 .withFlag(ForwardingObjective.Flag.VERSATILE)
                 .withPriority(1000)
                 .makePermanent()
                 .withSelector(upstream)
                 .fromApp(appId)
-                .withTreatment(upstreamTreatment)
+                .withTreatment(upstreamTreatment);
-                .add(new ObjectiveContext() {
+
+
+        ForwardingObjective.Builder downFwd = DefaultForwardingObjective.builder()
+                .withFlag(ForwardingObjective.Flag.VERSATILE)
+                .withPriority(1000)
+                .makePermanent()
+                .withSelector(downstream)
+                .fromApp(appId)
+                .withTreatment(downstreamTreatment);
+
+        ConnectPoint cp = new ConnectPoint(deviceId, subscriberPort);
+
+        subscribers.put(cp, subscriberVlan);
+        objectives.put(cp, Sets.newHashSet(upFwd, downFwd));
+
+
+        flowObjectiveService.forward(deviceId, upFwd.add(new ObjectiveContext() {
             @Override
             public void onSuccess(Objective objective) {
                 upFuture.complete(null);
@@ -207,16 +293,10 @@ public class Olt
             public void onError(Objective objective, ObjectiveError error) {
                 upFuture.complete(error);
             }
-                });
+        }));
 
-        ForwardingObjective downFwd = DefaultForwardingObjective.builder()
+
-                .withFlag(ForwardingObjective.Flag.VERSATILE)
+        flowObjectiveService.forward(deviceId, downFwd.add(new ObjectiveContext() {
-                .withPriority(1000)
-                .makePermanent()
-                .withSelector(downstream)
-                .fromApp(appId)
-                .withTreatment(downstreamTreatment)
-                .add(new ObjectiveContext() {
             @Override
             public void onSuccess(Objective objective) {
                 downFuture.complete(null);
@@ -226,10 +306,7 @@ public class Olt
             public void onError(Objective objective, ObjectiveError error) {
                 downFuture.complete(error);
             }
-                });
+        }));
-
-        flowObjectiveService.forward(deviceId, upFwd);
-        flowObjectiveService.forward(deviceId, downFwd);
 
         upFuture.thenAcceptBothAsync(downFuture, (upStatus, downStatus) -> {
             if (upStatus == null && downStatus == null) {