Adding Initial Flows on switch for performance improvement.

These sends minimum packets(ARP and LLDP) to controller for
learning addresses and programming flows as l2switch requires.

Change-Id: Id7f0ba5ca7dea6e83b02c4a8a997bcc2bade05e0
Signed-off-by: Amit Mandke <ammandke@cisco.com>

diff --git a/l2switch-main/pom.xml b/l2switch-main/pom.xml
index 5572604677ef0c151f53f3ba74859978f7265229..3dfbe3524a6f12582d8d07d7f511b680e3885e2d 100644 (file)
--- a/l2switch-main/pom.xml
+++ b/l2switch-main/pom.xml
@@ -61,6 +61,11 @@
       <artifactId>junit</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-all</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

diff --git a/l2switch-main/src/main/java/org/opendaylight/l2switch/L2SwitchProvider.java b/l2switch-main/src/main/java/org/opendaylight/l2switch/L2SwitchProvider.java
index 722c5c5d00f4789327f28c74d2fc2bae18c3ca3e..4de284c438c7b528671189741a05f1520a767c38 100644 (file)
--- a/l2switch-main/src/main/java/org/opendaylight/l2switch/L2SwitchProvider.java
+++ b/l2switch-main/src/main/java/org/opendaylight/l2switch/L2SwitchProvider.java
@@ -13,6 +13,7 @@ import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
 import org.opendaylight.controller.sal.binding.api.NotificationService;
 import org.opendaylight.l2switch.flow.FlowWriterService;
 import org.opendaylight.l2switch.flow.FlowWriterServiceImpl;
+import org.opendaylight.l2switch.flow.InitialFlowWriter;
 import org.opendaylight.l2switch.inventory.InventoryReader;
 import org.opendaylight.l2switch.packet.ArpPacketHandler;
 import org.opendaylight.l2switch.packet.PacketDispatcher;
@@ -29,7 +30,7 @@ public class L2SwitchProvider extends AbstractBindingAwareConsumer
     implements AutoCloseable {
 
   private final static Logger _logger = LoggerFactory.getLogger(L2SwitchProvider.class);
-  private Registration listenerRegistration;
+  private Registration listenerRegistration,invListenerReg;
 
   /**
    * Setup the L2Switch.
@@ -40,7 +41,11 @@ public class L2SwitchProvider extends AbstractBindingAwareConsumer
   public void onSessionInitialized(BindingAwareBroker.ConsumerContext consumerContext) {
     // Setup FlowWriterService
     DataBroker dataService = consumerContext.<DataBroker>getSALService(DataBroker.class);
-    FlowWriterService flowWriterService = new FlowWriterServiceImpl(consumerContext.getRpcService(SalFlowService.class));
+    NotificationService notificationService = consumerContext.<NotificationService>getSALService(NotificationService.class);
+    SalFlowService  salFlowService = consumerContext.getRpcService(SalFlowService.class);
+
+
+    FlowWriterService flowWriterService = new FlowWriterServiceImpl(salFlowService);
 
     // Setup InventoryReader
     InventoryReader inventoryReader = new InventoryReader(dataService);
@@ -58,8 +63,12 @@ public class L2SwitchProvider extends AbstractBindingAwareConsumer
 
 
     // Register ArpPacketHandler
-    NotificationService notificationService = consumerContext.<NotificationService>getSALService(NotificationService.class);
     this.listenerRegistration = notificationService.registerNotificationListener(arpPacketHandler);
+
+    //Write initial flows
+    InitialFlowWriter initialFlowWriter = new InitialFlowWriter(salFlowService);
+    //initialFlowWriter.registerAsNodeDataListener(dataService);
+    invListenerReg = notificationService.registerNotificationListener(initialFlowWriter);
   }
 
   /**
@@ -70,5 +79,8 @@ public class L2SwitchProvider extends AbstractBindingAwareConsumer
     if(listenerRegistration != null) {
       listenerRegistration.close();
     }
+    if(invListenerReg!=null) {
+      invListenerReg.close();
+    }
   }
 }

diff --git a/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/FlowWriterServiceImpl.java b/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/FlowWriterServiceImpl.java
index 2be3d21b611803f951ecef51d810a5811cb6324d..1ce87f1d6eba3386591938632ae1806d8169eed2 100644 (file)
--- a/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/FlowWriterServiceImpl.java
+++ b/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/FlowWriterServiceImpl.java
@@ -101,7 +101,7 @@ public class FlowWriterServiceImpl implements FlowWriterService {
     InstanceIdentifier<Flow> flowPath = buildFlowPath(destNodeConnectorRef, flowTableKey);
 
     // build a flow that target given mac id
-    Flow flowBody = createMacToMacFlow(flowTableKey.getId(), 0, sourceMac, destMac, destNodeConnectorRef);
+    Flow flowBody = createMacToMacFlow(flowTableKey.getId(), 10, sourceMac, destMac, destNodeConnectorRef);
 
     // commit the flow in config data
     writeFlowToConfigData(flowPath, flowBody);

diff --git a/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/InitialFlowWriter.java b/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/InitialFlowWriter.java
new file mode 100644 (file)
index 0000000..c8a51ae
--- /dev/null
+++ b/l2switch-main/src/main/java/org/opendaylight/l2switch/flow/InitialFlowWriter.java
@@ -0,0 +1,324 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.l2switch.flow;
+
+import com.google.common.collect.ImmutableList;
+import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.Uri;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.DropActionCaseBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.OutputActionCaseBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.output.action._case.OutputActionBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.ActionBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.ActionKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.TableKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowTableRef;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowCookie;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowModFlags;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.FlowRef;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.OutputPortValues;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.InstructionsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.Match;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.MatchBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.ApplyActionsCaseBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.apply.actions._case.ApplyActions;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.apply.actions._case.ApplyActionsBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.InstructionBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRemoved;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorUpdated;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemoved;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.OpendaylightInventoryListener;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.l2.types.rev130827.EtherType;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.model.match.types.rev131026.ethernet.match.fields.EthernetTypeBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.model.match.types.rev131026.match.EthernetMatchBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.ethernet.rev140528.KnownEtherType;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Adds default flows on all switches. Registers as ODL Inventory listener so that
+ * it can add flows once a new node i.e. switch is added
+ */
+public class InitialFlowWriter implements OpendaylightInventoryListener {
+  private final Logger _logger = LoggerFactory.getLogger(InitialFlowWriter.class);
+
+  private final ExecutorService initialFlowExecutor = Executors.newCachedThreadPool();
+  private final SalFlowService salFlowService;
+  private final short FLOW_TABLE_ID = 0;//TODO:hard coded to 0 may need change if multiple tables are used.
+
+  private AtomicLong flowIdInc = new AtomicLong();
+  private AtomicLong flowCookieInc = new AtomicLong(0x2b00000000000000L);
+
+
+  public InitialFlowWriter(SalFlowService salFlowService) {
+    this.salFlowService = salFlowService;
+  }
+
+  @Override
+  public void onNodeConnectorRemoved(NodeConnectorRemoved nodeConnectorRemoved) {
+    //do nothing
+  }
+
+  @Override
+  public void onNodeConnectorUpdated(NodeConnectorUpdated nodeConnectorUpdated) {
+    //do nothing
+  }
+
+  @Override
+  public void onNodeRemoved(NodeRemoved nodeRemoved) {
+    //do nothing
+  }
+
+  @Override
+  public void onNodeUpdated(NodeUpdated nodeUpdated) {
+    initialFlowExecutor.submit(new InitialFlowWriterProcessor(nodeUpdated));
+    _logger.info("Adding initial flows to {}", nodeUpdated.getNodeRef().getValue());
+  }
+
+  /**
+   * A private class to process the node updated event in separate thread. Allows to release the
+   * thread that invoked the data node updated event. Avoids any thread lock it may cause.
+   */
+  private class InitialFlowWriterProcessor implements Runnable {
+    private NodeUpdated nodeUpdated;
+
+    public InitialFlowWriterProcessor(NodeUpdated nodeUpdated) {
+      this.nodeUpdated = nodeUpdated;
+    }
+
+    @Override
+    public void run() {
+
+      if(nodeUpdated == null) {
+        return;
+      }
+
+      addInitialFlows((InstanceIdentifier<Node>) nodeUpdated.getNodeRef().getValue());
+
+    }
+
+    public void addInitialFlows(InstanceIdentifier<Node> nodeId) {
+      _logger.info("adding initial flows for node {} ", nodeId);
+
+      InstanceIdentifier<Table> tableId = getTableInstanceId(nodeId);
+      InstanceIdentifier<Flow> flowId = getFlowInstanceId(tableId);
+
+      //add drop all flow
+      writeFlowToController(nodeId, tableId, flowId, createDropAllFlow(FLOW_TABLE_ID, 0));
+
+      //add lldpToController flow
+      flowId = getFlowInstanceId(tableId);
+      writeFlowToController(nodeId, tableId, flowId, createLldpToControllerFlow(FLOW_TABLE_ID, 1));
+
+      //add arpToController flow
+      flowId = getFlowInstanceId(tableId);
+      writeFlowToController(nodeId, tableId, flowId, createArpToControllerFlow(FLOW_TABLE_ID, 1));
+      _logger.info("Added initial flows for node {} ", nodeId);
+    }
+
+    private InstanceIdentifier<Table> getTableInstanceId(InstanceIdentifier<Node> nodeId) {
+      // get flow table key
+      TableKey flowTableKey = new TableKey(FLOW_TABLE_ID); //TODO: Hard coded Table Id 0, need to get it from Configuration data.
+
+      return nodeId.builder()
+          .augmentation(FlowCapableNode.class)
+          .child(Table.class, flowTableKey)
+          .build();
+    }
+
+    private InstanceIdentifier<Flow> getFlowInstanceId(InstanceIdentifier<Table> tableId) {
+      // generate unique flow key
+      FlowId flowId = new FlowId(String.valueOf(flowIdInc.getAndIncrement()));
+      FlowKey flowKey = new FlowKey(flowId);
+      return tableId.child(Flow.class, flowKey);
+    }
+
+    private Flow createDropAllFlow(Short tableId, int priority) {
+
+      // start building flow
+      FlowBuilder dropAll = new FlowBuilder() //
+          .setTableId(tableId) //
+          .setFlowName("dropall");
+
+      // use its own hash code for id.
+      dropAll.setId(new FlowId(Long.toString(dropAll.hashCode())));
+
+      Match match = new MatchBuilder().build();
+
+
+      Action dropAllAction = new ActionBuilder() //
+          .setOrder(0)
+          .setAction(new DropActionCaseBuilder().build())
+          .build();
+
+      // Create an Apply Action
+      ApplyActions applyActions = new ApplyActionsBuilder().setAction(ImmutableList.of(dropAllAction))
+          .build();
+
+      // Wrap our Apply Action in an Instruction
+      Instruction applyActionsInstruction = new InstructionBuilder() //
+          .setOrder(0)
+          .setInstruction(new ApplyActionsCaseBuilder()//
+              .setApplyActions(applyActions) //
+              .build()) //
+          .build();
+
+      // Put our Instruction in a list of Instructions
+      dropAll
+          .setMatch(match) //
+          .setInstructions(new InstructionsBuilder() //
+              .setInstruction(ImmutableList.of(applyActionsInstruction)) //
+              .build()) //
+          .setPriority(priority) //
+          .setBufferId(0L) //
+          .setHardTimeout(0) //
+          .setIdleTimeout(0) //
+          .setCookie(new FlowCookie(BigInteger.valueOf(flowCookieInc.getAndIncrement())))
+          .setFlags(new FlowModFlags(false, false, false, false, false));
+
+      return dropAll.build();
+    }
+
+    private Flow createArpToControllerFlow(Short tableId, int priority) {
+
+      // start building flow
+      FlowBuilder arpFlow = new FlowBuilder() //
+          .setTableId(tableId) //
+          .setFlowName("arptocntrl");
+
+      // use its own hash code for id.
+      arpFlow.setId(new FlowId(Long.toString(arpFlow.hashCode())));
+      EthernetMatchBuilder ethernetMatchBuilder = new EthernetMatchBuilder()
+          .setEthernetType(new EthernetTypeBuilder()
+              .setType(new EtherType(Long.valueOf(KnownEtherType.Arp.getIntValue()))).build());
+
+      Match match = new MatchBuilder()
+          .setEthernetMatch(ethernetMatchBuilder.build())
+          .build();
+
+      // Create an Apply Action
+      ApplyActions applyActions = new ApplyActionsBuilder().setAction(ImmutableList.of(getSendToControllerAction()))
+          .build();
+
+      // Wrap our Apply Action in an Instruction
+      Instruction applyActionsInstruction = new InstructionBuilder() //
+          .setOrder(0)
+          .setInstruction(new ApplyActionsCaseBuilder()//
+              .setApplyActions(applyActions) //
+              .build()) //
+          .build();
+
+      // Put our Instruction in a list of Instructions
+      arpFlow
+          .setMatch(match) //
+          .setInstructions(new InstructionsBuilder() //
+              .setInstruction(ImmutableList.of(applyActionsInstruction)) //
+              .build()) //
+          .setPriority(priority) //
+          .setBufferId(0L) //
+          .setHardTimeout(0) //
+          .setIdleTimeout(0) //
+          .setCookie(new FlowCookie(BigInteger.valueOf(flowCookieInc.getAndIncrement())))
+          .setFlags(new FlowModFlags(false, false, false, false, false));
+
+      return arpFlow.build();
+    }
+
+    private Flow createLldpToControllerFlow(Short tableId, int priority) {
+
+      // start building flow
+      FlowBuilder lldpFlow = new FlowBuilder() //
+          .setTableId(tableId) //
+          .setFlowName("lldptocntrl");
+
+      // use its own hash code for id.
+      lldpFlow.setId(new FlowId(Long.toString(lldpFlow.hashCode())));
+      EthernetMatchBuilder ethernetMatchBuilder = new EthernetMatchBuilder()
+          .setEthernetType(new EthernetTypeBuilder()
+              .setType(new EtherType(Long.valueOf(KnownEtherType.Lldp.getIntValue()))).build());
+
+      Match match = new MatchBuilder()
+          .setEthernetMatch(ethernetMatchBuilder.build())
+          .build();
+
+      // Create an Apply Action
+      ApplyActions applyActions = new ApplyActionsBuilder().setAction(ImmutableList.of(getSendToControllerAction()))
+          .build();
+
+      // Wrap our Apply Action in an Instruction
+      Instruction applyActionsInstruction = new InstructionBuilder() //
+          .setOrder(0)
+          .setInstruction(new ApplyActionsCaseBuilder()//
+              .setApplyActions(applyActions) //
+              .build()) //
+          .build();
+
+      // Put our Instruction in a list of Instructions
+      lldpFlow
+          .setMatch(match) //
+          .setInstructions(new InstructionsBuilder() //
+              .setInstruction(ImmutableList.of(applyActionsInstruction)) //
+              .build()) //
+          .setPriority(priority) //
+          .setBufferId(0L) //
+          .setHardTimeout(0) //
+          .setIdleTimeout(0) //
+          .setCookie(new FlowCookie(BigInteger.valueOf(flowCookieInc.getAndIncrement())))
+          .setFlags(new FlowModFlags(false, false, false, false, false));
+
+      return lldpFlow.build();
+    }
+
+    private Action getSendToControllerAction() {
+      Action sendToController = new ActionBuilder()
+          .setOrder(0)
+          .setKey(new ActionKey(0))
+          .setAction(new OutputActionCaseBuilder()
+              .setOutputAction(new OutputActionBuilder()
+                  .setMaxLength(new Integer(0xffff))
+                  .setOutputNodeConnector(new Uri(OutputPortValues.CONTROLLER.toString()))
+                  .build())
+              .build())
+          .build();
+
+      return sendToController;
+    }
+
+    private Future<RpcResult<AddFlowOutput>> writeFlowToController(InstanceIdentifier<Node> nodeInstanceId,
+                                                                   InstanceIdentifier<Table> tableInstanceId,
+                                                                   InstanceIdentifier<Flow> flowPath,
+                                                                   Flow flow) {
+      final AddFlowInputBuilder builder = new AddFlowInputBuilder(flow);
+      builder.setNode(new NodeRef(nodeInstanceId));
+      builder.setFlowRef(new FlowRef(flowPath));
+      builder.setFlowTable(new FlowTableRef(tableInstanceId));
+      builder.setTransactionUri(new Uri(flow.getId().getValue()));
+      return salFlowService.addFlow(builder.build());
+    }
+  }
+}

diff --git a/l2switch-main/src/test/java/org/opendaylight/l2switch/flow/InitialFlowWriterTest.java b/l2switch-main/src/test/java/org/opendaylight/l2switch/flow/InitialFlowWriterTest.java
new file mode 100644 (file)
index 0000000..09fed78
--- /dev/null
+++ b/l2switch-main/src/test/java/org/opendaylight/l2switch/flow/InitialFlowWriterTest.java
@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.opendaylight.l2switch.flow;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.SalFlowService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
+import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ *
+ */
+public class InitialFlowWriterTest {
+  private SalFlowService salFlowService;
+  private NodeUpdated nodeUpdated;
+  private AddFlowInput addFlowInput;
+
+  @Before
+  public void init() {
+    salFlowService = mock(SalFlowService.class);
+    nodeUpdated = mock(NodeUpdated.class);
+    addFlowInput = mock(AddFlowInput.class);
+    NodeRef nodeRef = new NodeRef(getNodeId(new NodeId("openflow:1")));
+    when(nodeUpdated.getNodeRef()).thenReturn(nodeRef);
+  }
+
+  @Test
+  public void testOnNodeUpdated() throws Exception {
+    InitialFlowWriter initialFlowWriter = new InitialFlowWriter(salFlowService);
+    initialFlowWriter.onNodeUpdated(nodeUpdated);
+
+    verify(salFlowService,times(3));
+
+  }
+
+  private InstanceIdentifier<Node> getNodeId(final NodeId nodeId) {
+    return InstanceIdentifier.builder(Nodes.class) //
+        .child(Node.class, new NodeKey(nodeId)) //
+        .build();
+  }
+}