pcep/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/ServerSessionManager.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.bgpcep.pcep.topology.provider;
   9
  10 import io.netty.util.concurrent.FutureListener;
  11
  12 import java.net.InetAddress;
  13 import java.util.ArrayList;
  14 import java.util.HashMap;
  15 import java.util.Map;
  16 import java.util.Map.Entry;
  17 import java.util.concurrent.ExecutionException;
  18
  19 import javax.annotation.concurrent.GuardedBy;
  20
  21 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  22 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  23 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
  24 import org.opendaylight.protocol.framework.SessionListenerFactory;
  25 import org.opendaylight.protocol.pcep.PCEPSession;
  26 import org.opendaylight.protocol.pcep.PCEPSessionListener;
  27 import org.opendaylight.protocol.pcep.PCEPTerminationReason;
  28 import org.opendaylight.protocol.pcep.TerminationReason;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.PcinitiateBuilder;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.PcinitiateMessageBuilder;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.pcinitiate.message.RequestsBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.Pcerr;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcerrBuilder;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcupdBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.EndpointsObject;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.MessageHeader;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.PcrptMessage;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.PlspId;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.ProtocolVersion;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.SrpIdNumber;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.Lsp;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.LspBuilder;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.lsp.TlvsBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.open.object.open.Tlvs;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcerr.message.PcerrMessageBuilder;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.pcrpt.message.Reports;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.PcupdMessageBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.UpdatesBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.updates.PathBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.Srp;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.Srp.Flags;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.SrpBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.stateful.capability.tlv.Stateful;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathName;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathNameBuilder;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspArgs;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.EnsureLspOperationalInput;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.FailureType;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1Builder;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.OperationResult;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.PccSyncState;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspArgs;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1Builder;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspArgs;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClient;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClientBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLsps;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspsKey;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.StatefulTlvBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.topology.pcep.type.TopologyPcepBuilder;
  74 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  75 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  76 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
  77 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  78 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  79 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  80 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  81 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypesBuilder;
  82 import org.opendaylight.yangtools.yang.binding.DataContainer;
  83 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  84 import org.opendaylight.yangtools.yang.common.RpcResult;
  85 import org.slf4j.Logger;
  86 import org.slf4j.LoggerFactory;
  87
  88 import com.google.common.base.Preconditions;
  89 import com.google.common.collect.ImmutableList;
  90 import com.google.common.util.concurrent.FutureCallback;
  91 import com.google.common.util.concurrent.Futures;
  92 import com.google.common.util.concurrent.JdkFutureAdapters;
  93 import com.google.common.util.concurrent.ListenableFuture;
  94 import com.google.common.util.concurrent.SettableFuture;
  95
  96 /**
  97  *
  98  */
  99 final class ServerSessionManager implements SessionListenerFactory<PCEPSessionListener>, AutoCloseable {
 100         private static String createNodeId(final InetAddress addr) {
 101                 return "pcc://" + addr.getHostAddress();
 102         }
 103
 104         private final class SessionListener implements PCEPSessionListener {
 105                 private final Map<SrpIdNumber, SettableFuture<OperationResult>> waitingRequests = new HashMap<>();
 106                 private final Map<SrpIdNumber, SettableFuture<OperationResult>> sendingRequests = new HashMap<>();
 107                 private final Map<PlspId, SymbolicPathName> lsps = new HashMap<>();
 108                 private PathComputationClientBuilder pccBuilder;
 109                 private InstanceIdentifier<Node1> topologyAugment;
 110                 private InstanceIdentifier<Node> topologyNode;
 111                 private Node1Builder topologyAugmentBuilder;
 112                 private boolean ownsTopology = false;
 113                 private boolean synced = false;
 114                 private PCEPSession session;
 115                 private long requestId = 1;
 116                 private NodeId nodeId;
 117
 118                 final Node topologyNode(final DataModificationTransaction trans, final InetAddress address) {
 119                         final String pccId = createNodeId(address);
 120                         final Topology topo = (Topology) trans.readOperationalData(ServerSessionManager.this.topology);
 121
 122                         for (final Node n : topo.getNode()) {
 123                                 LOG.debug("Matching topology node {} to id {}", n, pccId);
 124                                 if (n.getNodeId().getValue().equals(pccId)) {
 125                                         return n;
 126                                 }
 127                         }
 128
 129                         /*
 130                          * We failed to find a matching node. Let's create a dynamic one
 131                          * and note that we are the owner (so we clean it up afterwards).
 132                          */
 133                         final NodeId id = new NodeId(pccId);
 134                         final NodeKey nk = new NodeKey(id);
 135                         final InstanceIdentifier<Node> nti = InstanceIdentifier.builder(ServerSessionManager.this.topology).child(Node.class, nk).toInstance();
 136
 137                         final Node ret = new NodeBuilder().setKey(nk).setNodeId(id).build();
 138
 139                         trans.putOperationalData(nti, ret);
 140                         this.ownsTopology = true;
 141                         this.topologyNode = nti;
 142                         this.nodeId = id;
 143                         return ret;
 144                 }
 145
 146                 @Override
 147                 public synchronized void onSessionUp(final PCEPSession session) {
 148                         /*
 149                          * The session went up. Look up the router in Inventory model,
 150                          * create it if it is not there (marking that fact for later
 151                          * deletion), and mark it as synchronizing. Also create it in
 152                          * the topology model, with empty LSP list.
 153                          */
 154                         final InetAddress peerAddress = session.getRemoteAddress();
 155                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 156
 157                         final Node topoNode = topologyNode(trans, peerAddress);
 158                         LOG.debug("Peer {} resolved to topology node {}", peerAddress, topoNode);
 159
 160                         // Our augmentation in the topology node
 161                         this.pccBuilder = new PathComputationClientBuilder();
 162
 163                         final Tlvs tlvs = session.getRemoteTlvs();
 164                         final Stateful stateful = tlvs.getStateful();
 165                         if (stateful != null) {
 166                                 this.pccBuilder.setStatefulTlv(new StatefulTlvBuilder(tlvs).build());
 167                                 this.pccBuilder.setStateSync(PccSyncState.InitialResync);
 168                         }
 169
 170                         this.topologyAugmentBuilder = new Node1Builder().setPathComputationClient(this.pccBuilder.build());
 171                         this.topologyAugment = InstanceIdentifier.builder(this.topologyNode).augmentation(Node1.class).toInstance();
 172                         trans.putOperationalData(this.topologyAugment, this.topologyAugmentBuilder.build());
 173
 174                         // All set, commit the modifications
 175                         final ListenableFuture<RpcResult<TransactionStatus>> f = JdkFutureAdapters.listenInPoolThread(trans.commit());
 176                         Futures.addCallback(f, new FutureCallback<RpcResult<TransactionStatus>>() {
 177                                 @Override
 178                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 179                                         // Nothing to do
 180                                 }
 181
 182                                 @Override
 183                                 public void onFailure(final Throwable t) {
 184                                         LOG.error("Failed to update internal state for session {}, terminating it", session, t);
 185                                         session.close(TerminationReason.Unknown);
 186                                 }
 187                         });
 188
 189                         ServerSessionManager.this.nodes.put(this.nodeId, this);
 190                         this.session = session;
 191                         LOG.info("Session with {} attached to topology node {}", session.getRemoteAddress(), topoNode.getNodeId());
 192                 }
 193
 194                 @GuardedBy("this")
 195                 private void tearDown(final PCEPSession session) {
 196                         this.session = null;
 197                         ServerSessionManager.this.nodes.remove(this.nodeId);
 198
 199                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 200
 201                         // The session went down. Undo all the Topology changes we have done.
 202                         trans.removeOperationalData(this.topologyAugment);
 203                         if (this.ownsTopology) {
 204                                 trans.removeOperationalData(this.topologyNode);
 205                         }
 206
 207                         Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
 208                                 @Override
 209                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 210                                         // Nothing to do
 211                                 }
 212
 213                                 @Override
 214                                 public void onFailure(final Throwable t) {
 215                                         LOG.error("Failed to cleanup internal state for session {}", session, t);
 216                                 }
 217                         });
 218
 219                         // Clear all requests which have not been sent to the peer: they result in cancellation
 220                         for (final Entry<SrpIdNumber, SettableFuture<OperationResult>> e : this.sendingRequests.entrySet()) {
 221                                 LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey());
 222                                 e.getValue().set(OPERATION_UNSENT);
 223                         }
 224                         this.sendingRequests.clear();
 225
 226                         // CLear all requests which have not been acked by the peer: they result in failure
 227                         for (final Entry<SrpIdNumber, SettableFuture<OperationResult>> e : this.waitingRequests.entrySet()) {
 228                                 LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey());
 229                                 e.getValue().set(OPERATION_NOACK);
 230                         }
 231                         this.waitingRequests.clear();
 232                 }
 233
 234                 @Override
 235                 public synchronized void onSessionDown(final PCEPSession session, final Exception e) {
 236                         LOG.warn("Session {} went down unexpectedly", e);
 237                         tearDown(session);
 238                 }
 239
 240                 @Override
 241                 public synchronized void onSessionTerminated(final PCEPSession session, final PCEPTerminationReason reason) {
 242                         LOG.info("Session {} terminated by peer with reason {}", session, reason);
 243                         tearDown(session);
 244                 }
 245
 246                 private InstanceIdentifier<ReportedLsps> lspIdentifier(final SymbolicPathName name) {
 247                         return InstanceIdentifier.builder(this.topologyAugment).child(PathComputationClient.class).child(ReportedLsps.class,
 248                                         new ReportedLspsKey(name.getPathName())).toInstance();
 249                 }
 250
 251                 @Override
 252                 public synchronized void onMessage(final PCEPSession session, final Message message) {
 253                         if (!(message instanceof PcrptMessage)) {
 254                                 LOG.info("Unhandled message {} on session {}", message, session);
 255                                 session.sendMessage(unhandledMessageError);
 256                         }
 257
 258                         final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.PcrptMessage rpt = ((PcrptMessage) message).getPcrptMessage();
 259
 260                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 261
 262                         for (final Reports r : rpt.getReports()) {
 263                                 final Lsp lsp = r.getLsp();
 264
 265                                 if (lsp.isSync() && !this.synced) {
 266                                         // Update synchronization flag
 267                                         this.synced = true;
 268                                         this.topologyAugmentBuilder.setPathComputationClient(this.pccBuilder.setStateSync(PccSyncState.Synchronized).build());
 269                                         trans.putOperationalData(this.topologyAugment, this.topologyAugmentBuilder.build());
 270                                         LOG.debug("Session {} achieved synchronized state", session);
 271                                 }
 272
 273                                 final Srp srp = r.getSrp();
 274                                 if (srp != null) {
 275                                         final SrpIdNumber id = srp.getOperationId();
 276                                         if (id.getValue() != 0) {
 277                                                 switch (lsp.getOperational()) {
 278                                                 case Active:
 279                                                 case Down:
 280                                                 case Up:
 281                                                         final SettableFuture<OperationResult> p = this.waitingRequests.remove(id);
 282                                                         if (p != null) {
 283                                                                 LOG.debug("Request {} resulted in LSP operational state {}", id, lsp.getOperational());
 284                                                                 p.set(OPERATION_SUCCESS);
 285                                                         } else {
 286                                                                 LOG.warn("Request ID {} not found in outstanding DB", id);
 287                                                         }
 288                                                         break;
 289                                                 case GoingDown:
 290                                                 case GoingUp:
 291                                                         // These are transitive states, so
 292                                                         break;
 293                                                 }
 294                                         }
 295                                 }
 296
 297                                 final PlspId id = lsp.getPlspId();
 298                                 if (lsp.isRemove()) {
 299                                         final SymbolicPathName name = this.lsps.remove(id);
 300                                         if (name != null) {
 301                                                 trans.removeOperationalData(lspIdentifier(name));
 302                                         }
 303
 304                                         LOG.debug("LSP {} removed", lsp);
 305                                 } else {
 306                                         if (!this.lsps.containsKey(id)) {
 307                                                 LOG.debug("PLSPID {} not known yet, looking for a symbolic name", id);
 308
 309                                                 final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.lsp.Tlvs tlvs = r.getLsp().getTlvs();
 310                                                 final SymbolicPathName name = tlvs.getSymbolicPathName();
 311                                                 if (name == null) {
 312                                                         LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id);
 313                                                         // FIXME: BUG-189
 314                                                         continue;
 315                                                 }
 316                                                 this.lsps.put(id, name);
 317                                         }
 318
 319                                         final SymbolicPathName name = this.lsps.get(id);
 320                                         trans.putOperationalData(lspIdentifier(name), lsp);
 321
 322                                         LOG.debug("LSP {} updated", lsp);
 323                                 }
 324                         }
 325
 326                         Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
 327                                 @Override
 328                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 329                                         // Nothing to do
 330                                 }
 331
 332                                 @Override
 333                                 public void onFailure(final Throwable t) {
 334                                         LOG.error("Failed to update internal state for session {}, closing it", session, t);
 335                                         session.close(TerminationReason.Unknown);
 336                                 }
 337                         });
 338                 }
 339
 340                 private synchronized SrpIdNumber nextRequest() {
 341                         return new SrpIdNumber(this.requestId++);
 342                 }
 343
 344                 private synchronized void messageSendingComplete(final SrpIdNumber requestId, final io.netty.util.concurrent.Future<Void> future) {
 345                         final SettableFuture<OperationResult> promise = this.sendingRequests.remove(requestId);
 346
 347                         if (future.isSuccess()) {
 348                                 this.waitingRequests.put(requestId, promise);
 349                         } else {
 350                                 LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
 351                                 promise.set(OPERATION_UNSENT);
 352                         }
 353                 }
 354
 355                 private synchronized ListenableFuture<OperationResult> sendMessage(final Message message, final SrpIdNumber requestId) {
 356                         final io.netty.util.concurrent.Future<Void> f = this.session.sendMessage(message);
 357                         final SettableFuture<OperationResult> ret = SettableFuture.create();
 358
 359                         this.sendingRequests.put(requestId, ret);
 360
 361                         f.addListener(new FutureListener<Void>() {
 362                                 @Override
 363                                 public void operationComplete(final io.netty.util.concurrent.Future<Void> future) {
 364                                         messageSendingComplete(requestId, future);
 365                                 }
 366                         });
 367
 368                         return ret;
 369                 }
 370         }
 371
 372         private static final Logger LOG = LoggerFactory.getLogger(ServerSessionManager.class);
 373         private static final Pcerr unhandledMessageError = new PcerrBuilder().setPcerrMessage(
 374                         new PcerrMessageBuilder().setErrorType(null).build()).build();
 375         private static final MessageHeader messageHeader = new MessageHeader() {
 376                 private final ProtocolVersion version = new ProtocolVersion((short) 1);
 377
 378                 @Override
 379                 public Class<? extends DataContainer> getImplementedInterface() {
 380                         return MessageHeader.class;
 381                 }
 382
 383                 @Override
 384                 public ProtocolVersion getVersion() {
 385                         return this.version;
 386                 }
 387         };
 388
 389         private static final OperationResult OPERATION_NOACK = createOperationResult(FailureType.NoAck);
 390         private static final OperationResult OPERATION_SUCCESS = createOperationResult(null);
 391         private static final OperationResult OPERATION_UNSENT = createOperationResult(FailureType.Unsent);
 392
 393         private final Map<NodeId, SessionListener> nodes = new HashMap<>();
 394         private final InstanceIdentifier<Topology> topology;
 395         private final DataProviderService dataProvider;
 396
 397         public ServerSessionManager(final DataProviderService dataProvider, final InstanceIdentifier<Topology> topology) {
 398                 this.dataProvider = Preconditions.checkNotNull(dataProvider);
 399                 this.topology = Preconditions.checkNotNull(topology);
 400
 401                 // Make sure the topology does not exist
 402                 final Object c = dataProvider.readOperationalData(topology);
 403                 Preconditions.checkArgument(c == null, "Topology %s already exists", topology);
 404
 405                 // Now create the base topology
 406                 final TopologyKey k = InstanceIdentifier.keyOf(topology);
 407                 final DataModificationTransaction t = dataProvider.beginTransaction();
 408                 t.putOperationalData(
 409                                 topology,
 410                                 new TopologyBuilder().setKey(k).setTopologyId(k.getTopologyId()).setTopologyTypes(
 411                                                 new TopologyTypesBuilder().addAugmentation(TopologyTypes1.class,
 412                                                                 new TopologyTypes1Builder().setTopologyPcep(new TopologyPcepBuilder().build()).build()).build()).setNode(
 413                                                                                 new ArrayList<Node>()).build());
 414
 415                 Futures.addCallback(JdkFutureAdapters.listenInPoolThread(t.commit()),
 416                                 new FutureCallback<RpcResult<TransactionStatus>>() {
 417                         @Override
 418                         public void onSuccess(final RpcResult<TransactionStatus> result) {
 419                                 // Nothing to do
 420                         }
 421
 422                         @Override
 423                         public void onFailure(final Throwable t) {
 424                                 LOG.error("Failed to create topology {}", topology);
 425                         }
 426                 });
 427         }
 428
 429         @Override
 430         public PCEPSessionListener getSessionListener() {
 431                 return new SessionListener();
 432         }
 433
 434         synchronized ListenableFuture<OperationResult> realAddLsp(final AddLspArgs input) {
 435                 // Get the listener corresponding to the node
 436                 final SessionListener l = this.nodes.get(input.getNode());
 437                 if (l == null) {
 438                         LOG.debug("Session for node {} not found", input.getNode());
 439                         return Futures.immediateFuture(OPERATION_UNSENT);
 440                 }
 441
 442                 // Make sure there is no such LSP
 443                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 444                                 ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
 445                 if (this.dataProvider.readOperationalData(lsp) != null) {
 446                         LOG.debug("Node {} already contains lsp {} at {}", input.getNode(), input.getName(), lsp);
 447                         return Futures.immediateFuture(OPERATION_UNSENT);
 448                 }
 449
 450                 // Build the request
 451                 final RequestsBuilder rb = new RequestsBuilder((EndpointsObject) input.getArguments());
 452                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 453                 rb.setLsp(new LspBuilder().setAdministrative(input.getArguments().isAdministrative()).setDelegate(Boolean.TRUE).setTlvs(
 454                                 new TlvsBuilder().setSymbolicPathName(new SymbolicPathNameBuilder().setPathName(input.getName()).build()).build()).build());
 455
 456                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
 457                 ib.setRequests(ImmutableList.of(rb.build()));
 458
 459                 // Send the message
 460                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId());
 461         }
 462
 463         private static OperationResult createOperationResult(final FailureType type) {
 464                 return new OperationResult() {
 465                         @Override
 466                         public Class<? extends DataContainer> getImplementedInterface() {
 467                                 return OperationResult.class;
 468                         }
 469
 470                         @Override
 471                         public FailureType getFailure() {
 472                                 return type;
 473                         }
 474                 };
 475         }
 476
 477         synchronized ListenableFuture<OperationResult> realRemoveLsp(final RemoveLspArgs input) {
 478                 // Get the listener corresponding to the node
 479                 final SessionListener l = this.nodes.get(input.getNode());
 480                 if (l == null) {
 481                         LOG.debug("Session for node {} not found", input.getNode());
 482                         return Futures.immediateFuture(OPERATION_UNSENT);
 483                 }
 484
 485                 // Make sure the LSP exists, we need it for PLSP-ID
 486                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 487                                 ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
 488                 final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
 489                 if (rep == null) {
 490                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 491                         return Futures.immediateFuture(OPERATION_UNSENT);
 492                 }
 493
 494                 // Build the request and send it
 495                 final RequestsBuilder rb = new RequestsBuilder();
 496                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).setFlags(new Flags(Boolean.TRUE)).build());
 497                 rb.setLsp(new LspBuilder().setRemove(Boolean.TRUE).setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 498
 499                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
 500                 ib.setRequests(ImmutableList.of(rb.build()));
 501                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId());
 502         }
 503
 504         synchronized ListenableFuture<OperationResult> realUpdateLsp(final UpdateLspArgs input) {
 505                 // Get the listener corresponding to the node
 506                 final SessionListener l = this.nodes.get(input.getNode());
 507                 if (l == null) {
 508                         LOG.debug("Session for node {} not found", input.getNode());
 509                         return Futures.immediateFuture(OPERATION_UNSENT);
 510                 }
 511
 512                 // Make sure the LSP exists
 513                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 514                                 ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
 515                 final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
 516                 if (rep == null) {
 517                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 518                         return Futures.immediateFuture(OPERATION_UNSENT);
 519                 }
 520
 521                 // Build the PCUpd request and send it
 522                 final UpdatesBuilder rb = new UpdatesBuilder();
 523                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 524                 rb.setLsp(new LspBuilder().setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 525                 final PathBuilder pb = new PathBuilder();
 526                 rb.setPath(pb.setEro(input.getArguments().getEro()).build());
 527
 528                 final PcupdMessageBuilder ub = new PcupdMessageBuilder(messageHeader);
 529                 ub.setUpdates(ImmutableList.of(rb.build()));
 530                 return l.sendMessage(new PcupdBuilder().setPcupdMessage(ub.build()).build(), rb.getSrp().getOperationId());
 531         }
 532
 533         synchronized ListenableFuture<OperationResult> realEnsureLspOperational(final EnsureLspOperationalInput input) {
 534                 // Get the listener corresponding to the node
 535                 final SessionListener l = this.nodes.get(input.getNode());
 536                 if (l == null) {
 537                         LOG.debug("Session for node {} not found", input.getNode());
 538                         return Futures.immediateFuture(OPERATION_UNSENT);
 539                 }
 540
 541                 // Make sure the LSP exists
 542                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 543                                 ReportedLsps.class, new ReportedLspsKey(input.getName())).toInstance();
 544                 LOG.debug("Checking if LSP {} has operational state {}", lsp, input.getArguments().getOperational());
 545                 final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
 546                 if (rep == null) {
 547                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 548                         return Futures.immediateFuture(OPERATION_UNSENT);
 549                 }
 550
 551                 if (rep.getLsp().getOperational().equals(input.getArguments().getOperational())) {
 552                         return Futures.immediateFuture(OPERATION_SUCCESS);
 553                 } else {
 554                         return Futures.immediateFuture(OPERATION_UNSENT);
 555                 }
 556         }
 557
 558         @Override
 559         public void close() throws InterruptedException, ExecutionException {
 560                 final DataModificationTransaction t = this.dataProvider.beginTransaction();
 561                 t.removeOperationalData(this.topology);
 562                 t.commit().get();
 563         }
 564 }