Removed some sonar warnings from pcep.
[bgpcep.git] / pcep / topology-provider / src / main / java / org / opendaylight / bgpcep / pcep / topology / provider / ServerSessionManager.java
index 7fd43de770899b3354c8265ba700f01368ed2b8a..ee4a16142dbf59f727fafd9e513b6b7211dc57f1 100644 (file)
@@ -7,16 +7,14 @@
  */
 package org.opendaylight.bgpcep.pcep.topology.provider;
 
-import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.FutureListener;
-import io.netty.util.concurrent.Promise;
 
 import java.net.InetAddress;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
 
 import javax.annotation.concurrent.GuardedBy;
 
@@ -28,10 +26,13 @@ import org.opendaylight.protocol.pcep.PCEPSession;
 import org.opendaylight.protocol.pcep.PCEPSessionListener;
 import org.opendaylight.protocol.pcep.PCEPTerminationReason;
 import org.opendaylight.protocol.pcep.TerminationReason;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.PcinitiateBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.PcinitiateMessageBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.pcinitiate.message.RequestsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.Pcerr;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcerrBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcinitiateBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcupdBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.EndpointsObject;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.MessageHeader;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.PcrptMessage;
@@ -43,8 +44,6 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.typ
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.lsp.TlvsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.open.object.open.Tlvs;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcerr.message.PcerrMessageBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcinitiate.message.PcinitiateMessageBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcinitiate.message.pcinitiate.message.RequestsBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.pcrpt.message.Reports;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.PcupdMessageBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.UpdatesBuilder;
@@ -56,22 +55,30 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.typ
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathName;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathNameBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspArgs;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.EnsureLspOperationalInput;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.FailureType;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1Builder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.OperationResult;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.PccSyncState;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspArgs;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1Builder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspArgs;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClient;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClientBuilder;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLsps;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspsKey;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.StatefulTlvBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.topology.pcep.type.TopologyPcepBuilder;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypesBuilder;
 import org.opendaylight.yangtools.yang.binding.DataContainer;
 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 import org.opendaylight.yangtools.yang.common.RpcResult;
@@ -80,18 +87,23 @@ import org.slf4j.LoggerFactory;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
+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 com.google.common.util.concurrent.SettableFuture;
 
 /**
  *
  */
-final class ServerSessionManager implements SessionListenerFactory<PCEPSessionListener> {
+final class ServerSessionManager implements SessionListenerFactory<PCEPSessionListener>, AutoCloseable {
        private static String createNodeId(final InetAddress addr) {
                return "pcc://" + addr.getHostAddress();
        }
 
        private final class SessionListener implements PCEPSessionListener {
-               private final Map<SrpIdNumber, Promise<OperationResult>> waitingRequests = new HashMap<>();
-               private final Map<SrpIdNumber, Promise<OperationResult>> sendingRequests = new HashMap<>();
+               private final Map<SrpIdNumber, SettableFuture<OperationResult>> waitingRequests = new HashMap<>();
+               private final Map<SrpIdNumber, SettableFuture<OperationResult>> sendingRequests = new HashMap<>();
                private final Map<PlspId, SymbolicPathName> lsps = new HashMap<>();
                private PathComputationClientBuilder pccBuilder;
                private InstanceIdentifier<Node1> topologyAugment;
@@ -103,7 +115,7 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                private long requestId = 1;
                private NodeId nodeId;
 
-               final Node topologyNode(final DataModificationTransaction trans, final InetAddress address) {
+               Node topologyNode(final DataModificationTransaction trans, final InetAddress address) {
                        final String pccId = createNodeId(address);
                        final Topology topo = (Topology) trans.readOperationalData(ServerSessionManager.this.topology);
 
@@ -120,11 +132,11 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                         */
                        final NodeId id = new NodeId(pccId);
                        final NodeKey nk = new NodeKey(id);
-                       final InstanceIdentifier<Node> nti = InstanceIdentifier.builder(ServerSessionManager.this.topology).node(Node.class, nk).toInstance();
+                       final InstanceIdentifier<Node> nti = InstanceIdentifier.builder(ServerSessionManager.this.topology).child(Node.class, nk).toInstance();
 
                        final Node ret = new NodeBuilder().setKey(nk).setNodeId(id).build();
 
-                       trans.putRuntimeData(nti, ret);
+                       trans.putOperationalData(nti, ret);
                        this.ownsTopology = true;
                        this.topologyNode = nti;
                        this.nodeId = id;
@@ -156,23 +168,23 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                        }
 
                        this.topologyAugmentBuilder = new Node1Builder().setPathComputationClient(this.pccBuilder.build());
-                       this.topologyAugment = InstanceIdentifier.builder(this.topologyNode).node(Node1.class).toInstance();
-                       trans.putRuntimeData(this.topologyAugment, this.topologyAugmentBuilder.build());
+                       this.topologyAugment = InstanceIdentifier.builder(this.topologyNode).augmentation(Node1.class).toInstance();
+                       trans.putOperationalData(this.topologyAugment, this.topologyAugmentBuilder.build());
 
                        // All set, commit the modifications
-                       final Future<RpcResult<TransactionStatus>> s = trans.commit();
+                       final ListenableFuture<RpcResult<TransactionStatus>> f = JdkFutureAdapters.listenInPoolThread(trans.commit());
+                       Futures.addCallback(f, new FutureCallback<RpcResult<TransactionStatus>>() {
+                               @Override
+                               public void onSuccess(final RpcResult<TransactionStatus> result) {
+                                       // Nothing to do
+                               }
 
-                       /*
-                        * FIXME: once this Future is listenable, attach to it so we can
-                        *        do cleanup if the commit fails. For now we force a commit.
-                        */
-                       try {
-                               s.get();
-                       } catch (InterruptedException | ExecutionException e) {
-                               LOG.error("Failed to update internal state for session {}, terminating it", session, e);
-                               session.close(TerminationReason.Unknown);
-                               return;
-                       }
+                               @Override
+                               public void onFailure(final Throwable t) {
+                                       LOG.error("Failed to update internal state for session {}, terminating it", session, t);
+                                       session.close(TerminationReason.Unknown);
+                               }
+                       });
 
                        ServerSessionManager.this.nodes.put(this.nodeId, this);
                        this.session = session;
@@ -187,33 +199,34 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                        final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 
                        // The session went down. Undo all the Topology changes we have done.
-                       trans.removeRuntimeData(this.topologyAugment);
+                       trans.removeOperationalData(this.topologyAugment);
                        if (this.ownsTopology) {
-                               trans.removeRuntimeData(this.topologyNode);
+                               trans.removeOperationalData(this.topologyNode);
                        }
 
-                       /*
-                        * FIXME: once this Future is listenable, attach to it so we can
-                        *        do cleanup if the commit fails. For now we force a commit.
-                        */
-                       final Future<RpcResult<TransactionStatus>> s = trans.commit();
-                       try {
-                               s.get();
-                       } catch (InterruptedException | ExecutionException e) {
-                               LOG.error("Failed to cleanup internal state for session {}", session, e);
-                       }
+                       Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
+                               @Override
+                               public void onSuccess(final RpcResult<TransactionStatus> result) {
+                                       // Nothing to do
+                               }
+
+                               @Override
+                               public void onFailure(final Throwable t) {
+                                       LOG.error("Failed to cleanup internal state for session {}", session, t);
+                               }
+                       });
 
                        // Clear all requests which have not been sent to the peer: they result in cancellation
-                       for (final Entry<SrpIdNumber, Promise<OperationResult>> e : this.sendingRequests.entrySet()) {
+                       for (final Entry<SrpIdNumber, SettableFuture<OperationResult>> e : this.sendingRequests.entrySet()) {
                                LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey());
-                               e.getValue().setSuccess(OPERATION_UNSENT);
+                               e.getValue().set(OPERATION_UNSENT);
                        }
                        this.sendingRequests.clear();
 
                        // CLear all requests which have not been acked by the peer: they result in failure
-                       for (final Entry<SrpIdNumber, Promise<OperationResult>> e : this.waitingRequests.entrySet()) {
+                       for (final Entry<SrpIdNumber, SettableFuture<OperationResult>> e : this.waitingRequests.entrySet()) {
                                LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey());
-                               e.getValue().setSuccess(OPERATION_NOACK);
+                               e.getValue().set(OPERATION_NOACK);
                        }
                        this.waitingRequests.clear();
                }
@@ -231,14 +244,15 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                }
 
                private InstanceIdentifier<ReportedLsps> lspIdentifier(final SymbolicPathName name) {
-                       return InstanceIdentifier.builder(this.topologyAugment).node(ReportedLsps.class, new ReportedLspsKey(name.getPathName())).toInstance();
+                       return InstanceIdentifier.builder(this.topologyAugment).child(PathComputationClient.class).child(ReportedLsps.class,
+                                       new ReportedLspsKey(name.getPathName())).toInstance();
                }
 
                @Override
                public synchronized void onMessage(final PCEPSession session, final Message message) {
                        if (!(message instanceof PcrptMessage)) {
                                LOG.info("Unhandled message {} on session {}", message, session);
-                               session.sendMessage(unhandledMessageError);
+                               session.sendMessage(UNHANDLED_MESSAGE_ERROR);
                        }
 
                        final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.PcrptMessage rpt = ((PcrptMessage) message).getPcrptMessage();
@@ -252,7 +266,7 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                                        // Update synchronization flag
                                        this.synced = true;
                                        this.topologyAugmentBuilder.setPathComputationClient(this.pccBuilder.setStateSync(PccSyncState.Synchronized).build());
-                                       trans.putRuntimeData(this.topologyAugment, this.topologyAugmentBuilder.build());
+                                       trans.putOperationalData(this.topologyAugment, this.topologyAugmentBuilder.build());
                                        LOG.debug("Session {} achieved synchronized state", session);
                                }
 
@@ -264,10 +278,10 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                                                case Active:
                                                case Down:
                                                case Up:
-                                                       final Promise<OperationResult> p = this.waitingRequests.remove(id);
+                                                       final SettableFuture<OperationResult> p = this.waitingRequests.remove(id);
                                                        if (p != null) {
                                                                LOG.debug("Request {} resulted in LSP operational state {}", id, lsp.getOperational());
-                                                               p.setSuccess(OPERATION_SUCCESS);
+                                                               p.set(OPERATION_SUCCESS);
                                                        } else {
                                                                LOG.warn("Request ID {} not found in outstanding DB", id);
                                                        }
@@ -284,7 +298,7 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                                if (lsp.isRemove()) {
                                        final SymbolicPathName name = this.lsps.remove(id);
                                        if (name != null) {
-                                               trans.removeRuntimeData(lspIdentifier(name));
+                                               trans.removeOperationalData(lspIdentifier(name));
                                        }
 
                                        LOG.debug("LSP {} removed", lsp);
@@ -295,30 +309,32 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                                                final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.lsp.Tlvs tlvs = r.getLsp().getTlvs();
                                                final SymbolicPathName name = tlvs.getSymbolicPathName();
                                                if (name == null) {
-                                                       LOG.error("PLSPID {} seen for the first time, not reporting the LSP");
-                                                       // TODO: what should we do here?
+                                                       LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id);
+                                                       // FIXME: BUG-189
                                                        continue;
                                                }
+                                               this.lsps.put(id, name);
                                        }
 
                                        final SymbolicPathName name = this.lsps.get(id);
-                                       trans.putRuntimeData(lspIdentifier(name), lsp);
+                                       trans.putOperationalData(lspIdentifier(name), lsp);
 
-                                       LOG.debug("LSP {} updated");
+                                       LOG.debug("LSP {} updated", lsp);
                                }
                        }
 
-                       /*
-                        * FIXME: once this Future is listenable, attach to it so we can
-                        *        do cleanup if the commit fails. For now we force a commit.
-                        */
-                       final Future<RpcResult<TransactionStatus>> s = trans.commit();
-                       try {
-                               s.get();
-                       } catch (InterruptedException | ExecutionException e) {
-                               LOG.error("Failed to update internal state for session {}, closing it", session, e);
-                               session.close(TerminationReason.Unknown);
-                       }
+                       Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
+                               @Override
+                               public void onSuccess(final RpcResult<TransactionStatus> result) {
+                                       // Nothing to do
+                               }
+
+                               @Override
+                               public void onFailure(final Throwable t) {
+                                       LOG.error("Failed to update internal state for session {}, closing it", session, t);
+                                       session.close(TerminationReason.Unknown);
+                               }
+                       });
                }
 
                private synchronized SrpIdNumber nextRequest() {
@@ -326,20 +342,19 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                }
 
                private synchronized void messageSendingComplete(final SrpIdNumber requestId, final io.netty.util.concurrent.Future<Void> future) {
-                       final Promise<OperationResult> promise = this.sendingRequests.remove(requestId);
+                       final SettableFuture<OperationResult> promise = this.sendingRequests.remove(requestId);
 
                        if (future.isSuccess()) {
                                this.waitingRequests.put(requestId, promise);
                        } else {
                                LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
-                               promise.setSuccess(OPERATION_UNSENT);
+                               promise.set(OPERATION_UNSENT);
                        }
                }
 
-               private synchronized io.netty.util.concurrent.Future<OperationResult> sendMessage(final Message message,
-                               final SrpIdNumber requestId) {
+               private synchronized ListenableFuture<OperationResult> sendMessage(final Message message, final SrpIdNumber requestId) {
                        final io.netty.util.concurrent.Future<Void> f = this.session.sendMessage(message);
-                       final Promise<OperationResult> ret = exec.newPromise();
+                       final SettableFuture<OperationResult> ret = SettableFuture.create();
 
                        this.sendingRequests.put(requestId, ret);
 
@@ -355,9 +370,9 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
        }
 
        private static final Logger LOG = LoggerFactory.getLogger(ServerSessionManager.class);
-       private static final Pcerr unhandledMessageError = new PcerrBuilder().setPcerrMessage(
+       private static final Pcerr UNHANDLED_MESSAGE_ERROR = new PcerrBuilder().setPcerrMessage(
                        new PcerrMessageBuilder().setErrorType(null).build()).build();
-       private static final MessageHeader messageHeader = new MessageHeader() {
+       private static final MessageHeader MESSAGE_HEADER = new MessageHeader() {
                private final ProtocolVersion version = new ProtocolVersion((short) 1);
 
                @Override
@@ -378,15 +393,36 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
        private final Map<NodeId, SessionListener> nodes = new HashMap<>();
        private final InstanceIdentifier<Topology> topology;
        private final DataProviderService dataProvider;
-       private final EventExecutor exec;
 
-       public ServerSessionManager(
-                       final EventExecutor exec,
-                       final DataProviderService dataProvider,
-                       final InstanceIdentifier<Topology> topology) {
+       public ServerSessionManager(final DataProviderService dataProvider, final InstanceIdentifier<Topology> topology) {
                this.dataProvider = Preconditions.checkNotNull(dataProvider);
                this.topology = Preconditions.checkNotNull(topology);
-               this.exec = Preconditions.checkNotNull(exec);
+
+               // Make sure the topology does not exist
+               final Object c = dataProvider.readOperationalData(topology);
+               Preconditions.checkArgument(c == null, "Topology %s already exists", topology);
+
+               // Now create the base topology
+               final TopologyKey k = InstanceIdentifier.keyOf(topology);
+               final DataModificationTransaction t = dataProvider.beginTransaction();
+               t.putOperationalData(
+                               topology,
+                               new TopologyBuilder().setKey(k).setTopologyId(k.getTopologyId()).setTopologyTypes(
+                                               new TopologyTypesBuilder().addAugmentation(TopologyTypes1.class,
+                                                               new TopologyTypes1Builder().setTopologyPcep(new TopologyPcepBuilder().build()).build()).build()).setNode(
+                                               new ArrayList<Node>()).build());
+
+               Futures.addCallback(JdkFutureAdapters.listenInPoolThread(t.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
+                       @Override
+                       public void onSuccess(final RpcResult<TransactionStatus> result) {
+                               // Nothing to do
+                       }
+
+                       @Override
+                       public void onFailure(final Throwable t) {
+                               LOG.error("Failed to create topology {}", topology);
+                       }
+               });
        }
 
        @Override
@@ -394,31 +430,29 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                return new SessionListener();
        }
 
-       synchronized io.netty.util.concurrent.Future<OperationResult> realAddLsp(final AddLspArgs input) {
+       synchronized ListenableFuture<OperationResult> realAddLsp(final AddLspArgs input) {
                // Get the listener corresponding to the node
                final SessionListener l = this.nodes.get(input.getNode());
                if (l == null) {
                        LOG.debug("Session for node {} not found", input.getNode());
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Make sure there is no such LSP
-               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
-                               new ReportedLspsKey(input.getName())).toInstance();
+               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
+                               ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
                if (this.dataProvider.readOperationalData(lsp) != null) {
                        LOG.debug("Node {} already contains lsp {} at {}", input.getNode(), input.getName(), lsp);
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Build the request
-               final RequestsBuilder rb = new RequestsBuilder(input.getArguments());
+               final RequestsBuilder rb = new RequestsBuilder((EndpointsObject) input.getArguments());
                rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
-               rb.setLsp(
-                               new LspBuilder().setAdministrative(input.getArguments().isAdministrative()).setDelegate(Boolean.TRUE).setTlvs(
-                                               new TlvsBuilder().setSymbolicPathName(
-                                                               new SymbolicPathNameBuilder().setPathName(input.getName()).build()).build()).build());
+               rb.setLsp(new LspBuilder().setAdministrative(input.getArguments().isAdministrative()).setDelegate(Boolean.TRUE).setTlvs(
+                               new TlvsBuilder().setSymbolicPathName(new SymbolicPathNameBuilder().setPathName(input.getName()).build()).build()).build());
 
-               final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
+               final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(MESSAGE_HEADER);
                ib.setRequests(ImmutableList.of(rb.build()));
 
                // Send the message
@@ -439,21 +473,21 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                };
        }
 
-       synchronized io.netty.util.concurrent.Future<OperationResult> realRemoveLsp(final RemoveLspArgs input) {
+       synchronized ListenableFuture<OperationResult> realRemoveLsp(final RemoveLspArgs input) {
                // Get the listener corresponding to the node
                final SessionListener l = this.nodes.get(input.getNode());
                if (l == null) {
                        LOG.debug("Session for node {} not found", input.getNode());
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Make sure the LSP exists, we need it for PLSP-ID
-               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
-                               new ReportedLspsKey(input.getName())).toInstance();
+               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
+                               ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
                final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
                if (rep == null) {
                        LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Build the request and send it
@@ -461,36 +495,69 @@ final class ServerSessionManager implements SessionListenerFactory<PCEPSessionLi
                rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).setFlags(new Flags(Boolean.TRUE)).build());
                rb.setLsp(new LspBuilder().setRemove(Boolean.TRUE).setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 
-               final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
+               final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(MESSAGE_HEADER);
                ib.setRequests(ImmutableList.of(rb.build()));
                return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId());
        }
 
-       synchronized io.netty.util.concurrent.Future<OperationResult> realUpdateLsp(final UpdateLspArgs input) {
+       synchronized ListenableFuture<OperationResult> realUpdateLsp(final UpdateLspArgs input) {
                // Get the listener corresponding to the node
                final SessionListener l = this.nodes.get(input.getNode());
                if (l == null) {
                        LOG.debug("Session for node {} not found", input.getNode());
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Make sure the LSP exists
-               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
-                               new ReportedLspsKey(input.getName())).toInstance();
+               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
+                               ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
                final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
                if (rep == null) {
                        LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
-                       return exec.newSucceededFuture(OPERATION_UNSENT);
+                       return Futures.immediateFuture(OPERATION_UNSENT);
                }
 
                // Build the PCUpd request and send it
                final UpdatesBuilder rb = new UpdatesBuilder();
                rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
                rb.setLsp(new LspBuilder().setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
-               rb.setPath(new PathBuilder(input.getArguments()).build());
+               final PathBuilder pb = new PathBuilder();
+               rb.setPath(pb.setEro(input.getArguments().getEro()).build());
 
-               final PcupdMessageBuilder ub = new PcupdMessageBuilder(messageHeader);
+               final PcupdMessageBuilder ub = new PcupdMessageBuilder(MESSAGE_HEADER);
                ub.setUpdates(ImmutableList.of(rb.build()));
                return l.sendMessage(new PcupdBuilder().setPcupdMessage(ub.build()).build(), rb.getSrp().getOperationId());
        }
+
+       synchronized ListenableFuture<OperationResult> realEnsureLspOperational(final EnsureLspOperationalInput input) {
+               // Get the listener corresponding to the node
+               final SessionListener l = this.nodes.get(input.getNode());
+               if (l == null) {
+                       LOG.debug("Session for node {} not found", input.getNode());
+                       return Futures.immediateFuture(OPERATION_UNSENT);
+               }
+
+               // Make sure the LSP exists
+               final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
+                               ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
+               LOG.debug("Checking if LSP {} has operational state {}", lsp, input.getArguments().getOperational());
+               final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
+               if (rep == null) {
+                       LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
+                       return Futures.immediateFuture(OPERATION_UNSENT);
+               }
+
+               if (rep.getLsp().getOperational().equals(input.getArguments().getOperational())) {
+                       return Futures.immediateFuture(OPERATION_SUCCESS);
+               } else {
+                       return Futures.immediateFuture(OPERATION_UNSENT);
+               }
+       }
+
+       @Override
+       public void close() throws InterruptedException, ExecutionException {
+               final DataModificationTransaction t = this.dataProvider.beginTransaction();
+               t.removeOperationalData(this.topology);
+               t.commit().get();
+       }
 }