Fix mounting logic in clustered topology

[netconf.git] / netconf / netconf-topology-singleton / src / main / java / org / opendaylight / netconf / topology / singleton / impl / NetconfTopologyManager.java

diff --git a/netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfTopologyManager.java b/netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfTopologyManager.java
index 86b7111f78ce325bfd0b6ee6285f08df3ee30922..9dd0a22da18dd17fa1d17f7d1116d5bdec9923ff 100644 (file)
--- a/netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfTopologyManager.java
+++ b/netconf/netconf-topology-singleton/src/main/java/org/opendaylight/netconf/topology/singleton/impl/NetconfTopologyManager.java
@@ -5,7 +5,6 @@
  * 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.netconf.topology.singleton.impl;
 
 import akka.actor.ActorSystem;
@@ -13,12 +12,13 @@ import akka.util.Timeout;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.util.concurrent.EventExecutor;
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.Nonnull;
 import org.opendaylight.aaa.encrypt.AAAEncryptionService;
@@ -34,6 +34,7 @@ import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
 import org.opendaylight.controller.md.sal.dom.api.DOMMountPointService;
 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
+import org.opendaylight.mdsal.common.api.CommitInfo;
 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
@@ -70,8 +71,8 @@ public class NetconfTopologyManager
     private final DataBroker dataBroker;
     private final RpcProviderRegistry rpcProviderRegistry;
     private final ClusterSingletonServiceProvider clusterSingletonServiceProvider;
-    private final ScheduledThreadPool keepaliveExecutor;
-    private final ThreadPool processingExecutor;
+    private final ScheduledExecutorService keepaliveExecutor;
+    private final ListeningExecutorService processingExecutor;
     private final ActorSystem actorSystem;
     private final EventExecutor eventExecutor;
     private final NetconfClientDispatcher clientDispatcher;
@@ -95,8 +96,8 @@ public class NetconfTopologyManager
         this.dataBroker = Preconditions.checkNotNull(dataBroker);
         this.rpcProviderRegistry = Preconditions.checkNotNull(rpcProviderRegistry);
         this.clusterSingletonServiceProvider = Preconditions.checkNotNull(clusterSingletonServiceProvider);
-        this.keepaliveExecutor = Preconditions.checkNotNull(keepaliveExecutor);
-        this.processingExecutor = Preconditions.checkNotNull(processingExecutor);
+        this.keepaliveExecutor = keepaliveExecutor.getExecutor();
+        this.processingExecutor = MoreExecutors.listeningDecorator(processingExecutor.getExecutor());
         this.actorSystem = Preconditions.checkNotNull(actorSystemProvider).getActorSystem();
         this.eventExecutor = Preconditions.checkNotNull(eventExecutor);
         this.clientDispatcher = Preconditions.checkNotNull(clientDispatcher);
@@ -197,8 +198,8 @@ public class NetconfTopologyManager
     }
 
     @VisibleForTesting
-    protected NetconfTopologyContext newNetconfTopologyContext(NetconfTopologySetup setup,
-            ServiceGroupIdentifier serviceGroupIdent, Timeout actorResponseWaitTime) {
+    protected NetconfTopologyContext newNetconfTopologyContext(final NetconfTopologySetup setup,
+            final ServiceGroupIdentifier serviceGroupIdent, final Timeout actorResponseWaitTime) {
         return new NetconfTopologyContext(setup, serviceGroupIdent, actorResponseWaitTime, mountPointService);
     }
 
@@ -209,17 +210,15 @@ public class NetconfTopologyManager
             dataChangeListenerRegistration = null;
         }
 
-        contexts.values().forEach(netconfTopologyContext -> close(netconfTopologyContext));
-
-        clusterRegistrations.values().forEach(
-            clusterSingletonServiceRegistration -> close(clusterSingletonServiceRegistration));
+        contexts.values().forEach(NetconfTopologyManager::close);
+        clusterRegistrations.values().forEach(NetconfTopologyManager::close);
 
         contexts.clear();
         clusterRegistrations.clear();
     }
 
     @SuppressWarnings("checkstyle:IllegalCatch")
-    private void close(AutoCloseable closeable) {
+    private static void close(final AutoCloseable closeable) {
         try {
             closeable.close();
         } catch (Exception e) {
@@ -245,9 +244,9 @@ public class NetconfTopologyManager
         final WriteTransaction wtx = dataBroker.newWriteOnlyTransaction();
         initTopology(wtx, LogicalDatastoreType.CONFIGURATION);
         initTopology(wtx, LogicalDatastoreType.OPERATIONAL);
-        Futures.addCallback(wtx.submit(), new FutureCallback<Void>() {
+        wtx.commit().addCallback(new FutureCallback<CommitInfo>() {
             @Override
-            public void onSuccess(final Void result) {
+            public void onSuccess(final CommitInfo result) {
                 LOG.debug("topology initialization successful");
             }