BUG-3157: Use InstanceIdentifier.firstIdentifierOf()

[openflowplugin.git] / test-common / src / main / java / org / opendaylight / openflowplugin / testcommon / DropTestRpcSender.java

diff --git a/test-common/src/main/java/org/opendaylight/openflowplugin/testcommon/DropTestRpcSender.java b/test-common/src/main/java/org/opendaylight/openflowplugin/testcommon/DropTestRpcSender.java
index 30089b69a18688d694648d85ffac5c17c0682167..f88532521c21d3ce7344e887961d908954909faa 100644 (file)
--- a/test-common/src/main/java/org/opendaylight/openflowplugin/testcommon/DropTestRpcSender.java
+++ b/test-common/src/main/java/org/opendaylight/openflowplugin/testcommon/DropTestRpcSender.java
@@ -7,28 +7,27 @@
  */
 package org.opendaylight.openflowplugin.testcommon;
 
-import static org.opendaylight.openflowplugin.testcommon.AbstractDropTest.BUFFER_ID;
-import static org.opendaylight.openflowplugin.testcommon.AbstractDropTest.HARD_TIMEOUT;
-import static org.opendaylight.openflowplugin.testcommon.AbstractDropTest.IDLE_TIMEOUT;
-import static org.opendaylight.openflowplugin.testcommon.AbstractDropTest.PRIORITY;
-import static org.opendaylight.openflowplugin.testcommon.AbstractDropTest.TABLE_ID;
-
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.JdkFutureAdapters;
+import com.google.common.util.concurrent.ListenableFuture;
 import java.math.BigInteger;
-import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import java.util.concurrent.Callable;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
+import org.opendaylight.openflowplugin.common.wait.SimpleTaskRetryLooper;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.AddFlowInput;
 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.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.flow.Instructions;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.Match;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
-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.concepts.ListenerRegistration;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
-import org.opendaylight.yangtools.yang.binding.NotificationListener;
+import org.opendaylight.yangtools.yang.common.RpcResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +43,7 @@ public class DropTestRpcSender extends AbstractDropTest {
     /**
      * @param flowService the flowService to set
      */
-    public void setFlowService(SalFlowService flowService) {
+    public void setFlowService(final SalFlowService flowService) {
         this.flowService = flowService;
     }
 
@@ -68,58 +67,79 @@ public class DropTestRpcSender extends AbstractDropTest {
         }
     };
 
-    private NotificationProviderService notificationService;
+    private NotificationService notificationService;
 
-    private ListenerRegistration<NotificationListener> notificationRegistration;
+    private ListenerRegistration<DropTestRpcSender> notificationRegistration;
 
     /**
      * start listening on packetIn
      */
     public void start() {
-        notificationRegistration = notificationService.registerNotificationListener(this);
+        final SimpleTaskRetryLooper looper = new SimpleTaskRetryLooper(STARTUP_LOOP_TICK,
+                STARTUP_LOOP_MAX_RETRIES);
+        try {
+            notificationRegistration = looper.loopUntilNoException(new Callable<ListenerRegistration<DropTestRpcSender>>() {
+                @Override
+                public ListenerRegistration<DropTestRpcSender> call() throws Exception {
+                    return notificationService.registerNotificationListener(DropTestRpcSender.this);
+                }
+            });
+        } catch (final Exception e) {
+            LOG.warn("DropTest sender notification listener registration fail!");
+            LOG.debug("DropTest sender notification listener registration fail! ..", e);
+            throw new IllegalStateException("DropTest startup fail! Try again later.", e);
+        }
     }
 
     @Override
-    protected void processPacket(final NodeKey node, final Match match, final Instructions instructions) {
+    protected void processPacket(final InstanceIdentifier<Node> node, final Match match, final Instructions instructions) {
         final AddFlowInputBuilder fb = BUILDER.get();
 
         // Finally build our flow
         fb.setMatch(match);
         fb.setInstructions(instructions);
-        //fb.setId(new FlowId(Long.toString(fb.hashCode)));
 
         // Construct the flow instance id
-        final InstanceIdentifier<Node> flowInstanceId = InstanceIdentifier
-                // File under nodes
-                .builder(Nodes.class)
-                // A particular node identified by nodeKey
-                .child(Node.class, node).build();
-        fb.setNode(new NodeRef(flowInstanceId));
+
+        fb.setNode(new NodeRef(node));
 
         // Add flow
-        AddFlowInput flow = fb.build();
+        final AddFlowInput flow = fb.build();
         if (LOG.isDebugEnabled()) {
             LOG.debug("onPacketReceived - About to write flow (via SalFlowService) {}", flow);
         }
-        flowService.addFlow(flow);
+        ListenableFuture<RpcResult<AddFlowOutput>> result = JdkFutureAdapters.listenInPoolThread(flowService.addFlow(flow));
+        Futures.addCallback(result, new FutureCallback<RpcResult<AddFlowOutput>>() {
+            @Override
+            public void onSuccess(final RpcResult<AddFlowOutput> o) {
+                countFutureSuccess();
+            }
+
+            @Override
+            public void onFailure(final Throwable throwable) {
+                countFutureError();
+            }
+        });
     }
 
     /**
      * @param notificationService
      */
-    public void setNotificationService(NotificationProviderService notificationService) {
+    public void setNotificationService(final NotificationService notificationService) {
         this.notificationService = notificationService;
     }
 
     @Override
     public void close() {
+        super.close();
         try {
             LOG.debug("DropTestProvider stopped.");
             if (notificationRegistration != null) {
                 notificationRegistration.close();
             }
-        } catch (Exception e) {
-            LOG.error("unregistration of notification listener failed", e);
+        } catch (final Exception e) {
+            LOG.warn("unregistration of notification listener failed: {}", e.getMessage());
+            LOG.debug("unregistration of notification listener failed.. ", e);
         }
     }
 }