Merge branch 'master' of ssh://gerrit.onlab.us:29418/onos-next

@@ -62,6 +62,32 @@
             <groupId>org.osgi</groupId>
             <groupId>org.osgi</groupId>
             <artifactId>org.osgi.core</artifactId>
             <artifactId>org.osgi.core</artifactId>
         </dependency>
         </dependency>
+
+    <dependency>
+      <groupId>org.onlab.onos</groupId>
+      <artifactId>onlab-thirdparty</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.onlab.onos</groupId>
+      <artifactId>onlab-misc</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.onlab.onos</groupId>
+      <artifactId>onlab-junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.onlab.onos</groupId>
+      <artifactId>onos-cli</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.karaf.shell</groupId>
+      <artifactId>org.apache.karaf.shell.console</artifactId>
+    </dependency>
     </dependencies>
     </dependencies>
 
 
     <build>
     <build>
@@ -77,6 +103,7 @@
                             ${project.groupId}.${project.artifactId}
                             ${project.groupId}.${project.artifactId}
                         </Bundle-SymbolicName>
                         </Bundle-SymbolicName>
                         <Import-Package>
                         <Import-Package>
+                            org.slf4j,
                             org.osgi.framework,
                             org.osgi.framework,
                             javax.ws.rs,javax.ws.rs.core,
                             javax.ws.rs,javax.ws.rs.core,
                             com.sun.jersey.api.core,
                             com.sun.jersey.api.core,

@@ -16,43 +16,47 @@
 package org.onlab.onos.calendar;
 package org.onlab.onos.calendar;
 
 
 import java.net.URI;
 import java.net.URI;
-
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.intent.IntentService;
+import org.onlab.rest.BaseResource;
 import javax.ws.rs.POST;
 import javax.ws.rs.POST;
-import javax.ws.rs.Path;
+import javax.ws.rs.DELETE;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response;
-
 import org.onlab.onos.core.ApplicationId;
 import org.onlab.onos.core.ApplicationId;
 import org.onlab.onos.core.CoreService;
 import org.onlab.onos.core.CoreService;
-import org.onlab.onos.net.ConnectPoint;
-import org.onlab.onos.net.DeviceId;
 import org.onlab.onos.net.flow.DefaultTrafficSelector;
 import org.onlab.onos.net.flow.DefaultTrafficSelector;
 import org.onlab.onos.net.flow.TrafficSelector;
 import org.onlab.onos.net.flow.TrafficSelector;
 import org.onlab.onos.net.flow.TrafficTreatment;
 import org.onlab.onos.net.flow.TrafficTreatment;
-import org.onlab.onos.net.intent.Intent;
-import org.onlab.onos.net.intent.IntentService;
-import org.onlab.onos.net.intent.PointToPointIntentWithBandwidthConstraint;
-import org.onlab.onos.net.resource.BandwidthResourceRequest;
+import org.onlab.onos.net.intent.PointToPointIntent;
 import org.onlab.packet.Ethernet;
 import org.onlab.packet.Ethernet;
-import org.onlab.rest.BaseResource;
-
 import static org.onlab.onos.net.PortNumber.portNumber;
 import static org.onlab.onos.net.PortNumber.portNumber;
 import static org.onlab.onos.net.flow.DefaultTrafficTreatment.builder;
 import static org.onlab.onos.net.flow.DefaultTrafficTreatment.builder;
 
 
+import static org.slf4j.LoggerFactory.getLogger;
+import org.slf4j.Logger;
+
 /**
 /**
  * Web resource for triggering calendared intents.
  * Web resource for triggering calendared intents.
  */
  */
-@Path("intent")
+@javax.ws.rs.Path("intent")
 public class BandwidthCalendarResource extends BaseResource {
 public class BandwidthCalendarResource extends BaseResource {
 
 
+    private static final Logger log = getLogger(BandwidthCalendarResource.class);
+
+    @javax.ws.rs.Path("/{src}/{dst}/{srcPort}/{dstPort}/{bandwidth}")
     @POST
     @POST
-    @Path("{src}/{dst}/{srcPort}/{dstPort}/{bandwidth}")
     public Response createIntent(@PathParam("src") String src,
     public Response createIntent(@PathParam("src") String src,
                                  @PathParam("dst") String dst,
                                  @PathParam("dst") String dst,
                                  @PathParam("srcPort") String srcPort,
                                  @PathParam("srcPort") String srcPort,
                                  @PathParam("dstPort") String dstPort,
                                  @PathParam("dstPort") String dstPort,
                                  @PathParam("bandwidth") String bandwidth) {
                                  @PathParam("bandwidth") String bandwidth) {
-        // TODO: implement calls to intent framework
+
+        log.info("Receiving Create Intent request...");
+        log.info("Path Constraints: Src = {} SrcPort = {} Dest = {} DestPort = {} BW = {}",
+                src, srcPort, dst, dstPort, bandwidth);
+
         IntentService service = get(IntentService.class);
         IntentService service = get(IntentService.class);
 
 
         ConnectPoint srcPoint = new ConnectPoint(deviceId(src), portNumber(srcPort));
         ConnectPoint srcPoint = new ConnectPoint(deviceId(src), portNumber(srcPort));
@@ -61,13 +65,38 @@ public class BandwidthCalendarResource extends BaseResource {
         TrafficSelector selector = buildTrafficSelector();
         TrafficSelector selector = buildTrafficSelector();
         TrafficTreatment treatment = builder().build();
         TrafficTreatment treatment = builder().build();
 
 
-        Intent intent = new PointToPointIntentWithBandwidthConstraint(
-                appId(), selector, treatment,
-                srcPoint, dstPoint, new BandwidthResourceRequest(Double.parseDouble(bandwidth)));
-        service.submit(intent);
+        PointToPointIntent intentP2P =
+                        new PointToPointIntent(appId(), selector, treatment,
+                                               srcPoint, dstPoint);
+        service.submit(intentP2P);
+        log.info("Submitted Calendar App intent: src = " + src + "dest = " + dst
+                + "srcPort = " + srcPort + "destPort" + dstPort + "intentID = " + intentP2P.id().toString());
+        String reply =  intentP2P.id().toString() + "\n";
 
 
-        return Response.ok("Yo! We got src=" + srcPoint + "; dst=" + dstPoint +
-                                   "; bw=" + bandwidth + "; intent service " + service).build();
+        return Response.ok(reply).build();
+    }
+
+    @javax.ws.rs.Path("/cancellation/{intentId}")
+    @DELETE
+    public Response withdrawIntent(@PathParam("intentId") String intentId) {
+
+        log.info("Receiving Teardown request...");
+        log.info("Withdraw intentId = {} ", intentId);
+
+        String reply =  "ok\n";
+        return Response.ok(reply).build();
+    }
+
+    @javax.ws.rs.Path("/modification/{intentId}/{bandwidth}")
+    @POST
+    public Response modifyBandwidth(@PathParam("intentId") String intentId,
+                                 @PathParam("bandwidth") String bandwidth) {
+
+        log.info("Receiving Modify request...");
+        log.info("Modify bw for intentId = {} with new bandwidth = {}", intentId, bandwidth);
+
+        String reply =  "ok\n";
+        return Response.ok(reply).build();
     }
     }
 
 
     private TrafficSelector buildTrafficSelector() {
     private TrafficSelector buildTrafficSelector() {
@@ -86,5 +115,4 @@ public class BandwidthCalendarResource extends BaseResource {
     protected ApplicationId appId() {
     protected ApplicationId appId() {
         return get(CoreService.class).registerApplication("org.onlab.onos.calendar");
         return get(CoreService.class).registerApplication("org.onlab.onos.calendar");
     }
     }
-
 }
 }

@@ -188,7 +188,7 @@ public final class DefaultTrafficSelector implements TrafficSelector {
         }
         }
 
 
         @Override
         @Override
-        public Builder matchOpticalSignalType(Byte signalType) {
+        public Builder matchOpticalSignalType(Short signalType) {
             return add(Criteria.matchOpticalSignalType(signalType));
             return add(Criteria.matchOpticalSignalType(signalType));
 
 
         }
         }

@@ -27,11 +27,14 @@ public final class FlowRuleBatchEvent extends AbstractEvent<FlowRuleBatchEvent.T
      */
      */
     public enum Type {
     public enum Type {
 
 
+        // Request has been forwarded to MASTER Node
         /**
         /**
          * Signifies that a batch operation has been initiated.
          * Signifies that a batch operation has been initiated.
          */
          */
         BATCH_OPERATION_REQUESTED,
         BATCH_OPERATION_REQUESTED,
 
 
+        // MASTER Node has pushed the batch down to the Device
+        // (e.g., Received barrier reply)
         /**
         /**
          * Signifies that a batch operation has completed.
          * Signifies that a batch operation has completed.
          */
          */

@@ -25,29 +25,29 @@ import com.google.common.collect.Lists;
 public class FlowRuleBatchRequest {
 public class FlowRuleBatchRequest {
 
 
     private final int batchId;
     private final int batchId;
-    private final List<FlowEntry> toAdd;
-    private final List<FlowEntry> toRemove;
+    private final List<FlowRule> toAdd;
+    private final List<FlowRule> toRemove;
 
 
-    public FlowRuleBatchRequest(int batchId, List<? extends FlowEntry> toAdd, List<? extends FlowEntry> toRemove) {
+    public FlowRuleBatchRequest(int batchId, List<? extends FlowRule> toAdd, List<? extends FlowRule> toRemove) {
         this.batchId = batchId;
         this.batchId = batchId;
         this.toAdd = Collections.unmodifiableList(toAdd);
         this.toAdd = Collections.unmodifiableList(toAdd);
         this.toRemove = Collections.unmodifiableList(toRemove);
         this.toRemove = Collections.unmodifiableList(toRemove);
     }
     }
 
 
-    public List<FlowEntry> toAdd() {
+    public List<FlowRule> toAdd() {
         return toAdd;
         return toAdd;
     }
     }
 
 
-    public List<FlowEntry> toRemove() {
+    public List<FlowRule> toRemove() {
         return toRemove;
         return toRemove;
     }
     }
 
 
     public FlowRuleBatchOperation asBatchOperation() {
     public FlowRuleBatchOperation asBatchOperation() {
         List<FlowRuleBatchEntry> entries = Lists.newArrayList();
         List<FlowRuleBatchEntry> entries = Lists.newArrayList();
-        for (FlowEntry e : toAdd) {
+        for (FlowRule e : toAdd) {
             entries.add(new FlowRuleBatchEntry(FlowRuleOperation.ADD, e));
             entries.add(new FlowRuleBatchEntry(FlowRuleOperation.ADD, e));
         }
         }
-        for (FlowEntry e : toRemove) {
+        for (FlowRule e : toRemove) {
             entries.add(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, e));
             entries.add(new FlowRuleBatchEntry(FlowRuleOperation.REMOVE, e));
         }
         }
         return new FlowRuleBatchOperation(entries);
         return new FlowRuleBatchOperation(entries);

@@ -147,7 +147,7 @@ public interface TrafficSelector {
          * @param signalType
          * @param signalType
          * @return a selection builder
          * @return a selection builder
          */
          */
-        public Builder matchOpticalSignalType(Byte signalType);
+        public Builder matchOpticalSignalType(Short signalType);
 
 
         /**
         /**
          * Builds an immutable traffic selector.
          * Builds an immutable traffic selector.

@@ -161,11 +161,11 @@ public final class Criteria {
     /**
     /**
      * Creates a match on lambda field using the specified value.
      * Creates a match on lambda field using the specified value.
      *
      *
-     * @param lambda
+     * @param sigType
      * @return match criterion
      * @return match criterion
      */
      */
-    public static Criterion matchOpticalSignalType(Byte lambda) {
-        return new OpticalSignalTypeCriterion(lambda, Type.OCH_SIGTYPE);
+    public static Criterion matchOpticalSignalType(Short sigType) {
+        return new OpticalSignalTypeCriterion(sigType, Type.OCH_SIGTYPE);
     }
     }
 
 
 
 
@@ -587,10 +587,10 @@ public final class Criteria {
 
 
     public static final class OpticalSignalTypeCriterion implements Criterion {
     public static final class OpticalSignalTypeCriterion implements Criterion {
 
 
-        private final byte signalType;
+        private final Short signalType;
         private final Type type;
         private final Type type;
 
 
-        public OpticalSignalTypeCriterion(byte signalType, Type type) {
+        public OpticalSignalTypeCriterion(Short signalType, Type type) {
             this.signalType = signalType;
             this.signalType = signalType;
             this.type = type;
             this.type = type;
         }
         }
@@ -600,7 +600,7 @@ public final class Criteria {
             return this.type;
             return this.type;
         }
         }
 
 
-        public Byte signalType() {
+        public Short signalType() {
             return this.signalType;
             return this.signalType;
         }
         }
 
 

@@ -15,7 +15,11 @@
  */
  */
 package org.onlab.onos.net.intent;
 package org.onlab.onos.net.intent;
 
 
+import java.util.List;
+
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+
 import org.onlab.onos.core.ApplicationId;
 import org.onlab.onos.core.ApplicationId;
 import org.onlab.onos.net.Path;
 import org.onlab.onos.net.Path;
 import org.onlab.onos.net.flow.TrafficSelector;
 import org.onlab.onos.net.flow.TrafficSelector;
@@ -28,7 +32,7 @@ import org.onlab.onos.net.resource.LinkResourceRequest;
 public class PathIntent extends ConnectivityIntent {
 public class PathIntent extends ConnectivityIntent {
 
 
     private final Path path;
     private final Path path;
-    private final LinkResourceRequest[] resourceRequests;
+    private final List<LinkResourceRequest> resourceRequests;
 
 
     /**
     /**
      * Creates a new point-to-point intent with the supplied ingress/egress
      * Creates a new point-to-point intent with the supplied ingress/egress
@@ -45,7 +49,7 @@ public class PathIntent extends ConnectivityIntent {
         super(id(PathIntent.class, selector, treatment, path), appId,
         super(id(PathIntent.class, selector, treatment, path), appId,
               resources(path.links()), selector, treatment);
               resources(path.links()), selector, treatment);
         this.path = path;
         this.path = path;
-        this.resourceRequests = resourceRequests;
+        this.resourceRequests = ImmutableList.copyOf(resourceRequests);
     }
     }
 
 
     /**
     /**
@@ -54,7 +58,7 @@ public class PathIntent extends ConnectivityIntent {
     protected PathIntent() {
     protected PathIntent() {
         super();
         super();
         this.path = null;
         this.path = null;
-        this.resourceRequests = new LinkResourceRequest[0];
+        this.resourceRequests = ImmutableList.of();
     }
     }
 
 
     /**
     /**
@@ -71,8 +75,9 @@ public class PathIntent extends ConnectivityIntent {
         return true;
         return true;
     }
     }
 
 
+    // TODO: consider changing return type
     public LinkResourceRequest[] resourceRequests() {
     public LinkResourceRequest[] resourceRequests() {
-        return resourceRequests;
+        return resourceRequests.toArray(new LinkResourceRequest[resourceRequests.size()]);
     }
     }
 
 
     @Override
     @Override

@@ -371,10 +371,11 @@ public class FlowRuleManager
             final FlowRuleBatchRequest request = event.subject();
             final FlowRuleBatchRequest request = event.subject();
             switch (event.type()) {
             switch (event.type()) {
             case BATCH_OPERATION_REQUESTED:
             case BATCH_OPERATION_REQUESTED:
-                for (FlowEntry entry : request.toAdd()) {
+                // Request has been forwarded to MASTER Node, and was
+                for (FlowRule entry : request.toAdd()) {
                     eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADD_REQUESTED, entry));
                     eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADD_REQUESTED, entry));
                 }
                 }
-                for (FlowEntry entry : request.toRemove()) {
+                for (FlowRule entry : request.toRemove()) {
                     eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVE_REQUESTED, entry));
                     eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVE_REQUESTED, entry));
                 }
                 }
                 // FIXME: what about op.equals(FlowRuleOperation.MODIFY) ?
                 // FIXME: what about op.equals(FlowRuleOperation.MODIFY) ?
@@ -392,21 +393,15 @@ public class FlowRuleManager
                                                                                   Futures.getUnchecked(result)));
                                                                                   Futures.getUnchecked(result)));
                     }
                     }
                 }, futureListeners);
                 }, futureListeners);
-
                 break;
                 break;
+
             case BATCH_OPERATION_COMPLETED:
             case BATCH_OPERATION_COMPLETED:
-                Set<FlowRule> failedItems = event.result().failedItems();
-                for (FlowEntry entry : request.toAdd()) {
-                    if (!failedItems.contains(entry)) {
-                        eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_ADDED, entry));
-                    }
-                }
-                for (FlowEntry entry : request.toRemove()) {
-                    if (!failedItems.contains(entry)) {
-                            eventDispatcher.post(new FlowRuleEvent(FlowRuleEvent.Type.RULE_REMOVED, entry));
-                    }
-                }
+                // MASTER Node has pushed the batch down to the Device
+
+                // Note: RULE_ADDED will be posted
+                // when Flow was actually confirmed by stats reply.
                 break;
                 break;
+
             default:
             default:
                 break;
                 break;
             }
             }

@@ -79,6 +79,7 @@ public class OpticalPathIntentInstaller implements IntentInstaller<OpticalPathIn
     private ApplicationId appId;
     private ApplicationId appId;
 
 
     //final short WAVELENGTH = 80;
     //final short WAVELENGTH = 80;
+    static final short SIGNAL_TYPE = (short) 1;
 
 
     @Activate
     @Activate
     public void activate() {
     public void activate() {
@@ -151,7 +152,9 @@ public class OpticalPathIntentInstaller implements IntentInstaller<OpticalPathIn
 
 
             prev = link.dst();
             prev = link.dst();
             selectorBuilder.matchInport(link.dst().port());
             selectorBuilder.matchInport(link.dst().port());
+            selectorBuilder.matchOpticalSignalType(SIGNAL_TYPE); //todo
             selectorBuilder.matchLambda((short) la.toInt());
             selectorBuilder.matchLambda((short) la.toInt());
+
         }
         }
 
 
         // build the last T port rule
         // build the last T port rule

@@ -148,7 +148,7 @@ public class FlowRuleManagerTest {
         int i = 0;
         int i = 0;
         System.err.println("events :" + listener.events);
         System.err.println("events :" + listener.events);
         for (FlowRuleEvent e : listener.events) {
         for (FlowRuleEvent e : listener.events) {
-            assertTrue("unexpected event", e.type().equals(events[i]));
+            assertEquals("unexpected event", events[i], e.type());
             i++;
             i++;
         }
         }
 
 
@@ -178,15 +178,13 @@ public class FlowRuleManagerTest {
                        RULE_ADDED, RULE_ADDED);
                        RULE_ADDED, RULE_ADDED);
 
 
         addFlowRule(1);
         addFlowRule(1);
+        System.err.println("events :" + listener.events);
         assertEquals("should still be 2 rules", 2, flowCount());
         assertEquals("should still be 2 rules", 2, flowCount());
 
 
         providerService.pushFlowMetrics(DID, ImmutableList.of(fe1));
         providerService.pushFlowMetrics(DID, ImmutableList.of(fe1));
         validateEvents(RULE_UPDATED);
         validateEvents(RULE_UPDATED);
     }
     }
 
 
-
-    // TODO: If preserving iteration order is a requirement, redo FlowRuleStore.
-    //backing store is sensitive to the order of additions/removals
     private boolean validateState(Map<FlowRule, FlowEntryState> expected) {
     private boolean validateState(Map<FlowRule, FlowEntryState> expected) {
         Map<FlowRule, FlowEntryState> expectedToCheck = new HashMap<>(expected);
         Map<FlowRule, FlowEntryState> expectedToCheck = new HashMap<>(expected);
         Iterable<FlowEntry> rules = service.getFlowEntries(DID);
         Iterable<FlowEntry> rules = service.getFlowEntries(DID);
@@ -539,17 +537,17 @@ public class FlowRuleManagerTest {
 
 
             @Override
             @Override
             public boolean cancel(boolean mayInterruptIfRunning) {
             public boolean cancel(boolean mayInterruptIfRunning) {
-                return true;
+                return false;
             }
             }
 
 
             @Override
             @Override
             public boolean isCancelled() {
             public boolean isCancelled() {
-                return true;
+                return false;
             }
             }
 
 
             @Override
             @Override
             public boolean isDone() {
             public boolean isDone() {
-                return false;
+                return true;
             }
             }
 
 
             @Override
             @Override
@@ -562,12 +560,14 @@ public class FlowRuleManagerTest {
             public CompletedBatchOperation get(long timeout, TimeUnit unit)
             public CompletedBatchOperation get(long timeout, TimeUnit unit)
                     throws InterruptedException,
                     throws InterruptedException,
                     ExecutionException, TimeoutException {
                     ExecutionException, TimeoutException {
-                return null;
+                return new CompletedBatchOperation(true, Collections.<FlowRule>emptySet());
             }
             }
 
 
             @Override
             @Override
             public void addListener(Runnable task, Executor executor) {
             public void addListener(Runnable task, Executor executor) {
-                // TODO: add stuff.
+                if (isDone()) {
+                    executor.execute(task);
+                }
             }
             }
         }
         }
 
 

@@ -447,7 +447,13 @@ implements MastershipStore {
             RoleValue oldValue = event.getOldValue();
             RoleValue oldValue = event.getOldValue();
             RoleValue newValue = event.getValue();
             RoleValue newValue = event.getValue();
 
 
-            if (Objects.equal(oldValue.get(MASTER), newValue.get(MASTER))) {
+            NodeId oldMaster = null;
+            if (oldValue != null) {
+                oldMaster = oldValue.get(MASTER);
+            }
+            NodeId newMaster = newValue.get(MASTER);
+
+            if (Objects.equal(oldMaster, newMaster)) {
                 notifyDelegate(new MastershipEvent(
                 notifyDelegate(new MastershipEvent(
                         MASTER_CHANGED, event.getKey(), event.getValue().roleInfo()));
                         MASTER_CHANGED, event.getKey(), event.getValue().roleInfo()));
             } else {
             } else {

@@ -16,8 +16,12 @@
 package org.onlab.onos.store.trivial.impl;
 package org.onlab.onos.store.trivial.impl;
 
 
 import com.google.common.base.Function;
 import com.google.common.base.Function;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.FluentIterable;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.SettableFuture;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -43,13 +47,15 @@ import org.onlab.onos.store.AbstractStore;
 import org.onlab.util.NewConcurrentHashMap;
 import org.onlab.util.NewConcurrentHashMap;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Collections;
 import java.util.List;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.Future;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 
 import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
 import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
 import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
 import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
@@ -72,6 +78,18 @@ public class SimpleFlowRuleStore
     private final ConcurrentMap<DeviceId, ConcurrentMap<FlowId, List<StoredFlowEntry>>>
     private final ConcurrentMap<DeviceId, ConcurrentMap<FlowId, List<StoredFlowEntry>>>
             flowEntries = new ConcurrentHashMap<>();
             flowEntries = new ConcurrentHashMap<>();
 
 
+    private final AtomicInteger localBatchIdGen = new AtomicInteger();
+
+    // TODO: make this configurable
+    private int pendingFutureTimeoutMinutes = 5;
+
+    private Cache<Integer, SettableFuture<CompletedBatchOperation>> pendingFutures =
+            CacheBuilder.newBuilder()
+                .expireAfterWrite(pendingFutureTimeoutMinutes, TimeUnit.MINUTES)
+                // TODO Explicitly fail the future if expired?
+                //.removalListener(listener)
+                .build();
+
     @Activate
     @Activate
     public void activate() {
     public void activate() {
         log.info("Started");
         log.info("Started");
@@ -173,10 +191,6 @@ public class SimpleFlowRuleStore
             }
             }
             // new flow rule added
             // new flow rule added
             existing.add(f);
             existing.add(f);
-            notifyDelegate(FlowRuleBatchEvent.requested(
-                    new FlowRuleBatchRequest(1, /* FIXME generate something */
-                                             Arrays.<FlowEntry>asList(f),
-                                             Collections.<FlowEntry>emptyList())));
         }
         }
     }
     }
 
 
@@ -190,11 +204,6 @@ public class SimpleFlowRuleStore
                 if (entry.equals(rule)) {
                 if (entry.equals(rule)) {
                     synchronized (entry) {
                     synchronized (entry) {
                         entry.setState(FlowEntryState.PENDING_REMOVE);
                         entry.setState(FlowEntryState.PENDING_REMOVE);
-                        // TODO: Should we notify only if it's "remote" event?
-                        notifyDelegate(FlowRuleBatchEvent.requested(
-                                new FlowRuleBatchRequest(1, /* FIXME generate something */
-                                                         Collections.<FlowEntry>emptyList(),
-                                                         Arrays.<FlowEntry>asList(entry))));
                     }
                     }
                 }
                 }
             }
             }
@@ -251,20 +260,47 @@ public class SimpleFlowRuleStore
     @Override
     @Override
     public Future<CompletedBatchOperation> storeBatch(
     public Future<CompletedBatchOperation> storeBatch(
             FlowRuleBatchOperation batchOperation) {
             FlowRuleBatchOperation batchOperation) {
+        List<FlowRule> toAdd = new ArrayList<>();
+        List<FlowRule> toRemove = new ArrayList<>();
         for (FlowRuleBatchEntry entry : batchOperation.getOperations()) {
         for (FlowRuleBatchEntry entry : batchOperation.getOperations()) {
+            final FlowRule flowRule = entry.getTarget();
             if (entry.getOperator().equals(FlowRuleOperation.ADD)) {
             if (entry.getOperator().equals(FlowRuleOperation.ADD)) {
-                storeFlowRule(entry.getTarget());
+                if (!getFlowEntries(flowRule.deviceId(), flowRule.id()).contains(flowRule)) {
+                    storeFlowRule(flowRule);
+                    toAdd.add(flowRule);
+                }
             } else if (entry.getOperator().equals(FlowRuleOperation.REMOVE)) {
             } else if (entry.getOperator().equals(FlowRuleOperation.REMOVE)) {
-                deleteFlowRule(entry.getTarget());
+                if (getFlowEntries(flowRule.deviceId(), flowRule.id()).contains(flowRule)) {
+                    deleteFlowRule(flowRule);
+                    toRemove.add(flowRule);
+                }
             } else {
             } else {
                 throw new UnsupportedOperationException("Unsupported operation type");
                 throw new UnsupportedOperationException("Unsupported operation type");
             }
             }
         }
         }
-        return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowEntry>emptySet()));
+
+        if (toAdd.isEmpty() && toRemove.isEmpty()) {
+            return Futures.immediateFuture(new CompletedBatchOperation(true, Collections.<FlowRule>emptySet()));
+        }
+
+        SettableFuture<CompletedBatchOperation> r = SettableFuture.create();
+        final int batchId = localBatchIdGen.incrementAndGet();
+
+        pendingFutures.put(batchId, r);
+        notifyDelegate(FlowRuleBatchEvent.requested(new FlowRuleBatchRequest(batchId, toAdd, toRemove)));
+
+        return r;
     }
     }
 
 
     @Override
     @Override
     public void batchOperationComplete(FlowRuleBatchEvent event) {
     public void batchOperationComplete(FlowRuleBatchEvent event) {
+        final Integer batchId = event.subject().batchId();
+        SettableFuture<CompletedBatchOperation> future
+            = pendingFutures.getIfPresent(batchId);
+        if (future != null) {
+            future.set(event.result());
+            pendingFutures.invalidate(batchId);
+        }
         notifyDelegate(event);
         notifyDelegate(event);
     }
     }
 }
 }

@@ -352,13 +352,6 @@
     </dependencies>
     </dependencies>
 
 
     <build>
     <build>
-        <extensions>
-            <extension>
-                <groupId>kr.motd.maven</groupId>
-                <artifactId>os-maven-plugin</artifactId>
-                <version>1.2.3.Final</version>
-            </extension>
-        </extensions>
         <pluginManagement>
         <pluginManagement>
             <plugins>
             <plugins>
                 <plugin>
                 <plugin>

@@ -289,7 +289,10 @@ public class FlowEntryBuilder {
             case OCH_SIGID:
             case OCH_SIGID:
                 builder.matchLambda(match.get(MatchField.OCH_SIGID).getChannelNumber());
                 builder.matchLambda(match.get(MatchField.OCH_SIGID).getChannelNumber());
                 break;
                 break;
-            case OCH_SIGTYPE_BASIC:
+            case OCH_SIGTYPE:
+                builder.matchOpticalSignalType(match.get(MatchField
+                                                                 .OCH_SIGTYPE).getValue());
+                break;
             case ARP_OP:
             case ARP_OP:
             case ARP_SHA:
             case ARP_SHA:
             case ARP_SPA:
             case ARP_SPA:

@@ -19,6 +19,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 
 
 import org.onlab.onos.net.flow.FlowRule;
 import org.onlab.onos.net.flow.FlowRule;
 import org.onlab.onos.net.flow.TrafficSelector;
 import org.onlab.onos.net.flow.TrafficSelector;
+import org.onlab.onos.net.flow.criteria.Criteria;
 import org.onlab.onos.net.flow.criteria.Criteria.EthCriterion;
 import org.onlab.onos.net.flow.criteria.Criteria.EthCriterion;
 import org.onlab.onos.net.flow.criteria.Criteria.EthTypeCriterion;
 import org.onlab.onos.net.flow.criteria.Criteria.EthTypeCriterion;
 import org.onlab.onos.net.flow.criteria.Criteria.IPCriterion;
 import org.onlab.onos.net.flow.criteria.Criteria.IPCriterion;
@@ -46,6 +47,7 @@ import org.projectfloodlight.openflow.types.Masked;
 import org.projectfloodlight.openflow.types.OFPort;
 import org.projectfloodlight.openflow.types.OFPort;
 import org.projectfloodlight.openflow.types.OFVlanVidMatch;
 import org.projectfloodlight.openflow.types.OFVlanVidMatch;
 import org.projectfloodlight.openflow.types.TransportPort;
 import org.projectfloodlight.openflow.types.TransportPort;
+import org.projectfloodlight.openflow.types.U8;
 import org.projectfloodlight.openflow.types.VlanPcp;
 import org.projectfloodlight.openflow.types.VlanPcp;
 import org.projectfloodlight.openflow.types.VlanVid;
 import org.projectfloodlight.openflow.types.VlanVid;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
@@ -197,6 +199,12 @@ public abstract class FlowModBuilder {
                 mBuilder.setExact(MatchField.OCH_SIGID,
                 mBuilder.setExact(MatchField.OCH_SIGID,
                         new CircuitSignalID((byte) 1, (byte) 2, lc.lambda(), (short) 1));
                         new CircuitSignalID((byte) 1, (byte) 2, lc.lambda(), (short) 1));
                 break;
                 break;
+            case OCH_SIGTYPE:
+                Criteria.OpticalSignalTypeCriterion sc =
+                        (Criteria.OpticalSignalTypeCriterion) c;
+                mBuilder.setExact(MatchField.OCH_SIGTYPE,
+                                  U8.of(sc.signalType()));
+                break;
             case ARP_OP:
             case ARP_OP:
             case ARP_SHA:
             case ARP_SHA:
             case ARP_SPA:
             case ARP_SPA:

@@ -30,7 +30,6 @@ import static com.google.common.base.Preconditions.checkState;
 
 
 /**
 /**
  * A class representing an IP address.
  * A class representing an IP address.
- * TODO: Add support for IPv6 as well.
  */
  */
 public final class IpAddress implements Comparable<IpAddress> {
 public final class IpAddress implements Comparable<IpAddress> {
     // IP Versions
     // IP Versions

@@ -17,8 +17,6 @@ package org.onlab.packet;
 
 
 import java.util.Objects;
 import java.util.Objects;
 
 
-// TODO: Add support for IPv6 as well.
-
 /**
 /**
  * A class representing an IP prefix. A prefix consists of an IP address and
  * A class representing an IP prefix. A prefix consists of an IP address and
  * a subnet mask.
  * a subnet mask.
@@ -40,26 +38,39 @@ public final class IpPrefix {
      *
      *
      * @param address the IP address
      * @param address the IP address
      * @param prefixLength the prefix length
      * @param prefixLength the prefix length
+     * @throws IllegalArgumentException if the prefix length value is invalid
      */
      */
     private IpPrefix(IpAddress address, int prefixLength) {
     private IpPrefix(IpAddress address, int prefixLength) {
-        checkPrefixLength(prefixLength);
+        checkPrefixLength(address.version(), prefixLength);
         this.address = IpAddress.makeMaskedAddress(address, prefixLength);
         this.address = IpAddress.makeMaskedAddress(address, prefixLength);
         this.prefixLength = (short) prefixLength;
         this.prefixLength = (short) prefixLength;
     }
     }
 
 
     /**
     /**
-     * Checks whether the prefix length is valid.
+     * Returns the IP version of the prefix.
      *
      *
-     * @param prefixLength the prefix length value to check
-     * @throws IllegalArgumentException if the prefix length value is invalid
+     * @return the IP version of the prefix
      */
      */
-    private static void checkPrefixLength(int prefixLength) {
-        if ((prefixLength < 0) || (prefixLength > MAX_INET_MASK_LENGTH)) {
-            String msg = "Invalid prefix length " + prefixLength + ". " +
-                "The value must be in the interval [0, " +
-                MAX_INET_MASK_LENGTH + "]";
-            throw new IllegalArgumentException(msg);
+    public IpAddress.Version version() {
+        return address.version();
+    }
+
+    /**
+     * Returns the IP address value of the prefix.
+     *
+     * @return the IP address value of the prefix
+     */
+    public IpAddress address() {
+        return address;
     }
     }
+
+    /**
+     * Returns the IP address prefix length.
+     *
+     * @return the IP address prefix length
+     */
+    public int prefixLength() {
+        return prefixLength;
     }
     }
 
 
     /**
     /**
@@ -68,6 +79,7 @@ public final class IpPrefix {
      * @param address an integer representing the IPv4 address
      * @param address an integer representing the IPv4 address
      * @param prefixLength the prefix length
      * @param prefixLength the prefix length
      * @return an IP prefix
      * @return an IP prefix
+     * @throws IllegalArgumentException if the prefix length value is invalid
      */
      */
     public static IpPrefix valueOf(int address, int prefixLength) {
     public static IpPrefix valueOf(int address, int prefixLength) {
         return new IpPrefix(IpAddress.valueOf(address), prefixLength);
         return new IpPrefix(IpAddress.valueOf(address), prefixLength);
@@ -80,11 +92,11 @@ public final class IpPrefix {
      * @param address the IP address value stored in network byte order
      * @param address the IP address value stored in network byte order
      * @param prefixLength the prefix length
      * @param prefixLength the prefix length
      * @return an IP prefix
      * @return an IP prefix
+     * @throws IllegalArgumentException if the prefix length value is invalid
      */
      */
     public static IpPrefix valueOf(IpAddress.Version version, byte[] address,
     public static IpPrefix valueOf(IpAddress.Version version, byte[] address,
                                    int prefixLength) {
                                    int prefixLength) {
-        return new IpPrefix(IpAddress.valueOf(version, address),
-                            prefixLength);
+        return new IpPrefix(IpAddress.valueOf(version, address), prefixLength);
     }
     }
 
 
     /**
     /**
@@ -93,6 +105,7 @@ public final class IpPrefix {
      * @param address the IP address
      * @param address the IP address
      * @param prefixLength the prefix length
      * @param prefixLength the prefix length
      * @return an IP prefix
      * @return an IP prefix
+     * @throws IllegalArgumentException if the prefix length value is invalid
      */
      */
     public static IpPrefix valueOf(IpAddress address, int prefixLength) {
     public static IpPrefix valueOf(IpAddress address, int prefixLength) {
         return new IpPrefix(address, prefixLength);
         return new IpPrefix(address, prefixLength);
@@ -104,6 +117,7 @@ public final class IpPrefix {
      *
      *
      * @param address an IP prefix in string form, e.g. "10.1.0.0/16"
      * @param address an IP prefix in string form, e.g. "10.1.0.0/16"
      * @return an IP prefix
      * @return an IP prefix
+     * @throws IllegalArgumentException if the arguments are invalid
      */
      */
     public static IpPrefix valueOf(String address) {
     public static IpPrefix valueOf(String address) {
         final String[] parts = address.split("/");
         final String[] parts = address.split("/");
@@ -119,33 +133,6 @@ public final class IpPrefix {
     }
     }
 
 
     /**
     /**
-     * Returns the IP version of the prefix.
-     *
-     * @return the IP version of the prefix
-     */
-    public IpAddress.Version version() {
-        return address.version();
-    }
-
-    /**
-     * Returns the IP address value of the prefix.
-     *
-     * @return the IP address value of the prefix
-     */
-    public IpAddress address() {
-        return address;
-    }
-
-    /**
-     * Returns the IP address prefix length.
-     *
-     * @return the IP address prefix length
-     */
-    public int prefixLength() {
-        return prefixLength;
-    }
-
-    /**
      * Determines whether a given IP prefix is contained within this prefix.
      * Determines whether a given IP prefix is contained within this prefix.
      *
      *
      * @param other the IP prefix to test
      * @param other the IP prefix to test
@@ -217,4 +204,35 @@ public final class IpPrefix {
         builder.append(String.format("%d", prefixLength));
         builder.append(String.format("%d", prefixLength));
         return builder.toString();
         return builder.toString();
     }
     }
+
+    /**
+     * Checks whether the prefix length is valid.
+     *
+     * @param version the IP address version
+     * @param prefixLength the prefix length value to check
+     * @throws IllegalArgumentException if the prefix length value is invalid
+     */
+    private static void checkPrefixLength(IpAddress.Version version,
+                                          int prefixLength) {
+        int maxPrefixLen = 0;
+
+        switch (version) {
+        case INET:
+            maxPrefixLen = MAX_INET_MASK_LENGTH;
+            break;
+        case INET6:
+            maxPrefixLen = MAX_INET6_MASK_LENGTH;
+            break;
+        default:
+            String msg = "Invalid IP version " + version;
+            throw new IllegalArgumentException(msg);
+        }
+
+        if ((prefixLength < 0) || (prefixLength > maxPrefixLen)) {
+            String msg = "Invalid prefix length " + prefixLength + ". " +
+                "The value must be in the interval [0, " +
+                maxPrefixLen + "]";
+            throw new IllegalArgumentException(msg);
+        }
+    }
 }
 }

@@ -135,7 +135,7 @@ public class IpAddressTest {
      * Tests returning an IPv4 address asn an integer.
      * Tests returning an IPv4 address asn an integer.
      */
      */
     @Test
     @Test
-    public void testToint() {
+    public void testToInt() {
         IpAddress ipAddress;
         IpAddress ipAddress;
 
 
         ipAddress = IpAddress.valueOf("1.2.3.4");
         ipAddress = IpAddress.valueOf("1.2.3.4");
@@ -149,10 +149,10 @@ public class IpAddressTest {
     }
     }
 
 
     /**
     /**
-     * Tests valueOf() converter for an integer value.
+     * Tests valueOf() converter for IPv4 integer value.
      */
      */
     @Test
     @Test
-    public void testValueOfForInteger() {
+    public void testValueOfForIntegerIPv4() {
         IpAddress ipAddress;
         IpAddress ipAddress;
 
 
         ipAddress = IpAddress.valueOf(0x01020304);
         ipAddress = IpAddress.valueOf(0x01020304);