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=df1be1c1bc96ed82c9d260acb33611639860bc87;hb=b85eb1ab9921837175ac1923c1a8a808947e6f03;hp=71cfb6e76ceb4e28a2b4081872279f6ae98bbb31;hpb=2db6622d02d1e19621fd35342da64c0a6663cdc3;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 old mode 100755 new mode 100644 index 71cfb6e76c..df1be1c1bc --- 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,14 +9,13 @@ package org.opendaylight.bgpcep.pcep.topology.provider; import static java.util.Objects.requireNonNull; -import com.google.common.base.Preconditions; -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; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; -import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import io.netty.util.concurrent.FutureListener; import java.net.InetAddress; import java.util.ArrayList; @@ -29,11 +28,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,32 +44,35 @@ 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.IpAddressBuilder; -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.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.lsp.object.Lsp; 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.topology.pcep.rev181109.LspId; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.Node1; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.Node1Builder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.OperationResult; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.PccSyncState; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.TearDownSessionInput; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.lsp.metadata.Metadata; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.PathComputationClient; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.PathComputationClientBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.path.computation.client.ReportedLsp; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.path.computation.client.ReportedLspBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.path.computation.client.ReportedLspKey; -import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev181109.pcep.client.attributes.path.computation.client.reported.lsp.Path; +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; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.OperationResult; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.PccSyncState; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.TearDownSessionInput; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.lsp.metadata.Metadata; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.PathComputationClient; +import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.topology.pcep.rev200120.pcep.client.attributes.PathComputationClientBuilder; +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.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; @@ -81,8 +85,11 @@ import org.slf4j.LoggerFactory; * @param identifier type for LSPs */ public abstract class AbstractTopologySessionListener implements TopologySessionListener, TopologySessionStats { + private static final Logger LOG = LoggerFactory.getLogger(AbstractTopologySessionListener.class); + + static final String MISSING_XML_TAG = "Mandatory XML tags are missing."; static final MessageHeader MESSAGE_HEADER = new MessageHeader() { - private final ProtocolVersion version = new ProtocolVersion((short) 1); + private final ProtocolVersion version = new ProtocolVersion(Uint8.ONE); @Override public Class implementedInterface() { @@ -91,19 +98,19 @@ public abstract class AbstractTopologySessionListener implements TopologyS @Override public ProtocolVersion getVersion() { - return this.version; + return 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<>(); @GuardedBy("this") - final SessionStateImpl listenerState; + SessionStateImpl listenerState; + @GuardedBy("this") 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") @@ -118,167 +125,182 @@ 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; - } - this.session = psession; - this.nodeState = state; - this.serverSessionManager.bind(this.nodeState.getNodeId(), this.listenerState); - - 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(IpAddressBuilder.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()); + 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 + onSessionUp(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()); + } } - state.storeNode(topologyAugment, - new Node1Builder().setPathComputationClient(pccBuilder.build()).build(), this.session); - this.listenerState.init(psession); - LOG.info("Session with {} attached to topology node {}", psession.getRemoteAddress(), state.getNodeId()); } + protected abstract void onSessionUp(PathComputationClientBuilder pccBuilder, InetAddress peerAddress, + @Nullable Tlvs remoteTlvs); + 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()) { + 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; + } + } + 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); @@ -314,26 +336,41 @@ 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 PCEPRequest ret = requests.remove(id); + if (ret != null && listenerState != null) { + listenerState.processRequestStats(ret.getElapsedMillis()); } LOG.trace("Removed request {} object {}", id, ret); return ret; @@ -341,20 +378,20 @@ public abstract class AbstractTopologySessionListener implements TopologyS final synchronized ListenableFuture sendMessage(final Message message, final S 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); LOG.info("Failed to send request {}, instruction cancelled", requestId, future.cause()); @@ -373,7 +410,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS @Override public void run() { synchronized (AbstractTopologySessionListener.this) { - AbstractTopologySessionListener.this.requests.remove(requestId); + requests.remove(requestId); } req.done(); LOG.info("Request {} timed-out waiting for response", requestId); @@ -397,7 +434,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS 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; @@ -407,13 +444,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); @@ -427,24 +463,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 - Preconditions.checkState(rlb.getPath().size() == 1); - final org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.rsvp.rev150820.LspId reportedLspId = - rlb.getPath().get(0).getLspId(); + final Path path = Iterables.getOnlyElement(rlb.getPath().values()); + 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 @@ -453,14 +488,15 @@ public abstract class AbstractTopologySessionListener implements TopologyS LOG.debug("Remove previous paths {} to this lsp name {}", previous.getPath(), name); } else { // check previous report for existing paths - updatedPaths = new ArrayList<>(previous.getPath()); + final Collection prev = previous.nonnullPath().values(); + updatedPaths = new ArrayList<>(prev); LOG.debug("Found previous paths {} to this lsp name {}", updatedPaths, name); - for (final Path path : previous.getPath()) { + for (final Path prevPath : prev) { //we found reported path in previous reports - if (path.getLspId().getValue().toJava() == 0 || path.getLspId().equals(reportedLspId)) { - LOG.debug("Match on lsp-id {}", path.getLspId().getValue()); + if (prevPath.getLspId().getValue().toJava() == 0 || prevPath.getLspId().equals(reportedLspId)) { + LOG.debug("Match on lsp-id {}", prevPath.getLspId().getValue()); // path that was reported previously and does have the same lsp-id, path will be updated - final boolean r = updatedPaths.remove(path); + final boolean r = updatedPaths.remove(prevPath); LOG.trace("Request removed? {}", r); } } @@ -468,8 +504,8 @@ public abstract class AbstractTopologySessionListener implements TopologyS // if the path does not exist in previous report, add it to path list, it's a new ERO // only one path will be added //lspId is 0 means confirmation message that shouldn't be added (because we have no means of deleting it later) - LOG.trace("Adding new path {} to {}", rlb.getPath(), updatedPaths); - updatedPaths.addAll(rlb.getPath()); + LOG.trace("Adding new path {} to {}", path, updatedPaths); + updatedPaths.add(path); if (remove) { if (reportedLspId.getValue().toJava() == 0) { // if lsp-id also 0, remove all paths @@ -477,13 +513,13 @@ public abstract class AbstractTopologySessionListener implements TopologyS updatedPaths.clear(); } else { // path is marked to be removed - LOG.debug("Removing path {} from {}", rlb.getPath(), updatedPaths); - final boolean r = updatedPaths.removeAll(rlb.getPath()); + LOG.debug("Removing path {} from {}", path, updatedPaths); + final boolean r = updatedPaths.remove(path); LOG.trace("Request removed? {}", r); } } LOG.debug("Setting new paths {} to lsp {}", updatedPaths, name); - return updatedPaths; + return Maps.uniqueIndex(updatedPaths, Path::key); } /** @@ -492,26 +528,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)); } /** @@ -521,28 +556,15 @@ public abstract class AbstractTopologySessionListener implements TopologyS * @param id Revision-specific LSP identifier */ protected final synchronized void removeLsp(final MessageContext ctx, final L id) { - final String name = this.lsps.remove(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); - + @Holding("this") final String lookupLspName(final L id) { - requireNonNull(id, "ID parameter null."); - return this.lsps.get(id); + return lsps.get(requireNonNull(id, "ID parameter null.")); } /** @@ -552,12 +574,9 @@ 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); @@ -566,11 +585,7 @@ public abstract class AbstractTopologySessionListener implements TopologyS 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(); } /** @@ -578,42 +593,37 @@ 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 + 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.get(0)) + .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(); + return synced.get(); } @Override @@ -631,13 +641,11 @@ 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) { + for (final PCEPRequest r : requests) { r.done(OperationResults.SUCCESS); } }