X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=pcep%2Ftopology%2Ftopology-provider%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fbgpcep%2Fpcep%2Ftopology%2Fprovider%2FAbstractTopologySessionListener.java;h=ae3bb07336eef92f1268d0878532663d081e4c78;hb=470b07fdc354e2e1ed69fb098fd0640849a9d03f;hp=72471ac4ea5d10d91ba0f20b8e3933ebe824691e;hpb=3860d202e14ae693d515cf40f4795fd086a5d75d;p=bgpcep.git diff --git a/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java b/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java index 72471ac4ea..ae3bb07336 100644 --- a/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java +++ b/pcep/topology/topology-provider/src/main/java/org/opendaylight/bgpcep/pcep/topology/provider/AbstractTopologySessionListener.java @@ -9,8 +9,8 @@ package org.opendaylight.bgpcep.pcep.topology.provider; import static java.util.Objects.requireNonNull; -import com.google.common.collect.ImmutableList; import com.google.common.collect.Iterables; +import com.google.common.collect.Maps; import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; @@ -29,11 +29,13 @@ import java.util.Objects; import java.util.Optional; import java.util.Timer; import java.util.TimerTask; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.stream.Stream; import org.checkerframework.checker.lock.qual.GuardedBy; import org.checkerframework.checker.lock.qual.Holding; +import org.eclipse.jdt.annotation.NonNull; +import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.SessionStateImpl; import org.opendaylight.bgpcep.pcep.topology.provider.session.stats.TopologySessionStats; import org.opendaylight.mdsal.binding.api.WriteTransaction; @@ -43,14 +45,19 @@ import org.opendaylight.protocol.pcep.PCEPCloseTermination; import org.opendaylight.protocol.pcep.PCEPSession; import org.opendaylight.protocol.pcep.PCEPTerminationReason; import org.opendaylight.protocol.pcep.TerminationReason; -import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddressNoZoneBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev181109.LspObject; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev181109.Path1; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev181109.lsp.object.Lsp; +import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IetfInetUtil; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.initiated.rev200720.Stateful1; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.LspObject; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.Path1; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.PlspId; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.SrpIdNumber; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.StatefulTlv1Builder; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.Tlvs1; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.lsp.object.Lsp; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.ietf.stateful.rev200720.stateful.capability.tlv.Stateful; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.Message; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.MessageHeader; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.Object; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.ProtocolVersion; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev181109.open.object.open.Tlvs; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.LspId; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.Node1Builder; @@ -63,13 +70,14 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLsp; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.ReportedLspKey; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.StatefulTlvBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.Path; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.path.computation.client.reported.lsp.PathKey; import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.network.topology.topology.Node; import org.opendaylight.yangtools.yang.binding.DataObject; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.opendaylight.yangtools.yang.common.RpcResult; import org.opendaylight.yangtools.yang.common.RpcResultBuilder; -import org.opendaylight.yangtools.yang.common.Uint8; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,34 +85,23 @@ import org.slf4j.LoggerFactory; * Base class for PCEP topology providers. It handles the common tasks involved in managing a PCEP server (PCE) * endpoint, and exposing a network topology based on it. It needs to be subclassed to form a fully functional block, * where the subclass provides handling of incoming messages. - * - * @param identifier type of requests - * @param identifier type for LSPs */ -public abstract class AbstractTopologySessionListener implements TopologySessionListener, TopologySessionStats { - static final MessageHeader MESSAGE_HEADER = new MessageHeader() { - private final ProtocolVersion version = new ProtocolVersion(Uint8.ONE); +public abstract class AbstractTopologySessionListener implements TopologySessionListener, TopologySessionStats { + private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class); - @Override - public Class implementedInterface() { - return MessageHeader.class; - } + private final AtomicBoolean statefulCapability = new AtomicBoolean(false); + private final AtomicBoolean lspUpdateCapability = new AtomicBoolean(false); + private final AtomicBoolean initiationCapability = new AtomicBoolean(false); - @Override - public ProtocolVersion getVersion() { - return this.version; - } - }; - static final String MISSING_XML_TAG = "Mandatory XML tags are missing."; - private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class); @GuardedBy("this") - final Map lsps = new HashMap<>(); + final Map lsps = new HashMap<>(); @GuardedBy("this") - final SessionStateImpl listenerState; + SessionStateImpl listenerState; + @GuardedBy("this") - private final Map requests = new HashMap<>(); + private final Map requests = new HashMap<>(); @GuardedBy("this") - private final Map lspData = new HashMap<>(); + private final Map lspData = new ConcurrentHashMap<>(); private final ServerSessionManager serverSessionManager; private InstanceIdentifier pccIdentifier; @GuardedBy("this") @@ -119,167 +116,218 @@ public abstract class AbstractTopologySessionListener implements TopologyS AbstractTopologySessionListener(final ServerSessionManager serverSessionManager) { this.serverSessionManager = requireNonNull(serverSessionManager); - this.listenerState = new SessionStateImpl(this); } @Override - public final synchronized void onSessionUp(final PCEPSession psession) { - /* - * The session went up. Look up the router in Inventory model, - * create it if it is not there (marking that fact for later - * deletion), and mark it as synchronizing. Also create it in - * the topology model, with empty LSP list. - */ - final InetAddress peerAddress = psession.getRemoteAddress(); - - this.syncOptimization = new SyncOptimization(psession); - - final TopologyNodeState state = this.serverSessionManager.takeNodeState(peerAddress, - this, isLspDbRetreived()); - - // takeNodeState(..) may fail when the server session manager is being restarted due to configuration change - if (state == null) { - LOG.error("Unable to fetch topology node state for PCEP session. Closing session {}", psession); - psession.close(TerminationReason.UNKNOWN); - this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); - return; - } + public final void onSessionUp(final PCEPSession psession) { + synchronized (serverSessionManager) { + synchronized (this) { + /* + * The session went up. Look up the router in Inventory model, create it if it + * is not there (marking that fact for later deletion), and mark it as + * synchronizing. Also create it in the topology model, with empty LSP list. + */ + final InetAddress peerAddress = psession.getRemoteAddress(); + + syncOptimization = new SyncOptimization(psession); + final boolean haveLspDbVersion = syncOptimization.isDbVersionPresent(); + + final TopologyNodeState state = + serverSessionManager.takeNodeState(peerAddress, this, haveLspDbVersion); + + // takeNodeState(..) may fail when the server session manager is being restarted + // due to configuration change + if (state == null) { + LOG.error("Unable to fetch topology node state for PCEP session. Closing session {}", psession); + psession.close(TerminationReason.UNKNOWN); + onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); + return; + } - if (this.session != null || this.nodeState != null) { - LOG.error("PCEP session is already up with {}. Closing session {}", psession.getRemoteAddress(), psession); - psession.close(TerminationReason.UNKNOWN); - this.onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); - return; + if (session != null || nodeState != null) { + LOG.error("PCEP session is already up with {}. Closing session {}", peerAddress, psession); + psession.close(TerminationReason.UNKNOWN); + onSessionTerminated(psession, new PCEPCloseTermination(TerminationReason.UNKNOWN)); + return; + } + session = psession; + nodeState = state; + + LOG.trace("Peer {} resolved to topology node {}", peerAddress, state.getNodeId()); + + // Our augmentation in the topology node + final PathComputationClientBuilder pccBuilder = new PathComputationClientBuilder() + .setIpAddress(IetfInetUtil.INSTANCE.ipAddressNoZoneFor(peerAddress)); + + // Let subclass fill the details + updateStatefulCapabilities(pccBuilder, peerAddress, psession.getRemoteTlvs()); + + synced.set(isSynchronized()); + + final InstanceIdentifier topologyAugment = state.getNodeId().augmentation(Node1.class); + pccIdentifier = topologyAugment.child(PathComputationClient.class); + + if (haveLspDbVersion) { + final Node initialNodeState = state.getInitialNodeState(); + if (initialNodeState != null) { + loadLspData(initialNodeState, lspData, lsps, isIncrementalSynchro()); + pccBuilder.setReportedLsp( + initialNodeState.augmentation(Node1.class).getPathComputationClient().getReportedLsp()); + } + } + state.storeNode(topologyAugment, + new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), psession); + + listenerState = new SessionStateImpl(this, psession); + serverSessionManager.bind(state.getNodeId(), listenerState); + LOG.info("Session with {} attached to topology node {}", peerAddress, state.getNodeId()); + } } - this.session = psession; - this.nodeState = state; - - LOG.trace("Peer {} resolved to topology node {}", peerAddress, state.getNodeId()); - - // Our augmentation in the topology node - final PathComputationClientBuilder pccBuilder = new PathComputationClientBuilder(); - - onSessionUp(psession, pccBuilder); - this.synced.set(isSynchronized()); - - pccBuilder.setIpAddress(IpAddressNoZoneBuilder.getDefaultInstance(peerAddress.getHostAddress())); - final InstanceIdentifier topologyAugment = state.getNodeId().augmentation(Node1.class); - this.pccIdentifier = topologyAugment.child(PathComputationClient.class); - final Node initialNodeState = state.getInitialNodeState(); - final boolean isNodePresent = isLspDbRetreived() && initialNodeState != null; - if (isNodePresent) { - loadLspData(initialNodeState, this.lspData, this.lsps, isIncrementalSynchro()); - pccBuilder.setReportedLsp(initialNodeState.augmentation(Node1.class) - .getPathComputationClient().getReportedLsp()); + } + + @Holding("this") + private void updateStatefulCapabilities(final PathComputationClientBuilder pccBuilder, + final InetAddress peerAddress, final @Nullable Tlvs remoteTlvs) { + if (remoteTlvs != null) { + final Tlvs1 statefulTlvs = remoteTlvs.augmentation(Tlvs1.class); + if (statefulTlvs != null) { + final Stateful stateful = statefulTlvs.getStateful(); + if (stateful != null) { + statefulCapability.set(true); + final var updateCap = stateful.getLspUpdateCapability(); + if (updateCap != null) { + lspUpdateCapability.set(updateCap); + } + final Stateful1 stateful1 = stateful.augmentation(Stateful1.class); + if (stateful1 != null) { + final var initiation = stateful1.getInitiation(); + if (initiation != null) { + initiationCapability.set(initiation); + } + } + + pccBuilder.setReportedLsp(Map.of()); + if (isSynchronized()) { + pccBuilder.setStateSync(PccSyncState.Synchronized); + } else if (isTriggeredInitialSynchro()) { + pccBuilder.setStateSync(PccSyncState.TriggeredInitialSync); + } else if (isIncrementalSynchro()) { + pccBuilder.setStateSync(PccSyncState.IncrementalSync); + } else { + pccBuilder.setStateSync(PccSyncState.InitialResync); + } + pccBuilder.setStatefulTlv(new StatefulTlvBuilder() + .addAugmentation(new StatefulTlv1Builder(statefulTlvs).build()) + .build()); + return; + } + } } - state.storeNode(topologyAugment, - new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), this.session); - this.listenerState.init(psession); - this.serverSessionManager.bind(this.nodeState.getNodeId(), this.listenerState); - LOG.info("Session with {} attached to topology node {}", psession.getRemoteAddress(), state.getNodeId()); + LOG.debug("Peer {} does not advertise stateful TLV", peerAddress); } synchronized void updatePccState(final PccSyncState pccSyncState) { - if (this.nodeState == null) { + if (nodeState == null) { LOG.info("Server Session Manager is closed."); - AbstractTopologySessionListener.this.session.close(TerminationReason.UNKNOWN); + session.close(TerminationReason.UNKNOWN); return; } - final MessageContext ctx = new MessageContext(this.nodeState.getChain().newWriteOnlyTransaction()); + final MessageContext ctx = new MessageContext(nodeState.getChain().newWriteOnlyTransaction()); updatePccNode(ctx, new PathComputationClientBuilder().setStateSync(pccSyncState).build()); if (pccSyncState != PccSyncState.Synchronized) { - this.synced.set(false); - this.triggeredResyncInProcess = true; + synced.set(false); + triggeredResyncInProcess = true; } // All set, commit the modifications ctx.trans.commit().addCallback(new FutureCallback() { @Override public void onSuccess(final CommitInfo result) { - LOG.trace("Pcc Internal state for session {} updated successfully", - AbstractTopologySessionListener.this.session); + LOG.trace("Pcc Internal state for session {} updated successfully", session); } @Override - public void onFailure(final Throwable throwable) { - LOG.error("Failed to update Pcc internal state for session {}", - AbstractTopologySessionListener.this.session, throwable); - AbstractTopologySessionListener.this.session.close(TerminationReason.UNKNOWN); + public void onFailure(final Throwable cause) { + LOG.error("Failed to update Pcc internal state for session {}", session, cause); + session.close(TerminationReason.UNKNOWN); } }, MoreExecutors.directExecutor()); } synchronized boolean isTriggeredSyncInProcess() { - return this.triggeredResyncInProcess; + return triggeredResyncInProcess; } /** * Tear down the given PCEP session. It's OK to call this method even after the session * is already down. It always clear up the current session status. */ - @Holding("this") @SuppressWarnings("checkstyle:IllegalCatch") - private synchronized void tearDown(final PCEPSession psession) { - + private void tearDown(final PCEPSession psession) { requireNonNull(psession); - this.serverSessionManager.releaseNodeState(this.nodeState, psession, isLspDbPersisted()); - clearNodeState(); - - try { - if (this.session != null) { - this.session.close(); - } - psession.close(); - } catch (final Exception e) { - LOG.error("Session {} cannot be closed.", psession, e); - } - this.session = null; - this.syncOptimization = null; - - // Clear all requests we know about - for (final Entry e : this.requests.entrySet()) { - final PCEPRequest r = e.getValue(); - switch (r.getState()) { - case DONE: - // Done is done, nothing to do - LOG.trace("Request {} was done when session went down.", e.getKey()); - break; - case UNACKED: - // Peer has not acked: results in failure - LOG.info("Request {} was incomplete when session went down, failing the instruction", e.getKey()); - r.done(OperationResults.NOACK); - break; - case UNSENT: - // Peer has not been sent to the peer: results in cancellation - LOG.debug("Request {} was not sent when session went down, cancelling the instruction", e.getKey()); - r.done(OperationResults.UNSENT); - break; - default: - break; + synchronized (serverSessionManager) { + synchronized (this) { + serverSessionManager.releaseNodeState(nodeState, psession.getRemoteAddress(), isLspDbPersisted()); + clearNodeState(); + + try { + if (session != null) { + session.close(); + } + psession.close(); + } catch (final Exception e) { + LOG.error("Session {} cannot be closed.", psession, e); + } + session = null; + listenerState = null; + syncOptimization = null; + + // Clear all requests we know about + for (final Entry e : requests.entrySet()) { + // FIXME: exhaustive when we have JDK17+ + switch (e.getValue().cancel()) { + case DONE: + // Done is done, nothing to do + LOG.trace("Request {} was done when session went down.", e.getKey()); + break; + case UNACKED: + // Peer has not acked: results in failure + LOG.info("Request {} was incomplete when session went down, failing the instruction", + e.getKey()); + break; + case UNSENT: + // Peer has not been sent to the peer: results in cancellation + LOG.debug("Request {} was not sent when session went down, cancelling the instruction", + e.getKey()); + break; + default: + break; + } + } + requests.clear(); } } - this.requests.clear(); } @Override - public final synchronized void onSessionDown(final PCEPSession psession, final Exception exception) { + public final void onSessionDown(final PCEPSession psession, final Exception exception) { LOG.warn("Session {} went down unexpectedly", psession, exception); tearDown(psession); } @Override - public final synchronized void onSessionTerminated(final PCEPSession psession, final PCEPTerminationReason reason) { + public final void onSessionTerminated(final PCEPSession psession, final PCEPTerminationReason reason) { LOG.info("Session {} terminated by peer with reason {}", psession, reason); tearDown(psession); } @Override public final synchronized void onMessage(final PCEPSession psession, final Message message) { - if (this.nodeState == null) { + if (nodeState == null) { LOG.warn("Topology node state is null. Unhandled message {} on session {}", message, psession); psession.close(TerminationReason.UNKNOWN); return; } - final MessageContext ctx = new MessageContext(this.nodeState.getChain().newWriteOnlyTransaction()); + final MessageContext ctx = new MessageContext(nodeState.getChain().newWriteOnlyTransaction()); if (onMessage(ctx, message)) { LOG.warn("Unhandled message {} on session {}", message, psession); @@ -315,52 +363,67 @@ public abstract class AbstractTopologySessionListener implements TopologyS }, MoreExecutors.directExecutor()); } + /** + * Perform revision-specific message processing when a message arrives. + * + * @param ctx Message processing context + * @param message Protocol message + * @return True if the message type is not handle. + */ + protected abstract boolean onMessage(MessageContext ctx, Message message); + @Override - public synchronized void close() { - clearNodeState(); - if (this.session != null) { - LOG.info("Closing session {}", session); - this.session.close(TerminationReason.UNKNOWN); + public void close() { + synchronized (serverSessionManager) { + synchronized (this) { + clearNodeState(); + if (session != null) { + LOG.info("Closing session {}", session); + session.close(TerminationReason.UNKNOWN); + } + } } } - private synchronized void clearNodeState() { - if (this.nodeState != null) { - this.serverSessionManager.unbind(this.nodeState.getNodeId()); - this.nodeState = null; + @Holding({"this.serverSessionManager", "this"}) + private void clearNodeState() { + if (nodeState != null) { + serverSessionManager.unbind(nodeState.getNodeId()); + LOG.debug("Clear Node state: {}", nodeState.getNodeId()); + nodeState = null; } } - final synchronized PCEPRequest removeRequest(final S id) { - final PCEPRequest ret = this.requests.remove(id); - if (ret != null) { - this.listenerState.processRequestStats(ret.getElapsedMillis()); + final synchronized PCEPRequest removeRequest(final SrpIdNumber id) { + final PCEPRequest ret = requests.remove(id); + if (ret != null && listenerState != null) { + listenerState.processRequestStats(ret.getElapsedMillis()); } LOG.trace("Removed request {} object {}", id, ret); return ret; } - final synchronized ListenableFuture sendMessage(final Message message, final S requestId, + final synchronized ListenableFuture sendMessage(final Message message, final SrpIdNumber requestId, final Metadata metadata) { - final io.netty.util.concurrent.Future f = this.session.sendMessage(message); - this.listenerState.updateStatefulSentMsg(message); + final var sendFuture = session.sendMessage(message); + listenerState.updateStatefulSentMsg(message); final PCEPRequest req = new PCEPRequest(metadata); - this.requests.put(requestId, req); - final short rpcTimeout = this.serverSessionManager.getRpcTimeout(); + requests.put(requestId, req); + final short rpcTimeout = serverSessionManager.getRpcTimeout(); LOG.trace("RPC response timeout value is {} seconds", rpcTimeout); if (rpcTimeout > 0) { setupTimeoutHandler(requestId, req, rpcTimeout); } - f.addListener((FutureListener) future -> { + sendFuture.addListener((FutureListener) future -> { if (!future.isSuccess()) { synchronized (AbstractTopologySessionListener.this) { - AbstractTopologySessionListener.this.requests.remove(requestId); + requests.remove(requestId); } - req.done(OperationResults.UNSENT); + req.cancel(); LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause()); } else { - req.sent(); + req.markUnacked(); LOG.trace("Request {} sent to peer (object {})", requestId, req); } }); @@ -368,15 +431,15 @@ public abstract class AbstractTopologySessionListener implements TopologyS return req.getFuture(); } - private void setupTimeoutHandler(final S requestId, final PCEPRequest req, final short timeout) { + private void setupTimeoutHandler(final SrpIdNumber requestId, final PCEPRequest req, final short timeout) { final Timer timer = req.getTimer(); timer.schedule(new TimerTask() { @Override public void run() { synchronized (AbstractTopologySessionListener.this) { - AbstractTopologySessionListener.this.requests.remove(requestId); + requests.remove(requestId); } - req.done(); + req.cancel(); LOG.info("Request {} timed-out waiting for response", requestId); } }, TimeUnit.SECONDS.toMillis(timeout)); @@ -393,12 +456,12 @@ public abstract class AbstractTopologySessionListener implements TopologyS * @param solicited True if the update was solicited * @param remove True if this is an LSP path removal */ - protected final synchronized void updateLsp(final MessageContext ctx, final L id, final String lspName, + protected final synchronized void updateLsp(final MessageContext ctx, final PlspId id, final String lspName, final ReportedLspBuilder rlb, final boolean solicited, final boolean remove) { final String name; if (lspName == null) { - name = this.lsps.get(id); + name = lsps.get(id); if (name == null) { LOG.error("PLSPID {} seen for the first time, not reporting the LSP", id); return; @@ -408,13 +471,12 @@ public abstract class AbstractTopologySessionListener implements TopologyS } LOG.debug("Saved LSP {} with name {}", id, name); - this.lsps.put(id, name); + lsps.put(id, name); - - final ReportedLsp previous = this.lspData.get(name); + final ReportedLsp previous = lspData.get(name); // if no previous report about the lsp exist, just proceed if (previous != null) { - final List updatedPaths = makeBeforeBreak(rlb, previous, name, remove); + final Map updatedPaths = makeBeforeBreak(rlb, previous, name, remove); // if all paths or the last path were deleted, delete whole tunnel if (updatedPaths.isEmpty()) { LOG.debug("All paths were removed, removing LSP with {}.", id); @@ -428,24 +490,23 @@ public abstract class AbstractTopologySessionListener implements TopologyS // If this is an unsolicited update. We need to make sure we retain the metadata already present if (solicited) { - this.nodeState.setLspMetadata(name, rlb.getMetadata()); + nodeState.setLspMetadata(name, rlb.getMetadata()); } else { - rlb.setMetadata(this.nodeState.getLspMetadata(name)); + rlb.setMetadata(nodeState.getLspMetadata(name)); } final ReportedLsp rl = rlb.build(); - ctx.trans.put(LogicalDatastoreType.OPERATIONAL, this.pccIdentifier.child(ReportedLsp.class, rlb.key()), rl); + ctx.trans.put(LogicalDatastoreType.OPERATIONAL, pccIdentifier.child(ReportedLsp.class, rlb.key()), rl); LOG.debug("LSP {} updated to MD-SAL", name); - this.lspData.put(name, rl); + lspData.put(name, rl); } - private static List makeBeforeBreak(final ReportedLspBuilder rlb, final ReportedLsp previous, + private static Map makeBeforeBreak(final ReportedLspBuilder rlb, final ReportedLsp previous, final String name, final boolean remove) { // just one path should be reported final Path path = Iterables.getOnlyElement(rlb.getPath().values()); - final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.LspId reportedLspId = - path.getLspId(); + final var reportedLspId = path.getLspId(); final List updatedPaths; //lspId = 0 and remove = false -> tunnel is down, still exists but no path is signaled //remove existing tunnel's paths now, as explicit path remove will not come @@ -485,7 +546,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS } } LOG.debug("Setting new paths {} to lsp {}", updatedPaths, name); - return updatedPaths; + return Maps.uniqueIndex(updatedPaths, Path::key); } /** @@ -494,26 +555,25 @@ public abstract class AbstractTopologySessionListener implements TopologyS * @param ctx Message context */ protected final synchronized void stateSynchronizationAchieved(final MessageContext ctx) { - if (this.synced.getAndSet(true)) { + if (synced.getAndSet(true)) { LOG.debug("State synchronization achieved while synchronizing, not updating state"); return; } - if (this.triggeredResyncInProcess) { - this.triggeredResyncInProcess = false; - } + + triggeredResyncInProcess = false; updatePccNode(ctx, new PathComputationClientBuilder().setStateSync(PccSyncState.Synchronized).build()); // The node has completed synchronization, cleanup metadata no longer reported back - this.nodeState.cleanupExcept(this.lsps.values()); - LOG.debug("Session {} achieved synchronized state", this.session); + nodeState.cleanupExcept(lsps.values()); + LOG.debug("Session {} achieved synchronized state", session); } protected final synchronized void updatePccNode(final MessageContext ctx, final PathComputationClient pcc) { - ctx.trans.merge(LogicalDatastoreType.OPERATIONAL, this.pccIdentifier, pcc); + ctx.trans.merge(LogicalDatastoreType.OPERATIONAL, pccIdentifier, pcc); } - protected final InstanceIdentifier lspIdentifier(final String name) { - return this.pccIdentifier.child(ReportedLsp.class, new ReportedLspKey(name)); + protected final @NonNull InstanceIdentifier lspIdentifier(final String name) { + return pccIdentifier.child(ReportedLsp.class, new ReportedLspKey(name)); } /** @@ -522,29 +582,16 @@ public abstract class AbstractTopologySessionListener implements TopologyS * @param ctx Message Context * @param id Revision-specific LSP identifier */ - protected final synchronized void removeLsp(final MessageContext ctx, final L id) { - final String name = this.lsps.remove(id); + protected final synchronized void removeLsp(final MessageContext ctx, final PlspId id) { + final String name = lsps.remove(id); LOG.debug("LSP {} removed", name); ctx.trans.delete(LogicalDatastoreType.OPERATIONAL, lspIdentifier(name)); - this.lspData.remove(name); + lspData.remove(name); } - @SuppressWarnings("checkstyle:OverloadMethodsDeclarationOrder") - protected abstract void onSessionUp(PCEPSession session, PathComputationClientBuilder pccBuilder); - - /** - * Perform revision-specific message processing when a message arrives. - * - * @param ctx Message processing context - * @param message Protocol message - * @return True if the message type is not handle. - */ - @SuppressWarnings("checkstyle:OverloadMethodsDeclarationOrder") - protected abstract boolean onMessage(MessageContext ctx, Message message); - - final String lookupLspName(final L id) { - requireNonNull(id, "ID parameter null."); - return this.lsps.get(id); + @Holding("this") + final String lookupLspName(final PlspId id) { + return lsps.get(requireNonNull(id, "ID parameter null.")); } /** @@ -554,25 +601,18 @@ public abstract class AbstractTopologySessionListener implements TopologyS * @param id InstanceIdentifier of the node * @return null if the node does not exists, or operational data */ - final synchronized FluentFuture> - readOperationalData(final InstanceIdentifier id) { - if (this.nodeState == null) { - return null; - } - return this.nodeState.readOperationalData(id); + final synchronized FluentFuture> readOperationalData( + final InstanceIdentifier id) { + return nodeState == null ? null : nodeState.readOperationalData(id); } protected abstract Object validateReportedLsp(Optional rep, LspId input); - protected abstract void loadLspData(Node node, Map lspData, Map lsps, + protected abstract void loadLspData(Node node, Map lspData, Map lsps, boolean incrementalSynchro); final boolean isLspDbPersisted() { - return this.syncOptimization != null && this.syncOptimization.isSyncAvoidanceEnabled(); - } - - final boolean isLspDbRetreived() { - return this.syncOptimization != null && this.syncOptimization.isDbVersionPresent(); + return syncOptimization != null && syncOptimization.isSyncAvoidanceEnabled(); } /** @@ -580,44 +620,55 @@ public abstract class AbstractTopologySessionListener implements TopologyS * LSP-DB-VERSION TLV values doesnt match, and LSP-SYNC-CAPABILITY is enabled. */ final synchronized boolean isIncrementalSynchro() { - return this.syncOptimization != null && this.syncOptimization.isSyncAvoidanceEnabled() - && this.syncOptimization.isDeltaSyncEnabled(); + return syncOptimization != null && syncOptimization.isSyncAvoidanceEnabled() + && syncOptimization.isDeltaSyncEnabled(); } final synchronized boolean isTriggeredInitialSynchro() { - return this.syncOptimization != null && this.syncOptimization.isTriggeredInitSyncEnabled(); + return syncOptimization != null && syncOptimization.isTriggeredInitSyncEnabled(); } final synchronized boolean isTriggeredReSyncEnabled() { - return this.syncOptimization != null && this.syncOptimization.isTriggeredReSyncEnabled(); + return syncOptimization != null && syncOptimization.isTriggeredReSyncEnabled(); } protected final synchronized boolean isSynchronized() { - return this.syncOptimization != null && this.syncOptimization.doesLspDbMatch(); + return syncOptimization != null && syncOptimization.doesLspDbMatch(); } @Override - public int getDelegatedLspsCount() { - final Stream stream; - synchronized (this) { - stream = ImmutableList.copyOf(this.lspData.values()).stream(); - } - - return Math.toIntExact(stream + public final int getDelegatedLspsCount() { + return Math.toIntExact(lspData.values().stream() .map(ReportedLsp::getPath).filter(pathList -> pathList != null && !pathList.isEmpty()) // pick the first path, as delegate status should be same in each path .map(pathList -> pathList.values().iterator().next()) .map(path -> path.augmentation(Path1.class)).filter(Objects::nonNull) .map(LspObject::getLsp).filter(Objects::nonNull) - .filter(Lsp::isDelegate) + .filter(Lsp::getDelegate) .count()); } @Override - public boolean isSessionSynchronized() { - return this.synced.get(); + public final boolean isSessionSynchronized() { + return synced.get(); + } + + @Override + public final boolean isInitiationCapability() { + return initiationCapability.get(); + } + + @Override + public final boolean isStatefulCapability() { + return statefulCapability.get(); } + @Override + public final boolean isLspUpdateCapability() { + return lspUpdateCapability.get(); + } + + @Override public synchronized ListenableFuture> tearDownSession(final TearDownSessionInput input) { close(); @@ -633,14 +684,14 @@ public abstract class AbstractTopologySessionListener implements TopologyS } void resolveRequest(final PCEPRequest req) { - this.requests.add(req); + requests.add(req); } @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD", justification = "https://github.com/spotbugs/spotbugs/issues/811") private void notifyRequests() { - for (final PCEPRequest r : this.requests) { - r.done(OperationResults.SUCCESS); + for (final PCEPRequest r : requests) { + r.finish(OperationResults.SUCCESS); } } }