Merge pull request #614 from FortnoxAB/patch-5

Closing connection instead of just creating observable that is not subscribed to
ReactiveX · Feb 13, 2019 · 878d6ef · 878d6ef
2 parents ca6d0e1 + d67b6b0
commit 878d6ef
Show file tree

Hide file tree

Showing 4 changed files with 117 additions and 14 deletions.
diff --git a/rxnetty-common/src/main/java/io/reactivex/netty/channel/Connection.java b/rxnetty-common/src/main/java/io/reactivex/netty/channel/Connection.java
@@ -317,7 +317,7 @@ protected void connectCloseToChannelClose() {
                     .addListener(new ChannelFutureListener() {
                         @Override
                         public void operationComplete(ChannelFuture future) throws Exception {
-                            close(false); // Close this connection when the channel is closed.
+                            closeNow(); // Close this connection when the channel is closed.
                         }
                     });
         nettyChannel.attr(CONNECTION_ATTRIBUTE_KEY).set(this);

diff --git a/...common/src/test/java/io/reactivex/netty/client/pool/PooledConnectionProviderImplTest.java b/...common/src/test/java/io/reactivex/netty/client/pool/PooledConnectionProviderImplTest.java
@@ -213,7 +213,7 @@ public void testMetricEventCallback() throws Throwable {
         assertThat("Unexpected release attempted count.", eventsListener.getReleaseAttemptedCount(), is(2L));
         assertThat("Unexpected release succeeded count.", eventsListener.getReleaseSucceededCount(), is(2L));
         assertThat("Unexpected release failed count.", eventsListener.getReleaseFailedCount(), is(0L));
-        assertThat("Unexpected connection eviction count.", eventsListener.getEvictionCount(), is(1L));
+        assertThat("Unexpected connection eviction count.", eventsListener.getEvictionCount(), is(2L));
     }
 
     private PooledConnection<String, String> _testRelease() throws Exception {
@@ -379,4 +379,4 @@ public void call(Subscriber<? super Connection<String, String>> s) {
     public @interface MaxConnections {
         int value() default DEFAULT_MAX_CONNECTIONS;
     }
-}
+}
diff --git a/...-tcp/src/test/java/io/reactivex/netty/protocol/tcp/client/PoolingWithRealChannelTest.java b/...-tcp/src/test/java/io/reactivex/netty/protocol/tcp/client/PoolingWithRealChannelTest.java
@@ -23,6 +23,11 @@
 import org.junit.Rule;
 import org.junit.Test;
 import rx.Observable;
+import rx.functions.Func0;
+import rx.functions.Func1;
+import rx.observers.AssertableSubscriber;
+import org.junit.Test;
+import rx.Observable;
 import rx.functions.Action1;
 import rx.functions.Func0;
 import rx.functions.Func1;
@@ -31,6 +36,14 @@
 import java.util.ArrayList;
 import java.util.List;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static rx.Observable.fromCallable;
+import static rx.Observable.just;
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
 import static rx.Observable.fromCallable;
@@ -134,4 +147,83 @@ public void call(Throwable throwable) {
             test.assertNoErrors();
         }
     }
+
+    @Test
+    /**
+     *
+     * Load test to prove concurrency issues mainly seen on heavy load.
+     *
+     */
+    public void assertPermitsAreReleasedWhenMergingObservablesWithExceptions() {
+        clientRule.startServer(10, true);
+
+        MockTcpClientEventListener listener = new MockTcpClientEventListener();
+        clientRule.getClient().subscribe(listener);
+
+        int number_of_iterations = 1;
+        int  numberOfRequests = 3;
+
+        makeRequests(number_of_iterations, numberOfRequests);
+
+        sleep(clientRule.getPoolConfig().getMaxIdleTimeMillis());
+
+        assertThat("Permits should be 10", clientRule.getPoolConfig().getPoolLimitDeterminationStrategy().getAvailablePermits(), equalTo(10));
+    }
+
+    private void sleep(long i) {
+        try {
+            Thread.sleep(i);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    private void makeRequests(int number_of_iterations, int numberOfRequests) {
+        for (int j = 0; j < number_of_iterations; j++) {
+
+            //List<Observable<String>> results = new ArrayList<>();
+
+            sleep(100);
+
+            List<Observable<String>> results = new ArrayList<>();
+
+            //Just giving the client some time to recover
+            sleep(100);
+
+            for (int i = 0; i < numberOfRequests; i++) {
+                results.add(
+                    fromCallable(new Func0<PooledConnection<ByteBuf, ByteBuf>>() {
+                        @Override
+                        public PooledConnection<ByteBuf, ByteBuf> call() {
+                            return clientRule.connect();
+                        }
+                    })
+                        .flatMap(new Func1<PooledConnection<ByteBuf, ByteBuf>, Observable<String>>() {
+                            @Override
+                            public Observable<String> call(PooledConnection<ByteBuf, ByteBuf> connection) {
+                                return connection.writeStringAndFlushOnEach(just("Hello"))
+                                    .toCompletable()
+                                    .<ByteBuf>toObservable()
+                                    .concatWith(connection.getInput())
+                                    .take(1)
+                                    .single()
+                                    .map(new Func1<ByteBuf, String>() {
+                                        @Override
+                                        public String call(ByteBuf byteBuf) {
+                                            try {
+                                                byte[] bytes = new byte[byteBuf.readableBytes()];
+                                                byteBuf.readBytes(bytes);
+                                                return new String(bytes);
+                                            } finally {
+                                                byteBuf.release();
+                                            }
+                                        }
+                                    });
+                            }
+                        }));
+            }
+            AssertableSubscriber<String> test = Observable.merge(results).test();
+            test.awaitTerminalEvent();
+        }
+    }
 }
diff --git a/rxnetty-tcp/src/test/java/io/reactivex/netty/protocol/tcp/client/TcpClientRule.java b/rxnetty-tcp/src/test/java/io/reactivex/netty/protocol/tcp/client/TcpClientRule.java
@@ -18,12 +18,9 @@
 
 import io.netty.buffer.ByteBuf;
 import io.reactivex.netty.channel.Connection;
-import io.reactivex.netty.client.ConnectionProvider;
-import io.reactivex.netty.client.ConnectionProviderFactory;
 import io.reactivex.netty.client.Host;
-import io.reactivex.netty.client.HostConnector;
+import io.reactivex.netty.client.pool.PoolConfig;
 import io.reactivex.netty.client.pool.PooledConnection;
-import io.reactivex.netty.client.pool.PooledConnectionProvider;
 import io.reactivex.netty.client.pool.SingleHostPoolingProviderFactory;
 import io.reactivex.netty.protocol.tcp.server.ConnectionHandler;
 import io.reactivex.netty.protocol.tcp.server.TcpServer;
@@ -39,13 +36,14 @@
 import java.net.InetSocketAddress;
 import java.util.concurrent.atomic.AtomicBoolean;
 
-import static org.hamcrest.MatcherAssert.*;
-import static org.hamcrest.Matchers.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasSize;
 
 public class TcpClientRule extends ExternalResource {
 
-    private TcpServer<ByteBuf, ByteBuf> server;
-    private TcpClient<ByteBuf, ByteBuf> client;
+    private TcpServer<ByteBuf, ByteBuf>  server;
+    private TcpClient<ByteBuf, ByteBuf>  client;
+    private PoolConfig<ByteBuf, ByteBuf> poolConfig;
 
     @Override
     public Statement apply(final Statement base, Description description) {
@@ -59,10 +57,18 @@ public void evaluate() throws Throwable {
     }
 
     public void startServer(int maxConnections) {
+        startServer(maxConnections, false);
+    }
+
+    public void startServer(int maxConnections, final boolean failing) {
         server.start(new ConnectionHandler<ByteBuf, ByteBuf>() {
             @Override
             public Observable<Void> handle(Connection<ByteBuf, ByteBuf> newConnection) {
-                return newConnection.writeAndFlushOnEach(newConnection.getInput());
+                if(failing) {
+                    throw new RuntimeException("exception");
+                } else {
+                    return newConnection.writeAndFlushOnEach(newConnection.getInput());
+                }
             }
         });
         createClient(maxConnections);
@@ -131,8 +137,9 @@ public Observable<PooledConnection<ByteBuf, ByteBuf>> call() {
 
     private void createClient(final int maxConnections) {
         InetSocketAddress serverAddr = new InetSocketAddress("127.0.0.1", server.getServerPort());
-        client = TcpClient.newClient(SingleHostPoolingProviderFactory.<ByteBuf, ByteBuf>createBounded(maxConnections),
-                                     Observable.just(new Host(serverAddr)));
+        poolConfig = new PoolConfig<ByteBuf, ByteBuf>().maxConnections(maxConnections);
+        SingleHostPoolingProviderFactory<ByteBuf, ByteBuf> bounded = SingleHostPoolingProviderFactory.create(poolConfig);
+        client = TcpClient.newClient(bounded, Observable.just(new Host(serverAddr)));
     }
 
     public TcpServer<ByteBuf, ByteBuf> getServer() {
@@ -142,4 +149,8 @@ public TcpServer<ByteBuf, ByteBuf> getServer() {
     public TcpClient<ByteBuf, ByteBuf> getClient() {
         return client;
     }
+
+    public PoolConfig<ByteBuf, ByteBuf> getPoolConfig() {
+        return poolConfig;
+    }
 }