Switch to MD-SAL APIs

[openflowplugin.git] / applications / topology-manager / src / main / java / org / opendaylight / openflowplugin / applications / topology / manager / OperationProcessor.java

diff --git a/applications/topology-manager/src/main/java/org/opendaylight/openflowplugin/applications/topology/manager/OperationProcessor.java b/applications/topology-manager/src/main/java/org/opendaylight/openflowplugin/applications/topology/manager/OperationProcessor.java
index 4d9b5019f96faf16549185b3530f00ec8d510bc9..3312779c8ef1c98926ac3cb1d5bd558ad3dde016 100644 (file)
--- a/applications/topology-manager/src/main/java/org/opendaylight/openflowplugin/applications/topology/manager/OperationProcessor.java
+++ b/applications/topology-manager/src/main/java/org/opendaylight/openflowplugin/applications/topology/manager/OperationProcessor.java
@@ -9,30 +9,37 @@ package org.opendaylight.openflowplugin.applications.topology.manager;
 
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import org.opendaylight.controller.md.sal.binding.api.DataBroker;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.aries.blueprint.annotation.service.Reference;
+import org.opendaylight.mdsal.binding.api.DataBroker;
 import org.opendaylight.openflowplugin.common.txchain.TransactionChainManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Singleton
 public final class OperationProcessor implements AutoCloseable, Runnable {
     private static final Logger LOG = LoggerFactory.getLogger(OperationProcessor.class);
     private static final int MAX_TRANSACTION_OPERATIONS = 100;
     private static final int OPERATION_QUEUE_DEPTH = 500;
-    private static final String TOPOLOGY_MANAGER = "topology-manager";
+    private static final String TOPOLOGY_MANAGER = "ofp-topo-processor";
 
     private final BlockingQueue<TopologyOperation> queue = new LinkedBlockingQueue<>(OPERATION_QUEUE_DEPTH);
     private final Thread thread;
-    private TransactionChainManager transactionChainManager;
+    private final TransactionChainManager transactionChainManager;
     private volatile boolean finishing = false;
 
-    public OperationProcessor(final DataBroker dataBroker) {
+    @Inject
+    public OperationProcessor(@Reference final DataBroker dataBroker) {
         transactionChainManager = new TransactionChainManager(dataBroker, TOPOLOGY_MANAGER);
         transactionChainManager.activateTransactionManager();
         transactionChainManager.initialSubmitWriteTransaction();
 
         thread = new Thread(this);
         thread.setDaemon(true);
-        thread.setName("FlowCapableTopologyExporter-" + FlowCapableTopologyProvider.TOPOLOGY_ID);
+        thread.setName("ofp-topo-expo-" + FlowCapableTopologyProvider.TOPOLOGY_ID);
     }
 
     void enqueueOperation(final TopologyOperation task) {
@@ -43,42 +50,43 @@ public final class OperationProcessor implements AutoCloseable, Runnable {
         }
     }
 
+    @PostConstruct
     public void start() {
         thread.start();
     }
 
     @Override
     public void run() {
-            while (!finishing) {
-                try {
-                    TopologyOperation op = queue.take();
+        while (!finishing) {
+            try {
+                TopologyOperation op = queue.take();
 
-                    LOG.debug("New {} operation available, starting transaction", op);
+                LOG.debug("New {} operation available, starting transaction", op);
 
-                    int ops = 0;
-                    do {
-                        op.applyOperation(transactionChainManager);
+                int ops = 0;
+                do {
+                    op.applyOperation(transactionChainManager);
 
-                        ops++;
-                        if (ops < MAX_TRANSACTION_OPERATIONS) {
-                            op = queue.poll();
-                        } else {
-                            op = null;
-                        }
+                    ops++;
+                    if (ops < MAX_TRANSACTION_OPERATIONS) {
+                        op = queue.poll();
+                    } else {
+                        op = null;
+                    }
 
-                        LOG.debug("Next operation {}", op);
-                    } while (op != null);
+                    LOG.debug("Next operation {}", op);
+                } while (op != null);
 
-                    LOG.debug("Processed {} operations, submitting transaction", ops);
-                    if (!transactionChainManager.submitTransaction()) {
-                        cleanDataStoreOperQueue();
-                    }
-                } catch (final InterruptedException e) {
-                    // This should mean we're shutting down.
-                    LOG.debug("Stat Manager DS Operation thread interrupted!", e);
-                    finishing = true;
+                LOG.debug("Processed {} operations, submitting transaction", ops);
+                if (!transactionChainManager.submitTransaction()) {
+                    cleanDataStoreOperQueue();
                 }
+            } catch (final InterruptedException e) {
+                // This should mean we're shutting down.
+                LOG.debug("Stat Manager DS Operation thread interrupted!", e);
+                finishing = true;
             }
+        }
         // Drain all events, making sure any blocked threads are unblocked
         cleanDataStoreOperQueue();
     }
@@ -90,11 +98,12 @@ public final class OperationProcessor implements AutoCloseable, Runnable {
     }
 
     @Override
+    @PreDestroy
     public void close() {
         thread.interrupt();
         try {
             thread.join();
-        } catch(InterruptedException e) {
+        } catch (InterruptedException e) {
             LOG.debug("Join of thread {} was interrupted", thread.getName(), e);
         }