Finish the sentence

[netconf.git] / netconf / netconf-netty-util / src / main / java / org / opendaylight / netconf / nettyutil / AbstractNetconfSessionNegotiator.java

diff --git a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
index 04f4aa90b5b1c24f7ac7fe5157aa3bc6fe08c342..89c0be106837c21a907587dedca8dcb491b20a39 100644 (file)
--- a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
+++ b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
@@ -20,11 +20,12 @@ import io.netty.handler.ssl.SslHandler;
 import io.netty.util.Timeout;
 import io.netty.util.Timer;
 import io.netty.util.concurrent.Promise;
-import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import org.checkerframework.checker.index.qual.NonNegative;
 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.netconf.api.NetconfDocumentedException;
 import org.opendaylight.netconf.api.NetconfMessage;
 import org.opendaylight.netconf.api.NetconfSessionListener;
@@ -84,8 +85,8 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
     private final L sessionListener;
     private final Timer timer;
 
+    @GuardedBy("this")
     private Timeout timeoutTask;
-
     @GuardedBy("this")
     private State state = State.IDLE;
 
@@ -119,9 +120,9 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
         if (ifNegotiatedAlready()) {
             LOG.debug("Negotiation on channel {} already started", channel);
         } else {
-            final Optional<SslHandler> sslHandler = getSslHandler(channel);
-            if (sslHandler.isPresent()) {
-                sslHandler.get().handshakeFuture().addListener(future -> {
+            final var sslHandler = getSslHandler(channel);
+            if (sslHandler != null) {
+                sslHandler.handshakeFuture().addListener(future -> {
                     checkState(future.isSuccess(), "Ssl handshake was not successful");
                     LOG.debug("Ssl handshake complete");
                     start();
@@ -137,8 +138,8 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
         return this.state != State.IDLE;
     }
 
-    private static Optional<SslHandler> getSslHandler(final Channel channel) {
-        return Optional.ofNullable(channel.pipeline().get(SslHandler.class));
+    private static @Nullable SslHandler getSslHandler(final Channel channel) {
+        return channel.pipeline().get(SslHandler.class);
     }
 
     private void start() {
@@ -149,15 +150,26 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
         sendMessage(localHello);
 
         replaceHelloMessageOutboundHandler();
-        changeState(State.OPEN_WAIT);
 
-        timeoutTask = this.timer.newTimeout(this::timeoutExpired, connectionTimeoutMillis, TimeUnit.MILLISECONDS);
+        synchronized (this) {
+            lockedChangeState(State.OPEN_WAIT);
+
+            // Service the timeout on channel's eventloop, so that we do not get state transition problems
+            timeoutTask = timer.newTimeout(unused -> channel.eventLoop().execute(this::timeoutExpired),
+                connectionTimeoutMillis, TimeUnit.MILLISECONDS);
+        }
     }
 
-    private synchronized void timeoutExpired(final Timeout timeout) {
+    private synchronized void timeoutExpired() {
+        if (timeoutTask == null) {
+            // cancelTimeout() between expiry and execution on the loop
+            return;
+        }
+        timeoutTask = null;
+
         if (state != State.ESTABLISHED) {
-            LOG.debug("Connection timeout after {}, session backed by channel {} is in state {}", timeout, channel,
-                state);
+            LOG.debug("Connection timeout after {}ms, session backed by channel {} is in state {}",
+                connectionTimeoutMillis, channel, state);
 
             // Do not fail negotiation if promise is done or canceled
             // It would result in setting result of the promise second time and that throws exception
@@ -179,9 +191,10 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
         }
     }
 
-    private void cancelTimeout() {
-        if (timeoutTask != null) {
-            timeoutTask.cancel();
+    private synchronized void cancelTimeout() {
+        if (timeoutTask != null && !timeoutTask.cancel()) {
+            // Late-coming cancel: make sure the task does not actually run
+            timeoutTask = null;
         }
     }
 
@@ -253,6 +266,11 @@ public abstract class AbstractNetconfSessionNegotiator<S extends AbstractNetconf
     }
 
     private synchronized void changeState(final State newState) {
+        lockedChangeState(newState);
+    }
+
+    @Holding("this")
+    private void lockedChangeState(final State newState) {
         LOG.debug("Changing state from : {} to : {} for channel: {}", state, newState, channel);
         checkState(isStateChangePermitted(state, newState),
                 "Cannot change state from %s to %s for channel %s", state, newState, channel);