Merge #1090 into 1.2.0-SNAPSHOT

Signed-off-by: OlegDokuka <odokuka@vmware.com>
rsocket · Apr 21, 2023 · 308e4c3 · 308e4c3
2 parents 47e4e3b + 5547cb1
commit 308e4c3
Show file tree

Hide file tree

Showing 25 changed files with 1,180 additions and 785 deletions.
diff --git a/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java b/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java
@@ -450,8 +450,8 @@ public void accept(RSocket rSocket, Throwable t) {
 
     @Override
     public void request(long n) {
-      this.main.request(n);
       super.request(n);
+      this.main.request(n);
     }
 
     public void cancel() {

diff --git a/rsocket-core/src/test/java/io/rsocket/core/ClientServerInputMultiplexerTest.java b/rsocket-core/src/test/java/io/rsocket/core/ClientServerInputMultiplexerTest.java
@@ -56,12 +56,20 @@ public void clientSplits() {
     clientMultiplexer
         .asClientConnection()
         .receive()
-        .doOnNext(f -> clientFrames.incrementAndGet())
+        .doOnNext(
+            f -> {
+              clientFrames.incrementAndGet();
+              f.release();
+            })
         .subscribe();
     clientMultiplexer
         .asServerConnection()
         .receive()
-        .doOnNext(f -> serverFrames.incrementAndGet())
+        .doOnNext(
+            f -> {
+              serverFrames.incrementAndGet();
+              f.release();
+            })
         .subscribe();
 
     source.addToReceivedBuffer(errorFrame(1).retain());
@@ -101,12 +109,20 @@ public void serverSplits() {
     serverMultiplexer
         .asClientConnection()
         .receive()
-        .doOnNext(f -> clientFrames.incrementAndGet())
+        .doOnNext(
+            f -> {
+              clientFrames.incrementAndGet();
+              f.release();
+            })
         .subscribe();
     serverMultiplexer
         .asServerConnection()
         .receive()
-        .doOnNext(f -> serverFrames.incrementAndGet())
+        .doOnNext(
+            f -> {
+              serverFrames.incrementAndGet();
+              f.release();
+            })
         .subscribe();
 
     source.addToReceivedBuffer(errorFrame(1).retain());

diff --git a/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java b/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java
@@ -19,6 +19,7 @@
 import io.netty.util.CharsetUtil;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
+import io.rsocket.FrameAssert;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.RaceTestConstants;
@@ -434,6 +435,8 @@ public void shouldBeAbleToResolveOriginalSource() {
     assertSubscriber1.assertTerminated().assertValueCount(1);
 
     Assertions.assertThat(assertSubscriber1.values()).isEqualTo(assertSubscriber.values());
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -457,6 +460,13 @@ public void shouldDisposeOriginalSource() {
         .assertErrorMessage("Disposed");
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
+
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -494,6 +504,13 @@ public Mono<Void> onClose() {
     onCloseSubscriber.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
+
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -515,6 +532,13 @@ public void shouldResolveOnStartSource() {
     assertSubscriber1.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
+
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -536,6 +560,13 @@ public void shouldNotStartIfAlreadyDisposed() {
     assertSubscriber1.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
+
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -553,6 +584,11 @@ public void shouldBeRestartedIfSourceWasClosed() {
 
     rule.socket.dispose();
 
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
     terminateSubscriber.assertNotTerminated();
     Assertions.assertThat(rule.client.isDisposed()).isFalse();
 
@@ -576,6 +612,13 @@ public void shouldBeRestartedIfSourceWasClosed() {
     Assertions.assertThat(rule.client.connect()).isFalse();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
+
+    FrameAssert.assertThat(rule.connection.awaitFrame())
+        .hasStreamIdZero()
+        .hasData("Disposed")
+        .hasNoLeaks();
+
+    rule.allocator.assertHasNoLeaks();
   }
 
   @Test
@@ -603,6 +646,13 @@ public void shouldDisposeOriginalSourceIfRacing() {
           .assertTerminated()
           .assertError(CancellationException.class)
           .assertErrorMessage("Disposed");
+
+      ByteBuf buf;
+      while ((buf = rule.connection.pollFrame()) != null) {
+        FrameAssert.assertThat(buf).hasStreamIdZero().hasData("Disposed").hasNoLeaks();
+      }
+
+      rule.allocator.assertHasNoLeaks();
     }
   }
 
@@ -632,8 +682,14 @@ public void shouldStartOriginalSourceOnceIfRacing() {
       AssertSubscriber<Void> assertSubscriber1 = AssertSubscriber.create();
 
       rule.client.onClose().subscribe(assertSubscriber1);
+      FrameAssert.assertThat(rule.connection.awaitFrame())
+          .hasStreamIdZero()
+          .hasData("Disposed")
+          .hasNoLeaks();
 
       assertSubscriber1.assertTerminated().assertComplete();
+
+      rule.allocator.assertHasNoLeaks();
     }
   }