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