Fix transaction manager closing.

[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 bbfae52b84910b179fe182d9b3fa345a0ef90934..4d9b5019f96faf16549185b3530f00ec8d510bc9 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
@@ -7,33 +7,28 @@
  */
 package org.opendaylight.openflowplugin.applications.topology.manager;
 
-import com.google.common.base.Preconditions;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
-import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
-import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
-import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
+import org.opendaylight.openflowplugin.common.txchain.TransactionChainManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public final class OperationProcessor implements AutoCloseable, Runnable, TransactionChainListener {
+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 final BlockingQueue<TopologyOperation> queue = new LinkedBlockingQueue<>(OPERATION_QUEUE_DEPTH);
-    private final DataBroker dataBroker;
     private final Thread thread;
-    private BindingTransactionChain transactionChain;
+    private TransactionChainManager transactionChainManager;
     private volatile boolean finishing = false;
 
     public OperationProcessor(final DataBroker dataBroker) {
-        this.dataBroker = Preconditions.checkNotNull(dataBroker);
-        transactionChain = this.dataBroker.createTransactionChain(this);
+        transactionChainManager = new TransactionChainManager(dataBroker, TOPOLOGY_MANAGER);
+        transactionChainManager.activateTransactionManager();
+        transactionChainManager.initialSubmitWriteTransaction();
 
         thread = new Thread(this);
         thread.setDaemon(true);
@@ -60,11 +55,9 @@ public final class OperationProcessor implements AutoCloseable, Runnable, Transa
 
                     LOG.debug("New {} operation available, starting transaction", op);
 
-                    final ReadWriteTransaction tx = transactionChain.newReadWriteTransaction();
-
                     int ops = 0;
                     do {
-                        op.applyOperation(tx);
+                        op.applyOperation(transactionChainManager);
 
                         ops++;
                         if (ops < MAX_TRANSACTION_OPERATIONS) {
@@ -77,55 +70,25 @@ public final class OperationProcessor implements AutoCloseable, Runnable, Transa
                     } while (op != null);
 
                     LOG.debug("Processed {} operations, submitting transaction", ops);
-                    submitTransaction(tx);
-                } catch (final IllegalStateException e) {
-                    LOG.warn("Stat DataStoreOperation unexpected State!", e);
-                    transactionChain.close();
-                    transactionChain = dataBroker.createTransactionChain(this);
-                    cleanDataStoreOperQueue();
+                    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;
-                } catch (final Exception e) {
-                    LOG.warn("Stat DataStore Operation executor fail!", e);
                 }
             }
         // Drain all events, making sure any blocked threads are unblocked
         cleanDataStoreOperQueue();
     }
 
-    private void submitTransaction(ReadWriteTransaction tx) {
-        try {
-            tx.submit().checkedGet();
-        } catch (final TransactionCommitFailedException e) {
-            LOG.warn("Stat DataStoreOperation unexpected State!", e);
-            transactionChain.close();
-            transactionChain = dataBroker.createTransactionChain(this);
-            cleanDataStoreOperQueue();
-        }
-    }
-
     private void cleanDataStoreOperQueue() {
         while (!queue.isEmpty()) {
             queue.poll();
         }
     }
 
-    @Override
-    public void onTransactionChainFailed(TransactionChain<?, ?> chain, AsyncTransaction<?, ?> transaction, Throwable cause) {
-        LOG.warn("Failed to export Topology manager operations, Transaction {} failed: {}", transaction.getIdentifier(), cause.getMessage());
-        LOG.debug("Failed to export Topology manager operations.. ", cause);
-        transactionChain.close();
-        transactionChain = dataBroker.createTransactionChain(this);
-        cleanDataStoreOperQueue();
-    }
-
-    @Override
-    public void onTransactionChainSuccessful(TransactionChain<?, ?> chain) {
-        //NOOP
-    }
-
     @Override
     public void close() {
         thread.interrupt();
@@ -135,9 +98,7 @@ public final class OperationProcessor implements AutoCloseable, Runnable, Transa
             LOG.debug("Join of thread {} was interrupted", thread.getName(), e);
         }
 
-        if (transactionChain != null) {
-            transactionChain.close();
-        }
+        transactionChainManager.close();
 
         LOG.debug("OperationProcessor closed");
     }