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.GlobalEventExecutor;
  13 import io.netty.util.concurrent.Promise;
  14
  15 import java.net.InetAddress;
  16 import java.util.HashMap;
  17 import java.util.Map;
  18 import java.util.Map.Entry;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.Future;
  21
  22 import javax.annotation.concurrent.GuardedBy;
  23
  24 import org.opendaylight.bgpcep.programming.spi.ExecutionResult;
  25 import org.opendaylight.bgpcep.programming.spi.InstructionExecutor;
  26 import org.opendaylight.bgpcep.programming.spi.InstructionScheduler;
  27 import org.opendaylight.bgpcep.programming.spi.SuccessfulRpcResult;
  28 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  29 import org.opendaylight.controller.sal.binding.api.data.DataModificationTransaction;
  30 import org.opendaylight.controller.sal.binding.api.data.DataProviderService;
  31 import org.opendaylight.protocol.framework.SessionListenerFactory;
  32 import org.opendaylight.protocol.pcep.PCEPSession;
  33 import org.opendaylight.protocol.pcep.PCEPSessionListener;
  34 import org.opendaylight.protocol.pcep.PCEPTerminationReason;
  35 import org.opendaylight.protocol.pcep.TerminationReason;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.Pcerr;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcerrBuilder;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcinitiateBuilder;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcupdBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.MessageHeader;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.PcrptMessage;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.PlspId;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.ProtocolVersion;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.SrpIdNumber;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.lsp.object.TlvsBuilder;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.open.object.open.Tlvs;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcerr.message.PcerrMessageBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcinitiate.message.PcinitiateMessageBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcinitiate.message.pcinitiate.message.RequestsBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcinitiate.message.pcinitiate.message.requests.LspBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.pcrpt.message.Reports;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcrpt.message.pcrpt.message.reports.Lsp;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.PcupdMessageBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.UpdatesBuilder;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.updates.PathBuilder;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.Srp;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.Srp.Flags;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.srp.object.SrpBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.stateful.capability.tlv.Stateful;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathName;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.symbolic.path.name.tlv.SymbolicPathNameBuilder;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev130930.InstructionStatus;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev130930.instruction.status.changed.Details;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev130930.submit.instruction.output.result.FailureBuilder;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.programming.rev130930.submit.instruction.output.result.failure.Failure;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspInput;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspOutput;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspOutputBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.NetworkTopologyPcepService;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1Builder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.PccSyncState;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspInput;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspOutput;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspOutputBuilder;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspInput;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspOutput;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspOutputBuilder;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClientBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLsps;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspsKey;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.StatefulTlvBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  85 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  86 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  87 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  88 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  89 import org.opendaylight.yangtools.yang.binding.DataContainer;
  90 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  91 import org.opendaylight.yangtools.yang.common.RpcResult;
  92 import org.slf4j.Logger;
  93 import org.slf4j.LoggerFactory;
  94
  95 import com.google.common.base.Preconditions;
  96 import com.google.common.collect.ImmutableList;
  97
  98 /**
  99  *
 100  */
 101 final class ServerSessionManager implements SessionListenerFactory<PCEPSessionListener>, NetworkTopologyPcepService {
 102         private static String createNodeId(final InetAddress addr) {
 103                 return "pcc://" + addr.getHostAddress();
 104         }
 105
 106         private final class SessionListener implements PCEPSessionListener {
 107                 private final Map<SrpIdNumber, Promise<ExecutionResult<Details>>> waitingRequests = new HashMap<>();
 108                 private final Map<SrpIdNumber, Promise<ExecutionResult<Details>>> sendingRequests = new HashMap<>();
 109                 private final Map<PlspId, SymbolicPathName> lsps = new HashMap<>();
 110                 private PathComputationClientBuilder pccBuilder;
 111                 private InstanceIdentifier<Node1> topologyAugment;
 112                 private InstanceIdentifier<Node> topologyNode;
 113                 private Node1Builder topologyAugmentBuilder;
 114                 private boolean ownsTopology = false;
 115                 private boolean synced = false;
 116                 private PCEPSession session;
 117                 private long requestId = 1;
 118                 private NodeId nodeId;
 119
 120                 final Node topologyNode(final DataModificationTransaction trans, final InetAddress address) {
 121                         final String pccId = createNodeId(address);
 122                         final Topology topo = (Topology) trans.readOperationalData(ServerSessionManager.this.topology);
 123
 124                         for (final Node n : topo.getNode()) {
 125                                 LOG.debug("Matching topology node {} to id {}", n, pccId);
 126                                 if (n.getNodeId().getValue().equals(pccId)) {
 127                                         return n;
 128                                 }
 129                         }
 130
 131                         /*
 132                          * We failed to find a matching node. Let's create a dynamic one
 133                          * and note that we are the owner (so we clean it up afterwards).
 134                          */
 135                         final NodeId id = new NodeId(pccId);
 136                         final NodeKey nk = new NodeKey(id);
 137                         final InstanceIdentifier<Node> nti = InstanceIdentifier.builder(ServerSessionManager.this.topology).node(Node.class, nk).toInstance();
 138
 139                         final Node ret = new NodeBuilder().setKey(nk).setNodeId(id).build();
 140
 141                         trans.putRuntimeData(nti, ret);
 142                         this.ownsTopology = true;
 143                         this.topologyNode = nti;
 144                         this.nodeId = id;
 145                         return ret;
 146                 }
 147
 148                 @Override
 149                 public synchronized void onSessionUp(final PCEPSession session) {
 150                         /*
 151                          * The session went up. Look up the router in Inventory model,
 152                          * create it if it is not there (marking that fact for later
 153                          * deletion), and mark it as synchronizing. Also create it in
 154                          * the topology model, with empty LSP list.
 155                          */
 156                         final InetAddress peerAddress = session.getRemoteAddress();
 157                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 158
 159                         final Node topoNode = topologyNode(trans, peerAddress);
 160                         LOG.debug("Peer {} resolved to topology node {}", peerAddress, topoNode);
 161
 162                         // Our augmentation in the topology node
 163                         this.pccBuilder = new PathComputationClientBuilder();
 164
 165                         final Tlvs tlvs = session.getRemoteTlvs();
 166                         final Stateful stateful = tlvs.getStateful();
 167                         if (stateful != null) {
 168                                 this.pccBuilder.setStatefulTlv(new StatefulTlvBuilder(tlvs).build());
 169                                 this.pccBuilder.setStateSync(PccSyncState.InitialResync);
 170                         }
 171
 172                         this.topologyAugmentBuilder = new Node1Builder().setPathComputationClient(this.pccBuilder.build());
 173                         this.topologyAugment = InstanceIdentifier.builder(this.topologyNode).node(Node1.class).toInstance();
 174                         trans.putRuntimeData(this.topologyAugment, this.topologyAugmentBuilder.build());
 175
 176                         // All set, commit the modifications
 177                         final Future<RpcResult<TransactionStatus>> s = trans.commit();
 178
 179                         /*
 180                          * FIXME: once this Future is listenable, attach to it so we can
 181                          *        do cleanup if the commit fails. For now we force a commit.
 182                          */
 183                         try {
 184                                 s.get();
 185                         } catch (InterruptedException | ExecutionException e) {
 186                                 LOG.error("Failed to update internal state for session {}, terminating it", session, e);
 187                                 session.close(TerminationReason.Unknown);
 188                                 return;
 189                         }
 190
 191                         ServerSessionManager.this.nodes.put(this.nodeId, this);
 192                         this.session = session;
 193                         LOG.info("Session with {} attached to topology node {}", session.getRemoteAddress(), topoNode.getNodeId());
 194                 }
 195
 196                 @GuardedBy("this")
 197                 private void tearDown(final PCEPSession session) {
 198                         this.session = null;
 199                         ServerSessionManager.this.nodes.remove(this.nodeId);
 200
 201                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 202
 203                         // The session went down. Undo all the Topology changes we have done.
 204                         trans.removeRuntimeData(this.topologyAugment);
 205                         if (this.ownsTopology) {
 206                                 trans.removeRuntimeData(this.topologyNode);
 207                         }
 208
 209                         /*
 210                          * FIXME: once this Future is listenable, attach to it so we can
 211                          *        do cleanup if the commit fails. For now we force a commit.
 212                          */
 213                         final Future<RpcResult<TransactionStatus>> s = trans.commit();
 214                         try {
 215                                 s.get();
 216                         } catch (InterruptedException | ExecutionException e) {
 217                                 LOG.error("Failed to cleanup internal state for session {}", session, e);
 218                         }
 219
 220                         // Clear all requests which have not been sent to the peer: they result in cancellation
 221                         for (final Entry<SrpIdNumber, Promise<ExecutionResult<Details>>> e : this.sendingRequests.entrySet()) {
 222                                 LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey());
 223                                 e.getValue().setSuccess(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 224                         }
 225                         this.sendingRequests.clear();
 226
 227                         // CLear all requests which have not been acked by the peer: they result in failure
 228                         for (final Entry<SrpIdNumber, Promise<ExecutionResult<Details>>> e : this.waitingRequests.entrySet()) {
 229                                 LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey());
 230                                 e.getValue().setSuccess(new ExecutionResult<Details>(InstructionStatus.Failed, null));
 231                         }
 232                         this.waitingRequests.clear();
 233                 }
 234
 235                 @Override
 236                 public synchronized void onSessionDown(final PCEPSession session, final Exception e) {
 237                         LOG.warn("Session {} went down unexpectedly", e);
 238                         tearDown(session);
 239                 }
 240
 241                 @Override
 242                 public synchronized void onSessionTerminated(final PCEPSession session, final PCEPTerminationReason reason) {
 243                         LOG.info("Session {} terminated by peer with reason {}", session, reason);
 244                         tearDown(session);
 245                 }
 246
 247                 private InstanceIdentifier<ReportedLsps> lspIdentifier(final SymbolicPathName name) {
 248                         return InstanceIdentifier.builder(this.topologyAugment).node(ReportedLsps.class, 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.putRuntimeData(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 Promise<ExecutionResult<Details>> p = this.waitingRequests.remove(id);
 282                                                         if (p != null) {
 283                                                                 LOG.debug("Request {} resulted in LSP operational state {}", id, lsp.getOperational());
 284                                                                 p.setSuccess(new ExecutionResult<Details>(InstructionStatus.Successful, null));
 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.removeRuntimeData(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.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");
 313                                                         // TODO: what should we do here?
 314                                                         continue;
 315                                                 }
 316                                         }
 317
 318                                         final SymbolicPathName name = this.lsps.get(id);
 319                                         trans.putRuntimeData(lspIdentifier(name), lsp);
 320
 321                                         LOG.debug("LSP {} updated");
 322                                 }
 323                         }
 324
 325                         /*
 326                          * FIXME: once this Future is listenable, attach to it so we can
 327                          *        do cleanup if the commit fails. For now we force a commit.
 328                          */
 329                         final Future<RpcResult<TransactionStatus>> s = trans.commit();
 330                         try {
 331                                 s.get();
 332                         } catch (InterruptedException | ExecutionException e) {
 333                                 LOG.error("Failed to update internal state for session {}, closing it", session, e);
 334                                 session.close(TerminationReason.Unknown);
 335                         }
 336                 }
 337
 338                 private synchronized SrpIdNumber nextRequest() {
 339                         return new SrpIdNumber(this.requestId++);
 340                 }
 341
 342                 private synchronized void messageSendingComplete(final SrpIdNumber requestId, final io.netty.util.concurrent.Future<Void> future) {
 343                         final Promise<ExecutionResult<Details>> promise = this.sendingRequests.remove(requestId);
 344
 345                         if (future.isSuccess()) {
 346                                 this.waitingRequests.put(requestId, promise);
 347                         } else {
 348                                 LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
 349                                 promise.setSuccess(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 350                         }
 351                 }
 352
 353                 private synchronized io.netty.util.concurrent.Future<ExecutionResult<Details>> sendMessage(final Message message,
 354                                 final SrpIdNumber requestId) {
 355                         final io.netty.util.concurrent.Future<Void> f = this.session.sendMessage(message);
 356                         final Promise<ExecutionResult<Details>> ret = exec.newPromise();
 357
 358                         this.sendingRequests.put(requestId, ret);
 359
 360                         f.addListener(new FutureListener<Void>() {
 361                                 @Override
 362                                 public void operationComplete(final io.netty.util.concurrent.Future<Void> future) {
 363                                         messageSendingComplete(requestId, future);
 364                                 }
 365                         });
 366
 367                         return ret;
 368                 }
 369         }
 370
 371         private static final Logger LOG = LoggerFactory.getLogger(ServerSessionManager.class);
 372         private static final Pcerr unhandledMessageError = new PcerrBuilder().setPcerrMessage(
 373                         new PcerrMessageBuilder().setErrorType(null).build()).build();
 374         private static final MessageHeader messageHeader = new MessageHeader() {
 375                 private final ProtocolVersion version = new ProtocolVersion((short) 1);
 376
 377                 @Override
 378                 public Class<? extends DataContainer> getImplementedInterface() {
 379                         return MessageHeader.class;
 380                 }
 381
 382                 @Override
 383                 public ProtocolVersion getVersion() {
 384                         return this.version;
 385                 }
 386         };
 387         private static final EventExecutor exec = GlobalEventExecutor.INSTANCE;
 388         private final Map<NodeId, SessionListener> nodes = new HashMap<>();
 389         private final InstanceIdentifier<Topology> topology;
 390         private final DataProviderService dataProvider;
 391         private final InstructionScheduler scheduler;
 392
 393         public ServerSessionManager(final InstructionScheduler scheduler, final DataProviderService dataProvider,
 394                         final InstanceIdentifier<Topology> topology) {
 395                 this.dataProvider = Preconditions.checkNotNull(dataProvider);
 396                 this.topology = Preconditions.checkNotNull(topology);
 397                 this.scheduler = Preconditions.checkNotNull(scheduler);
 398         }
 399
 400         @Override
 401         public PCEPSessionListener getSessionListener() {
 402                 return new SessionListener();
 403         }
 404
 405         private synchronized io.netty.util.concurrent.Future<ExecutionResult<Details>> realAddLsp(final AddLspInput input) {
 406                 // Get the listener corresponding to the node
 407                 final SessionListener l = this.nodes.get(input.getNode());
 408                 if (l == null) {
 409                         LOG.debug("Session for node {} not found", input.getNode());
 410                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 411                 }
 412
 413                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
 414                                 new ReportedLspsKey(input.getName())).toInstance();
 415                 if (this.dataProvider.readOperationalData(lsp) != null) {
 416                         LOG.debug("Node {} already contains lsp {} at {}", input.getNode(), input.getName(), lsp);
 417                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 418                 }
 419
 420                 final RequestsBuilder rb = new RequestsBuilder(input.getArguments());
 421                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 422                 rb.setLsp(new LspBuilder().setAdministrative(input.getArguments().isAdministrative()).setTlvs(
 423                                 new TlvsBuilder().setSymbolicPathName(new SymbolicPathNameBuilder().setPathName(input.getName()).build()).build()).build());
 424
 425                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
 426                 ib.setRequests(ImmutableList.of(rb.build()));
 427
 428                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId());
 429         }
 430
 431         private synchronized io.netty.util.concurrent.Future<ExecutionResult<Details>> realRemoveLsp(final RemoveLspInput input) {
 432                 // Get the listener corresponding to the node
 433                 final SessionListener l = this.nodes.get(input.getNode());
 434                 if (l == null) {
 435                         LOG.debug("Session for node {} not found", input.getNode());
 436                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 437                 }
 438
 439                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
 440                                 new ReportedLspsKey(input.getName())).toInstance();
 441                 final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
 442                 if (rep == null) {
 443                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 444                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 445                 }
 446
 447                 final RequestsBuilder rb = new RequestsBuilder();
 448                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).setFlags(new Flags(Boolean.TRUE)).build());
 449                 rb.setLsp(new LspBuilder().setRemove(Boolean.TRUE).setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 450
 451                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(messageHeader);
 452                 ib.setRequests(ImmutableList.of(rb.build()));
 453                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId());
 454         }
 455
 456         private synchronized io.netty.util.concurrent.Future<ExecutionResult<Details>> realUpdateLsp(final UpdateLspInput input) {
 457                 // Get the listener corresponding to the node
 458                 final SessionListener l = this.nodes.get(input.getNode());
 459                 if (l == null) {
 460                         LOG.debug("Session for node {} not found", input.getNode());
 461                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 462                 }
 463
 464                 final InstanceIdentifier<ReportedLsps> lsp = InstanceIdentifier.builder(l.topologyAugment).node(ReportedLsps.class,
 465                                 new ReportedLspsKey(input.getName())).toInstance();
 466                 final ReportedLsps rep = (ReportedLsps) this.dataProvider.readOperationalData(lsp);
 467                 if (rep == null) {
 468                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 469                         return exec.newSucceededFuture(new ExecutionResult<Details>(InstructionStatus.Cancelled, null));
 470                 }
 471
 472                 final UpdatesBuilder rb = new UpdatesBuilder();
 473                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 474                 rb.setLsp(new org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcupd.message.pcupd.message.updates.LspBuilder().setPlspId(
 475                                 rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 476                 rb.setPath(new PathBuilder(input.getArguments()).build());
 477
 478                 final PcupdMessageBuilder ub = new PcupdMessageBuilder(messageHeader);
 479                 ub.setUpdates(ImmutableList.of(rb.build()));
 480                 return l.sendMessage(new PcupdBuilder().setPcupdMessage(ub.build()).build(), rb.getSrp().getOperationId());
 481         }
 482
 483         @Override
 484         public Future<RpcResult<AddLspOutput>> addLsp(final AddLspInput input) {
 485                 Preconditions.checkArgument(input.getNode() != null);
 486                 Preconditions.checkArgument(input.getName() != null);
 487
 488                 final InstructionExecutor e = new InstructionExecutor() {
 489                         @Override
 490                         public io.netty.util.concurrent.Future<ExecutionResult<Details>> execute() {
 491                                 return realAddLsp(input);
 492                         }
 493                 };
 494
 495                 final Failure f = this.scheduler.submitInstruction(input, e);
 496                 final AddLspOutputBuilder b = new AddLspOutputBuilder();
 497                 if (f != null) {
 498                         b.setResult(new FailureBuilder().setFailure(f).build());
 499                 }
 500
 501                 final RpcResult<AddLspOutput> res = SuccessfulRpcResult.create(b.build());
 502                 return exec.newSucceededFuture(res);
 503         }
 504
 505         @Override
 506         public Future<RpcResult<RemoveLspOutput>> removeLsp(final RemoveLspInput input) {
 507                 Preconditions.checkArgument(input.getNode() != null);
 508                 Preconditions.checkArgument(input.getName() != null);
 509
 510                 final InstructionExecutor e = new InstructionExecutor() {
 511                         @Override
 512                         public io.netty.util.concurrent.Future<ExecutionResult<Details>> execute() {
 513                                 return realRemoveLsp(input);
 514                         }
 515                 };
 516
 517                 final Failure f = this.scheduler.submitInstruction(input, e);
 518                 final RemoveLspOutputBuilder b = new RemoveLspOutputBuilder();
 519                 if (f != null) {
 520                         b.setResult(new FailureBuilder().setFailure(f).build());
 521                 }
 522
 523                 final RpcResult<RemoveLspOutput> res = SuccessfulRpcResult.create(b.build());
 524                 return exec.newSucceededFuture(res);
 525         }
 526
 527         @Override
 528         public Future<RpcResult<UpdateLspOutput>> updateLsp(final UpdateLspInput input) {
 529                 Preconditions.checkArgument(input.getNode() != null);
 530                 Preconditions.checkArgument(input.getName() != null);
 531
 532                 final InstructionExecutor e = new InstructionExecutor() {
 533                         @Override
 534                         public io.netty.util.concurrent.Future<ExecutionResult<Details>> execute() {
 535                                 return realUpdateLsp(input);
 536                         }
 537                 };
 538
 539                 final Failure f = this.scheduler.submitInstruction(input, e);
 540                 final UpdateLspOutputBuilder b = new UpdateLspOutputBuilder();
 541                 if (f != null) {
 542                         b.setResult(new FailureBuilder().setFailure(f).build());
 543                 }
 544
 545                 final RpcResult<UpdateLspOutput> res = SuccessfulRpcResult.create(b.build());
 546                 return exec.newSucceededFuture(res);
 547         }
 548 }