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.Collections;
  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.TimeUnit;
  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.ietf.params.xml.ns.yang.ietf.inet.types.rev100924.IpAddressBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.PcinitiateBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.PcinitiateMessageBuilder;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.crabbe.initiated.rev131126.pcinitiate.message.pcinitiate.message.RequestsBuilder;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.PcrptMessage;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.PcupdBuilder;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.PlspId;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.SrpIdNumber;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.Tlvs2;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.lsp.object.Lsp;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.lsp.object.LspBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.lsp.object.lsp.TlvsBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.pcrpt.message.pcrpt.message.Reports;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.pcupd.message.PcupdMessageBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.pcupd.message.pcupd.message.UpdatesBuilder;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.pcupd.message.pcupd.message.updates.PathBuilder;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.srp.object.Srp;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.srp.object.SrpBuilder;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.stateful.capability.tlv.Stateful;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.symbolic.path.name.tlv.SymbolicPathName;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.symbolic.path.name.tlv.SymbolicPathNameBuilder;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.Pcerr;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.PcerrBuilder;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.MessageHeader;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.ProtocolVersion;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.open.object.open.Tlvs;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.pcerr.message.PcerrMessageBuilder;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.AddLspArgs;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.EnsureLspOperationalInput;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.FailureType;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.Node1Builder;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.OperationResult;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.PccSyncState;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.RemoveLspArgs;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.TopologyTypes1Builder;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.UpdateLspArgs;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.lsp.metadata.Metadata;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClient;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.PathComputationClientBuilder;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLsp;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.ReportedLspKey;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.pcep.client.attributes.path.computation.client.StatefulTlvBuilder;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev131024.topology.pcep.type.TopologyPcepBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
  79 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.Topology;
  80 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.TopologyKey;
  82 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node;
  83 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.NodeKey;
  85 import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.TopologyTypesBuilder;
  86 import org.opendaylight.yangtools.yang.binding.DataContainer;
  87 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  88 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier.InstanceIdentifierBuilder;
  89 import org.opendaylight.yangtools.yang.common.RpcResult;
  90 import org.slf4j.Logger;
  91 import org.slf4j.LoggerFactory;
  92
  93 import com.google.common.base.Function;
  94 import com.google.common.base.Preconditions;
  95 import com.google.common.collect.Collections2;
  96 import com.google.common.collect.ImmutableList;
  97 import com.google.common.util.concurrent.FutureCallback;
  98 import com.google.common.util.concurrent.Futures;
  99 import com.google.common.util.concurrent.JdkFutureAdapters;
 100 import com.google.common.util.concurrent.ListenableFuture;
 101
 102 /**
 103  *
 104  */
 105 final class ServerSessionManager implements SessionListenerFactory<PCEPSessionListener>, AutoCloseable {
 106         private final class SessionListener implements PCEPSessionListener {
 107                 private final Map<SrpIdNumber, PCEPRequest> waitingRequests = new HashMap<>();
 108                 private final Map<SrpIdNumber, PCEPRequest> 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 TopologyNodeState nodeState;
 119
 120                 private 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                                         this.topologyNode =
 128                                                         InstanceIdentifier.builder(ServerSessionManager.this.topology).child(Node.class, n.getKey()).toInstance();
 129                                         LOG.debug("Reusing topology node {} for id {} at {}", n, pccId, this.topologyNode);
 130                                         return n;
 131                                 }
 132                         }
 133
 134                         /*
 135                          * We failed to find a matching node. Let's create a dynamic one
 136                          * and note that we are the owner (so we clean it up afterwards).
 137                          */
 138                         final NodeId id = new NodeId(pccId);
 139                         final NodeKey nk = new NodeKey(id);
 140                         final InstanceIdentifier<Node> nti = InstanceIdentifier.builder(ServerSessionManager.this.topology).child(Node.class, nk).toInstance();
 141
 142                         final Node ret = new NodeBuilder().setKey(nk).setNodeId(id).build();
 143
 144                         trans.putOperationalData(nti, ret);
 145                         LOG.debug("Created topology node {} for id {} at {}", ret, pccId, nti);
 146                         this.ownsTopology = true;
 147                         this.topologyNode = nti;
 148                         return ret;
 149                 }
 150
 151                 @Override
 152                 public synchronized void onSessionUp(final PCEPSession session) {
 153                         /*
 154                          * The session went up. Look up the router in Inventory model,
 155                          * create it if it is not there (marking that fact for later
 156                          * deletion), and mark it as synchronizing. Also create it in
 157                          * the topology model, with empty LSP list.
 158                          */
 159                         final InetAddress peerAddress = session.getRemoteAddress();
 160                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 161
 162                         final Node topoNode = topologyNode(trans, peerAddress);
 163                         LOG.debug("Peer {} resolved to topology node {}", peerAddress, topoNode);
 164
 165                         // Our augmentation in the topology node
 166                         this.synced = false;
 167                         this.pccBuilder = new PathComputationClientBuilder();
 168                         this.pccBuilder.setIpAddress(IpAddressBuilder.getDefaultInstance(peerAddress.getHostAddress()));
 169
 170                         final Tlvs tlvs = session.getRemoteTlvs();
 171                         if (tlvs.getAugmentation(Tlvs2.class) != null) {
 172                                 final Stateful stateful = tlvs.getAugmentation(Tlvs2.class).getStateful();
 173                                 if (stateful != null) {
 174                                         this.pccBuilder.setReportedLsp(Collections.<ReportedLsp> emptyList());
 175                                         this.pccBuilder.setStatefulTlv(new StatefulTlvBuilder().setStateful(stateful).build());
 176                                         this.pccBuilder.setStateSync(PccSyncState.InitialResync);
 177                                 } else {
 178                                         LOG.debug("Peer {} does not advertise stateful TLV", peerAddress);
 179                                 }
 180                         } else {
 181                                 LOG.debug("Peer {} does not advertise stateful TLV", peerAddress);
 182                         }
 183
 184                         this.topologyAugmentBuilder = new Node1Builder().setPathComputationClient(this.pccBuilder.build());
 185                         this.topologyAugment = InstanceIdentifier.builder(this.topologyNode).augmentation(Node1.class).toInstance();
 186                         final Node1 ta = topologyAugmentBuilder.build();
 187
 188                         trans.putOperationalData(this.topologyAugment, ta);
 189                         LOG.debug("Peer data {} set to {}", this.topologyAugment, ta);
 190
 191                         // All set, commit the modifications
 192                         final ListenableFuture<RpcResult<TransactionStatus>> f = JdkFutureAdapters.listenInPoolThread(trans.commit());
 193                         Futures.addCallback(f, new FutureCallback<RpcResult<TransactionStatus>>() {
 194                                 @Override
 195                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 196                                         LOG.trace("Internal state for session {} updated successfully", session);
 197                                 }
 198
 199                                 @Override
 200                                 public void onFailure(final Throwable t) {
 201                                         LOG.error("Failed to update internal state for session {}, terminating it", session, t);
 202                                         session.close(TerminationReason.Unknown);
 203                                 }
 204                         });
 205
 206                         this.nodeState = takeNodeState(topoNode.getNodeId(), this);
 207                         this.session = session;
 208                         LOG.info("Session with {} attached to topology node {}", session.getRemoteAddress(), topoNode.getNodeId());
 209                 }
 210
 211                 @GuardedBy("this")
 212                 private void tearDown(final PCEPSession session) {
 213                         releaseNodeState(this.nodeState);
 214                         this.nodeState = null;
 215                         this.session = null;
 216
 217                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 218
 219                         // The session went down. Undo all the Topology changes we have done.
 220                         trans.removeOperationalData(this.topologyAugment);
 221                         if (this.ownsTopology) {
 222                                 trans.removeOperationalData(this.topologyNode);
 223                         }
 224
 225                         Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
 226                                 @Override
 227                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 228                                         LOG.trace("Internal state for session {} cleaned up successfully", session);
 229                                 }
 230
 231                                 @Override
 232                                 public void onFailure(final Throwable t) {
 233                                         LOG.error("Failed to cleanup internal state for session {}", session, t);
 234                                 }
 235                         });
 236
 237                         // Clear all requests which have not been sent to the peer: they result in cancellation
 238                         for (final Entry<SrpIdNumber, PCEPRequest> e : this.sendingRequests.entrySet()) {
 239                                 LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey());
 240                                 e.getValue().setResult(OPERATION_UNSENT);
 241                         }
 242                         this.sendingRequests.clear();
 243
 244                         // CLear all requests which have not been acked by the peer: they result in failure
 245                         for (final Entry<SrpIdNumber, PCEPRequest> e : this.waitingRequests.entrySet()) {
 246                                 LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey());
 247                                 e.getValue().setResult(OPERATION_NOACK);
 248                         }
 249                         this.waitingRequests.clear();
 250                 }
 251
 252                 @Override
 253                 public synchronized void onSessionDown(final PCEPSession session, final Exception e) {
 254                         LOG.warn("Session {} went down unexpectedly", e);
 255                         tearDown(session);
 256                 }
 257
 258                 @Override
 259                 public synchronized void onSessionTerminated(final PCEPSession session, final PCEPTerminationReason reason) {
 260                         LOG.info("Session {} terminated by peer with reason {}", session, reason);
 261                         tearDown(session);
 262                 }
 263
 264                 private InstanceIdentifierBuilder<ReportedLsp> lspIdentifier(final ReportedLspKey key) {
 265                         return InstanceIdentifier.builder(this.topologyAugment).child(PathComputationClient.class).child(ReportedLsp.class, key);
 266                 }
 267
 268                 @Override
 269                 public synchronized void onMessage(final PCEPSession session, final Message message) {
 270                         if (!(message instanceof PcrptMessage)) {
 271                                 LOG.info("Unhandled message {} on session {}", message, session);
 272                                 session.sendMessage(UNHANDLED_MESSAGE_ERROR);
 273                         }
 274
 275                         final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.pcrpt.message.PcrptMessage rpt = ((PcrptMessage) message).getPcrptMessage();
 276
 277                         final DataModificationTransaction trans = ServerSessionManager.this.dataProvider.beginTransaction();
 278
 279                         for (final Reports r : rpt.getReports()) {
 280                                 final Lsp lsp = r.getLsp();
 281
 282                                 if (!lsp.isSync() && !this.synced) {
 283                                         // Update synchronization flag
 284                                         this.synced = true;
 285                                         this.topologyAugmentBuilder.setPathComputationClient(this.pccBuilder.setStateSync(PccSyncState.Synchronized).build());
 286                                         final Node1 ta = this.topologyAugmentBuilder.build();
 287                                         trans.putOperationalData(this.topologyAugment, ta);
 288                                         LOG.debug("Peer data {} set to {}", this.topologyAugment, ta);
 289
 290                                         // The node has completed synchronization, cleanup metadata no longer reported back
 291                                         this.nodeState.cleanupExcept(Collections2.transform(
 292                                                         this.lsps.values(),
 293                                                         new Function<SymbolicPathName, org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.SymbolicPathName>() {
 294                                                                 @Override
 295                                                                 public org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.SymbolicPathName apply(
 296                                                                                 final SymbolicPathName input) {
 297                                                                         return input.getPathName();
 298                                                                 }
 299                                                         }));
 300                                         LOG.debug("Session {} achieved synchronized state", session);
 301                                         continue;
 302                                 }
 303
 304                                 final ReportedLspBuilder rlb = new ReportedLspBuilder(r);
 305                                 boolean solicited = false;
 306
 307                                 final Srp srp = r.getSrp();
 308                                 if (srp != null) {
 309                                         final SrpIdNumber id = srp.getOperationId();
 310                                         if (id.getValue() != 0) {
 311                                                 solicited = true;
 312
 313                                                 switch (lsp.getOperational()) {
 314                                                 case Active:
 315                                                 case Down:
 316                                                 case Up:
 317                                                         final PCEPRequest req = this.waitingRequests.remove(id);
 318                                                         if (req != null) {
 319                                                                 LOG.debug("Request {} resulted in LSP operational state {}", id, lsp.getOperational());
 320                                                                 rlb.setMetadata(req.getMetadata());
 321                                                                 req.setResult(OPERATION_SUCCESS);
 322                                                         } else {
 323                                                                 LOG.warn("Request ID {} not found in outstanding DB", id);
 324                                                         }
 325                                                         break;
 326                                                 case GoingDown:
 327                                                 case GoingUp:
 328                                                         // These are transitive states, so we don't have to do anything, as they will be followed
 329                                                         // up...
 330                                                         break;
 331                                                 }
 332                                         }
 333                                 }
 334
 335                                 final PlspId id = lsp.getPlspId();
 336                                 if (lsp.isRemove()) {
 337                                         final SymbolicPathName name = this.lsps.remove(id);
 338                                         if (name != null) {
 339                                                 this.nodeState.removeLspMetadata(name.getPathName());
 340                                                 trans.removeOperationalData(lspIdentifier(new ReportedLspKey(name.getPathName())).build());
 341                                         }
 342
 343                                         LOG.debug("LSP {} removed", lsp);
 344                                 } else {
 345                                         if (!this.lsps.containsKey(id)) {
 346                                                 LOG.debug("PLSPID {} not known yet, looking for a symbolic name", id);
 347
 348                                                 final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev131222.lsp.object.lsp.Tlvs tlvs = r.getLsp().getTlvs();
 349                                                 if (tlvs != null && tlvs.getSymbolicPathName() != null) {
 350                                                         this.lsps.put(id, tlvs.getSymbolicPathName());
 351                                                 } else {
 352                                                         LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id);
 353                                                         continue;
 354                                                 }
 355                                         }
 356
 357                                         rlb.setKey(new ReportedLspKey(this.lsps.get(id).getPathName()));
 358
 359                                         // If this is an unsolicited update. We need to make sure we retain the metadata already present
 360                                         if (solicited) {
 361                                                 this.nodeState.setLspMetadata(rlb.getName(), rlb.getMetadata());
 362                                         } else {
 363                                                 rlb.setMetadata(this.nodeState.getLspMetadata(rlb.getName()));
 364                                         }
 365
 366                                         trans.putOperationalData(lspIdentifier(rlb.getKey()).build(), rlb.build());
 367
 368                                         LOG.debug("LSP {} updated", lsp);
 369                                 }
 370                         }
 371
 372                         Futures.addCallback(JdkFutureAdapters.listenInPoolThread(trans.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
 373                                 @Override
 374                                 public void onSuccess(final RpcResult<TransactionStatus> result) {
 375                                         LOG.trace("Internal state for session {} updated successfully", session);
 376                                 }
 377
 378                                 @Override
 379                                 public void onFailure(final Throwable t) {
 380                                         LOG.error("Failed to update internal state for session {}, closing it", session, t);
 381                                         session.close(TerminationReason.Unknown);
 382                                 }
 383                         });
 384                 }
 385
 386                 private synchronized SrpIdNumber nextRequest() {
 387                         return new SrpIdNumber(this.requestId++);
 388                 }
 389
 390                 private synchronized void messageSendingComplete(final SrpIdNumber requestId, final io.netty.util.concurrent.Future<Void> future) {
 391                         final PCEPRequest req = this.sendingRequests.remove(requestId);
 392
 393                         if (future.isSuccess()) {
 394                                 this.waitingRequests.put(requestId, req);
 395                         } else {
 396                                 LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause());
 397                                 req.setResult(OPERATION_UNSENT);
 398                         }
 399                 }
 400
 401                 private synchronized ListenableFuture<OperationResult> sendMessage(final Message message, final SrpIdNumber requestId,
 402                                 final Metadata metadata) {
 403                         final io.netty.util.concurrent.Future<Void> f = this.session.sendMessage(message);
 404                         final PCEPRequest req = new PCEPRequest(metadata);
 405
 406                         this.sendingRequests.put(requestId, req);
 407
 408                         f.addListener(new FutureListener<Void>() {
 409                                 @Override
 410                                 public void operationComplete(final io.netty.util.concurrent.Future<Void> future) {
 411                                         messageSendingComplete(requestId, future);
 412                                 }
 413                         });
 414
 415                         return req.getFuture();
 416                 }
 417         }
 418
 419         private static final Logger LOG = LoggerFactory.getLogger(ServerSessionManager.class);
 420         private static final Pcerr UNHANDLED_MESSAGE_ERROR = new PcerrBuilder().setPcerrMessage(
 421                         new PcerrMessageBuilder().setErrorType(null).build()).build();
 422         private static final MessageHeader MESSAGE_HEADER = new MessageHeader() {
 423                 private final ProtocolVersion version = new ProtocolVersion((short) 1);
 424
 425                 @Override
 426                 public Class<? extends DataContainer> getImplementedInterface() {
 427                         return MessageHeader.class;
 428                 }
 429
 430                 @Override
 431                 public ProtocolVersion getVersion() {
 432                         return this.version;
 433                 }
 434         };
 435
 436         private static final long DEFAULT_HOLD_STATE_NANOS = TimeUnit.MINUTES.toNanos(5);
 437         private static final OperationResult OPERATION_NOACK = createOperationResult(FailureType.NoAck);
 438         private static final OperationResult OPERATION_SUCCESS = createOperationResult(null);
 439         private static final OperationResult OPERATION_UNSENT = createOperationResult(FailureType.Unsent);
 440
 441         private final Map<NodeId, SessionListener> nodes = new HashMap<>();
 442         private final Map<NodeId, TopologyNodeState> state = new HashMap<>();
 443         private final InstanceIdentifier<Topology> topology;
 444         private final DataProviderService dataProvider;
 445
 446         public ServerSessionManager(final DataProviderService dataProvider, final InstanceIdentifier<Topology> topology) {
 447                 this.dataProvider = Preconditions.checkNotNull(dataProvider);
 448                 this.topology = Preconditions.checkNotNull(topology);
 449
 450                 // Make sure the topology does not exist
 451                 final Object c = dataProvider.readOperationalData(topology);
 452                 Preconditions.checkArgument(c == null, "Topology %s already exists", topology);
 453
 454                 // Now create the base topology
 455                 final TopologyKey k = InstanceIdentifier.keyOf(topology);
 456                 final DataModificationTransaction t = dataProvider.beginTransaction();
 457                 t.putOperationalData(
 458                                 topology,
 459                                 new TopologyBuilder().setKey(k).setTopologyId(k.getTopologyId()).setTopologyTypes(
 460                                                 new TopologyTypesBuilder().addAugmentation(TopologyTypes1.class,
 461                                                                 new TopologyTypes1Builder().setTopologyPcep(new TopologyPcepBuilder().build()).build()).build()).setNode(
 462                                                                                 new ArrayList<Node>()).build());
 463
 464                 Futures.addCallback(JdkFutureAdapters.listenInPoolThread(t.commit()), new FutureCallback<RpcResult<TransactionStatus>>() {
 465                         @Override
 466                         public void onSuccess(final RpcResult<TransactionStatus> result) {
 467                                 LOG.trace("Topology {} created successfully", topology);
 468                         }
 469
 470                         @Override
 471                         public void onFailure(final Throwable t) {
 472                                 LOG.error("Failed to create topology {}", topology);
 473                         }
 474                 });
 475         }
 476
 477         public void releaseNodeState(final TopologyNodeState nodeState) {
 478                 LOG.debug("Node {} unbound", nodeState.getNodeId());
 479                 this.nodes.remove(nodeState.getNodeId());
 480                 nodeState.released();
 481         }
 482
 483         private synchronized TopologyNodeState takeNodeState(final NodeId id, final SessionListener sessionListener) {
 484                 LOG.debug("Node {} bound to listener {}", id, sessionListener);
 485
 486                 TopologyNodeState ret = this.state.get(id);
 487                 if (ret == null) {
 488                         ret = new TopologyNodeState(id, DEFAULT_HOLD_STATE_NANOS);
 489                         this.state.put(id, ret);
 490                 }
 491
 492                 this.nodes.put(id, sessionListener);
 493                 ret.taken();
 494                 return ret;
 495         }
 496
 497         @Override
 498         public PCEPSessionListener getSessionListener() {
 499                 return new SessionListener();
 500         }
 501
 502         synchronized ListenableFuture<OperationResult> realAddLsp(final AddLspArgs input) {
 503                 // Get the listener corresponding to the node
 504                 final SessionListener l = this.nodes.get(input.getNode());
 505                 if (l == null) {
 506                         LOG.debug("Session for node {} not found", input.getNode());
 507                         return Futures.immediateFuture(OPERATION_UNSENT);
 508                 }
 509
 510                 // Make sure there is no such LSP
 511                 final InstanceIdentifier<ReportedLsp> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 512                                 ReportedLsp.class, new ReportedLspKey(input.getName())).toInstance();
 513                 if (this.dataProvider.readOperationalData(lsp) != null) {
 514                         LOG.debug("Node {} already contains lsp {} at {}", input.getNode(), input.getName(), lsp);
 515                         return Futures.immediateFuture(OPERATION_UNSENT);
 516                 }
 517
 518                 // Build the request
 519                 final RequestsBuilder rb = new RequestsBuilder();
 520                 rb.fieldsFrom(input.getArguments());
 521                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 522                 rb.setLsp(new LspBuilder().setAdministrative(input.getArguments().isAdministrative()).setDelegate(Boolean.TRUE).setTlvs(
 523                                 new TlvsBuilder().setSymbolicPathName(new SymbolicPathNameBuilder().setPathName(input.getName()).build()).build()).build());
 524
 525                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(MESSAGE_HEADER);
 526                 ib.setRequests(ImmutableList.of(rb.build()));
 527
 528                 // Send the message
 529                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId(),
 530                                 input.getArguments().getMetadata());
 531         }
 532
 533         private static OperationResult createOperationResult(final FailureType type) {
 534                 return new OperationResult() {
 535                         @Override
 536                         public Class<? extends DataContainer> getImplementedInterface() {
 537                                 return OperationResult.class;
 538                         }
 539
 540                         @Override
 541                         public FailureType getFailure() {
 542                                 return type;
 543                         }
 544                 };
 545         }
 546
 547         private static String createNodeId(final InetAddress addr) {
 548                 return "pcc://" + addr.getHostAddress();
 549         }
 550
 551         synchronized ListenableFuture<OperationResult> realRemoveLsp(final RemoveLspArgs input) {
 552                 // Get the listener corresponding to the node
 553                 final SessionListener l = this.nodes.get(input.getNode());
 554                 if (l == null) {
 555                         LOG.debug("Session for node {} not found", input.getNode());
 556                         return Futures.immediateFuture(OPERATION_UNSENT);
 557                 }
 558
 559                 // Make sure the LSP exists, we need it for PLSP-ID
 560                 final InstanceIdentifier<ReportedLsp> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 561                                 ReportedLsp.class, new ReportedLspKey(input.getName())).toInstance();
 562                 final ReportedLsp rep = (ReportedLsp) this.dataProvider.readOperationalData(lsp);
 563                 if (rep == null) {
 564                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 565                         return Futures.immediateFuture(OPERATION_UNSENT);
 566                 }
 567
 568                 // Build the request and send it
 569                 final RequestsBuilder rb = new RequestsBuilder();
 570                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 571                 rb.setLsp(new LspBuilder().setRemove(Boolean.TRUE).setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 572
 573                 final PcinitiateMessageBuilder ib = new PcinitiateMessageBuilder(MESSAGE_HEADER);
 574                 ib.setRequests(ImmutableList.of(rb.build()));
 575                 return l.sendMessage(new PcinitiateBuilder().setPcinitiateMessage(ib.build()).build(), rb.getSrp().getOperationId(), null);
 576         }
 577
 578         synchronized ListenableFuture<OperationResult> realUpdateLsp(final UpdateLspArgs input) {
 579                 // Get the listener corresponding to the node
 580                 final SessionListener l = this.nodes.get(input.getNode());
 581                 if (l == null) {
 582                         LOG.debug("Session for node {} not found", input.getNode());
 583                         return Futures.immediateFuture(OPERATION_UNSENT);
 584                 }
 585
 586                 // Make sure the LSP exists
 587                 final InstanceIdentifier<ReportedLsp> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 588                                 ReportedLsp.class, new ReportedLspKey(input.getName())).toInstance();
 589                 final ReportedLsp rep = (ReportedLsp) this.dataProvider.readOperationalData(lsp);
 590                 if (rep == null) {
 591                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 592                         return Futures.immediateFuture(OPERATION_UNSENT);
 593                 }
 594
 595                 // Build the PCUpd request and send it
 596                 final UpdatesBuilder rb = new UpdatesBuilder();
 597                 rb.setSrp(new SrpBuilder().setOperationId(l.nextRequest()).setProcessingRule(Boolean.TRUE).build());
 598                 rb.setLsp(new LspBuilder().setPlspId(rep.getLsp().getPlspId()).setDelegate(Boolean.TRUE).build());
 599                 final PathBuilder pb = new PathBuilder();
 600                 rb.setPath(pb.setEro(input.getArguments().getEro()).build());
 601
 602                 final PcupdMessageBuilder ub = new PcupdMessageBuilder(MESSAGE_HEADER);
 603                 ub.setUpdates(ImmutableList.of(rb.build()));
 604                 return l.sendMessage(new PcupdBuilder().setPcupdMessage(ub.build()).build(), rb.getSrp().getOperationId(),
 605                                 input.getArguments().getMetadata());
 606         }
 607
 608         synchronized ListenableFuture<OperationResult> realEnsureLspOperational(final EnsureLspOperationalInput input) {
 609                 // Get the listener corresponding to the node
 610                 final SessionListener l = this.nodes.get(input.getNode());
 611                 if (l == null) {
 612                         LOG.debug("Session for node {} not found", input.getNode());
 613                         return Futures.immediateFuture(OPERATION_UNSENT);
 614                 }
 615
 616                 // Make sure the LSP exists
 617                 final InstanceIdentifier<ReportedLsp> lsp = InstanceIdentifier.builder(l.topologyAugment).child(PathComputationClient.class).child(
 618                                 ReportedLsp.class, new ReportedLspKey(input.getName())).toInstance();
 619                 LOG.debug("Checking if LSP {} has operational state {}", lsp, input.getArguments().getOperational());
 620                 final ReportedLsp rep = (ReportedLsp) this.dataProvider.readOperationalData(lsp);
 621                 if (rep == null) {
 622                         LOG.debug("Node {} does not contain LSP {}", input.getNode(), input.getName());
 623                         return Futures.immediateFuture(OPERATION_UNSENT);
 624                 }
 625
 626                 if (rep.getLsp().getOperational().equals(input.getArguments().getOperational())) {
 627                         return Futures.immediateFuture(OPERATION_SUCCESS);
 628                 } else {
 629                         return Futures.immediateFuture(OPERATION_UNSENT);
 630                 }
 631         }
 632
 633         @Override
 634         public void close() throws InterruptedException, ExecutionException {
 635                 final DataModificationTransaction t = this.dataProvider.beginTransaction();
 636                 t.removeOperationalData(this.topology);
 637                 t.commit().get();
 638         }
 639 }