Bump versions by x.y.(z+1)

[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / AbstractBGPSessionNegotiator.java

diff --git a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/AbstractBGPSessionNegotiator.java b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/AbstractBGPSessionNegotiator.java
index 9c723e237a56fb7790ab936a10f8b5a5378f81af..cef181ab081f9f491a793992a493e7be62a0ee32 100644 (file)
--- a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/AbstractBGPSessionNegotiator.java
+++ b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/AbstractBGPSessionNegotiator.java
@@ -11,11 +11,11 @@ package org.opendaylight.protocol.bgp.rib.impl;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.util.concurrent.Promise;
+import io.netty.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import javax.annotation.concurrent.GuardedBy;
 import org.opendaylight.protocol.bgp.parser.BGPDocumentedException;
@@ -80,6 +80,8 @@ abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandlerAdapter
     private State state = State.IDLE;
     @GuardedBy("this")
     private BGPSessionImpl session;
+    @GuardedBy("this")
+    private ScheduledFuture<?> pending;
 
     AbstractBGPSessionNegotiator(final Promise<BGPSessionImpl> promise, final Channel channel,
             final BGPPeerRegistry registry) {
@@ -114,16 +116,14 @@ abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandlerAdapter
                 preferences.getBgpId()).setBgpParameters(preferences.getParams()).build());
             if (this.state != State.FINISHED) {
                 this.state = State.OPEN_SENT;
-                this.channel.eventLoop().schedule(new Runnable() {
-                    @Override
-                    public void run() {
-                        synchronized (AbstractBGPSessionNegotiator.this) {
-                            if (AbstractBGPSessionNegotiator.this.state != State.FINISHED) {
-                                AbstractBGPSessionNegotiator.this
-                                    .sendMessage(buildErrorNotify(BGPError.HOLD_TIMER_EXPIRED));
-                                negotiationFailed(new BGPDocumentedException("HoldTimer expired", BGPError.FSM_ERROR));
-                                AbstractBGPSessionNegotiator.this.state = State.FINISHED;
-                            }
+                this.pending = this.channel.eventLoop().schedule(() -> {
+                    synchronized (AbstractBGPSessionNegotiator.this) {
+                        AbstractBGPSessionNegotiator.this.pending = null;
+                        if (AbstractBGPSessionNegotiator.this.state != State.FINISHED) {
+                            AbstractBGPSessionNegotiator.this
+                                .sendMessage(buildErrorNotify(BGPError.HOLD_TIMER_EXPIRED));
+                            negotiationFailed(new BGPDocumentedException("HoldTimer expired", BGPError.FSM_ERROR));
+                            AbstractBGPSessionNegotiator.this.state = State.FINISHED;
                         }
                     }
                 }, INITIAL_HOLDTIMER, TimeUnit.MINUTES);
@@ -253,19 +253,21 @@ abstract class AbstractBGPSessionNegotiator extends ChannelInboundHandlerAdapter
     private void negotiationFailedCloseChannel(final Throwable cause) {
         LOG.debug("Negotiation on channel {} failed", this.channel, cause);
         this.channel.close();
-        this.promise.setFailure(cause);
+        synchronized (AbstractBGPSessionNegotiator.this) {
+            if (this.pending != null && this.pending.isCancellable()) {
+                this.pending.cancel(true);
+                this.pending = null;
+            }
+        }
     }
 
     private void sendMessage(final Notification msg) {
-        this.channel.writeAndFlush(msg).addListener(new ChannelFutureListener() {
-            @Override
-            public void operationComplete(final ChannelFuture f) {
-                if (!f.isSuccess()) {
-                    LOG.warn("Failed to send message {} to channel {}", msg,  AbstractBGPSessionNegotiator.this.channel, f.cause());
-                    negotiationFailedCloseChannel(f.cause());
-                } else {
-                    LOG.trace("Message {} sent to channel {}", msg, AbstractBGPSessionNegotiator.this.channel);
-                }
+        this.channel.writeAndFlush(msg).addListener((ChannelFutureListener) f -> {
+            if (!f.isSuccess()) {
+                LOG.warn("Failed to send message {} to channel {}", msg,  AbstractBGPSessionNegotiator.this.channel, f.cause());
+                negotiationFailedCloseChannel(f.cause());
+            } else {
+                LOG.trace("Message {} sent to channel {}", msg, AbstractBGPSessionNegotiator.this.channel);
             }
         });
     }