Do not use JdkFutureAdapters in BgpPeerRpc

ChannelFuture provides listeners, hence we can trivially bridge
results without having to listen in a threadpool.

Change-Id: Ia46c5f406b6be50b9eba293ecf80714db99dabf4
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
(cherry picked from commit 76ed78eb1ae79c0cbc67cd00b7ea3a087719304f)

diff --git a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpc.java b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpc.java
index e4249db2084e52da6b63cc29485ac6426438e955..f2bbc33f33438f761ef40ae677e81f46cbc2b9a4 100644 (file)
--- a/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpc.java
+++ b/bgp/rib-impl/src/main/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpc.java
@@ -11,7 +11,6 @@ import static java.util.Objects.requireNonNull;
 
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.JdkFutureAdapters;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
@@ -89,17 +88,19 @@ public class BgpPeerRpc implements BgpPeerRpcService {
     public ListenableFuture<RpcResult<RouteRefreshRequestOutput>> routeRefreshRequest(
             final RouteRefreshRequestInput input) {
         final ChannelFuture f = sendRRMessage(input);
-        if (f != null) {
-            return Futures.transform(JdkFutureAdapters.listenInPoolThread(f), input1 -> {
-                if (f.isSuccess()) {
-                    return RpcResultBuilder.success(new RouteRefreshRequestOutputBuilder().build()).build();
-                }
-                return RpcResultBuilder.<RouteRefreshRequestOutput>failed().withError(ErrorType.RPC, FAILURE_MSG)
-                        .build();
-            }, MoreExecutors.directExecutor());
+        if (f == null) {
+            return RpcResultBuilder.<RouteRefreshRequestOutput>failed().withError(ErrorType.RPC,
+                FAILURE_MSG + " due to unsupported address families.").buildFuture();
         }
-        return RpcResultBuilder.<RouteRefreshRequestOutput>failed().withError(ErrorType.RPC,
-            FAILURE_MSG + " due to unsupported address families.").buildFuture();
+
+        final SettableFuture<RpcResult<RouteRefreshRequestOutput>> ret = SettableFuture.create();
+        f.addListener(future -> {
+            ret.set(future.isSuccess()
+                ? RpcResultBuilder.success(new RouteRefreshRequestOutputBuilder().build()).build()
+                        : RpcResultBuilder.<RouteRefreshRequestOutput>failed().withError(ErrorType.RPC, FAILURE_MSG)
+                        .build());
+        });
+        return ret;
     }
 
     private ChannelFuture sendRRMessage(final RouteRefreshRequestInput input) {

diff --git a/bgp/rib-impl/src/test/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpcTest.java b/bgp/rib-impl/src/test/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpcTest.java
index 3180b7ff144a9b6e6b278abf27edcb6fd30841c8..f022a4889b4862fc9132de322c6d28631a4eb230 100644 (file)
--- a/bgp/rib-impl/src/test/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpcTest.java
+++ b/bgp/rib-impl/src/test/java/org/opendaylight/protocol/bgp/rib/impl/BgpPeerRpcTest.java
@@ -10,18 +10,21 @@ package org.opendaylight.protocol.bgp.rib.impl;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
 
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
 import java.util.Collections;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.MockitoAnnotations;
+import org.mockito.junit.MockitoJUnitRunner;
 import org.opendaylight.protocol.bgp.rib.spi.PeerRPCs;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.PeerRef;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.peer.rpc.rev180329.ResetSessionInput;
@@ -40,6 +43,7 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.type
 import org.opendaylight.yangtools.yang.binding.Notification;
 import org.opendaylight.yangtools.yang.common.RpcResult;
 
+@RunWith(MockitoJUnitRunner.StrictStubs.class)
 public final class BgpPeerRpcTest {
     @Mock
     private BGPSessionImpl session;
@@ -53,16 +57,19 @@ public final class BgpPeerRpcTest {
 
     @Before
     public void setUp() throws InterruptedException, ExecutionException {
-        MockitoAnnotations.initMocks(this);
         this.rpc = new BgpPeerRpc(this.peerRpcs, this.session,
                 Collections.singleton(new TablesKey(Ipv4AddressFamily.class, SubsequentAddressFamily.class)));
         final ChannelOutputLimiter limiter = new ChannelOutputLimiter(this.session);
 
-        Mockito.doReturn(limiter).when(this.session).getLimiter();
-        Mockito.doReturn(this.future).when(this.session).writeAndFlush(any(Notification.class));
-        Mockito.doReturn(true).when(this.future).isDone();
-        Mockito.doReturn(null).when(this.future).get();
-        Mockito.doReturn(true).when(this.future).isSuccess();
+        doReturn(limiter).when(this.session).getLimiter();
+        doReturn(this.future).when(this.session).writeAndFlush(any(Notification.class));
+
+        doReturn(true).when(this.future).isSuccess();
+        doAnswer(invocation -> {
+            GenericFutureListener<ChannelFuture> listener = invocation.getArgument(0);
+            listener.operationComplete(this.future);
+            return null;
+        }).when(this.future).addListener(any());
     }
 
     @Test
@@ -89,7 +96,7 @@ public final class BgpPeerRpcTest {
 
     @Test
     public void testResetSessionRequestSuccessRequest() throws InterruptedException, ExecutionException {
-        Mockito.doReturn(Futures.immediateFuture(null)).when(this.peerRpcs).releaseConnection();
+        doReturn(Futures.immediateFuture(null)).when(this.peerRpcs).releaseConnection();
         final ResetSessionInput input = new ResetSessionInputBuilder()
                 .setPeerRef(this.peer).build();
         final Future<RpcResult<ResetSessionOutput>> result = this.rpc.resetSession(input);
@@ -99,7 +106,7 @@ public final class BgpPeerRpcTest {
     @Test
     public void testRestartGracefullyRequestFailedRequest() throws ExecutionException, InterruptedException {
         final long referraltimerSeconds = 10L;
-        Mockito.doReturn(new SimpleSessionListener().restartGracefully(referraltimerSeconds))
+        doReturn(new SimpleSessionListener().restartGracefully(referraltimerSeconds))
                 .when(this.peerRpcs).restartGracefully(referraltimerSeconds);
         final RestartGracefullyInput input = new RestartGracefullyInputBuilder()
                 .setSelectionDeferralTime(referraltimerSeconds)