diff --git a/.github/workflows/gradle-all.yml b/.github/workflows/gradle-all.yml index f2df20620..abbd14106 100644 --- a/.github/workflows/gradle-all.yml +++ b/.github/workflows/gradle-all.yml @@ -5,7 +5,7 @@ on: # but only for the non master/1.0.x branches push: branches-ignore: - - 1.0.x + - 1.1.x - master jobs: diff --git a/.github/workflows/gradle-main.yml b/.github/workflows/gradle-main.yml index 904c45fb7..469ccb103 100644 --- a/.github/workflows/gradle-main.yml +++ b/.github/workflows/gradle-main.yml @@ -2,11 +2,11 @@ name: Main Branches Java CI on: # Trigger the workflow on push - # but only for the master/1.0.x branch + # but only for the master/1.1.x branch push: branches: - master - - 1.0.x + - 1.1.x jobs: build: diff --git a/README.md b/README.md index b495b6b45..96c155bc2 100644 --- a/README.md +++ b/README.md @@ -29,8 +29,8 @@ repositories { maven { url 'https://repo.spring.io/milestone' } // Reactor milestones (if needed) } dependencies { - implementation 'io.rsocket:rsocket-core:1.1.1' - implementation 'io.rsocket:rsocket-transport-netty:1.1.1' + implementation 'io.rsocket:rsocket-core:1.1.2' + implementation 'io.rsocket:rsocket-transport-netty:1.1.2' } ``` @@ -44,8 +44,8 @@ repositories { maven { url 'https://repo.spring.io/snapshot' } // Reactor snapshots (if needed) } dependencies { - implementation 'io.rsocket:rsocket-core:1.1.2-SNAPSHOT' - implementation 'io.rsocket:rsocket-transport-netty:1.1.2-SNAPSHOT' + implementation 'io.rsocket:rsocket-core:1.1.3-SNAPSHOT' + implementation 'io.rsocket:rsocket-transport-netty:1.1.3-SNAPSHOT' } ``` diff --git a/benchmarks/build.gradle b/benchmarks/build.gradle index 0b8bc601b..74e571d1f 100644 --- a/benchmarks/build.gradle +++ b/benchmarks/build.gradle @@ -14,8 +14,8 @@ dependencies { compileOnly "io.rsocket:rsocket-transport-local:${perfBaselineVersion}" compileOnly "io.rsocket:rsocket-transport-netty:${perfBaselineVersion}" - implementation "org.openjdk.jmh:jmh-core:1.21" - annotationProcessor "org.openjdk.jmh:jmh-generator-annprocess:1.21" + implementation "org.openjdk.jmh:jmh-core:1.35" + annotationProcessor "org.openjdk.jmh:jmh-generator-annprocess:1.35" current project(':rsocket-core') current project(':rsocket-transport-local') diff --git a/build.gradle b/build.gradle index a3ff46a4f..8399c27e1 100644 --- a/build.gradle +++ b/build.gradle @@ -16,8 +16,8 @@ plugins { id 'com.github.sherter.google-java-format' version '0.9' apply false - id 'me.champeau.jmh' version '0.6.6' apply false - id 'io.spring.dependency-management' version '1.0.11.RELEASE' apply false + id 'me.champeau.jmh' version '0.6.7' apply false + id 'io.spring.dependency-management' version '1.0.13.RELEASE' apply false id 'io.morethan.jmhreport' version '0.9.0' apply false id 'io.github.reyerizo.gradle.jcstress' version '0.8.13' apply false id 'com.github.vlsi.gradle-extensions' version '1.76' apply false @@ -33,14 +33,14 @@ subprojects { apply plugin: 'com.github.sherter.google-java-format' apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.17' + ext['reactor-bom.version'] = '2020.0.23' ext['logback.version'] = '1.2.10' - ext['netty-bom.version'] = '4.1.75.Final' - ext['netty-boringssl.version'] = '2.0.51.Final' + ext['netty-bom.version'] = '4.1.81.Final' + ext['netty-boringssl.version'] = '2.0.54.Final' ext['hdrhistogram.version'] = '2.1.12' ext['mockito.version'] = '4.4.0' ext['slf4j.version'] = '1.7.36' - ext['jmh.version'] = '1.33' + ext['jmh.version'] = '1.35' ext['junit.version'] = '5.8.1' ext['micrometer.version'] = '1.8.4' ext['assertj.version'] = '3.22.0' diff --git a/gradle.properties b/gradle.properties index e9219dfe6..7b5ac2349 100644 --- a/gradle.properties +++ b/gradle.properties @@ -11,5 +11,5 @@ # See the License for the specific language governing permissions and # limitations under the License. # -version=1.1.2 -perfBaselineVersion=1.1.1 +version=1.1.3 +perfBaselineVersion=1.1.2 diff --git a/rsocket-core/src/jcstress/java/io/rsocket/core/ReconnectMonoStressTest.java b/rsocket-core/src/jcstress/java/io/rsocket/core/ReconnectMonoStressTest.java index e01b1d704..ef79d344d 100644 --- a/rsocket-core/src/jcstress/java/io/rsocket/core/ReconnectMonoStressTest.java +++ b/rsocket-core/src/jcstress/java/io/rsocket/core/ReconnectMonoStressTest.java @@ -517,7 +517,7 @@ public void arbiter(IIIIII_Result r) { id = {"1, 0, 1, 0, 1, 2"}, expect = ACCEPTABLE) @State - public static class SubscribeBlockRace extends BaseStressTest { + public static class SubscribeBlockConnectRace extends BaseStressTest { String receivedValue; @@ -543,6 +543,11 @@ void subscribe() { reconnectMono.subscribe(stressSubscriber); } + @Actor + void connect() { + reconnectMono.resolvingInner.connect(); + } + @Arbiter public void arbiter(IIIIII_Result r) { r.r1 = stressSubscription.subscribes; diff --git a/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java b/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java index 4dc250158..5119814cd 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/core/DefaultRSocketClient.java @@ -35,6 +35,7 @@ import reactor.core.publisher.Mono; import reactor.core.publisher.MonoOperator; import reactor.core.publisher.Operators; +import reactor.core.publisher.Sinks; import reactor.util.annotation.Nullable; import reactor.util.context.Context; @@ -65,6 +66,8 @@ class DefaultRSocketClient extends ResolvingOperator final Mono source; + final Sinks.Empty onDisposeSink; + volatile Subscription s; static final AtomicReferenceFieldUpdater S = @@ -72,12 +75,18 @@ class DefaultRSocketClient extends ResolvingOperator DefaultRSocketClient(Mono source) { this.source = unwrapReconnectMono(source); + this.onDisposeSink = Sinks.empty(); } private Mono unwrapReconnectMono(Mono source) { return source instanceof ReconnectMono ? ((ReconnectMono) source).getSource() : source; } + @Override + public Mono onClose() { + return this.onDisposeSink.asMono(); + } + @Override public Mono source() { return Mono.fromDirect(this); @@ -194,6 +203,12 @@ protected void doOnValueExpired(RSocket value) { @Override protected void doOnDispose() { Operators.terminate(S, this); + final RSocket value = this.value; + if (value != null) { + value.onClose().subscribe(null, onDisposeSink::tryEmitError, onDisposeSink::tryEmitEmpty); + } else { + onDisposeSink.tryEmitEmpty(); + } } static final class FlatMapMain implements CoreSubscriber, Context, Scannable { diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketClient.java index 81392e661..32e3c229d 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketClient.java @@ -15,12 +15,13 @@ */ package io.rsocket.core; +import io.rsocket.Closeable; import io.rsocket.Payload; import io.rsocket.RSocket; import org.reactivestreams.Publisher; -import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import sun.reflect.generics.reflectiveObjects.NotImplementedException; /** * Contract for performing RSocket requests. @@ -74,7 +75,22 @@ * @since 1.1 * @see io.rsocket.loadbalance.LoadbalanceRSocketClient */ -public interface RSocketClient extends Disposable { +public interface RSocketClient extends Closeable { + + /** + * Connect to the remote rsocket endpoint, if not yet connected. This method is a shortcut for + * {@code RSocketClient#source().subscribe()}. + * + * @return {@code true} if an attempt to connect was triggered or if already connected, or {@code + * false} if the client is terminated. + */ + default boolean connect() { + throw new NotImplementedException(); + } + + default Mono onClose() { + return Mono.error(new NotImplementedException()); + } /** Return the underlying source used to obtain a shared {@link RSocket} connection. */ Mono source(); diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketClientAdapter.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketClientAdapter.java index cc94f4102..ae8b7da97 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketClientAdapter.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketClientAdapter.java @@ -41,11 +41,21 @@ public RSocket rsocket() { return rsocket; } + @Override + public boolean connect() { + throw new UnsupportedOperationException("Connect does not apply to a server side RSocket"); + } + @Override public Mono source() { return Mono.just(rsocket); } + @Override + public Mono onClose() { + return rsocket.onClose(); + } + @Override public Mono fireAndForget(Mono payloadMono) { return payloadMono.flatMap(rsocket::fireAndForget); diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java index edd13b48c..432c0f0f5 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketConnector.java @@ -590,7 +590,7 @@ public Mono connect(Supplier transportSupplier) { dataMimeType, setupPayload); - sourceConnection.sendFrame(0, setupFrame.retain()); + sourceConnection.sendFrame(0, setupFrame.retainedSlice()); return clientSetup .init(sourceConnection) diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java index c10e86d56..bf298706a 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java @@ -345,15 +345,17 @@ private void terminate(Throwable e) { requesterLeaseTracker.dispose(e); } + final Collection activeStreamsCopy; synchronized (this) { final IntObjectMap activeStreams = this.activeStreams; - final Collection activeStreamsCopy = new ArrayList<>(activeStreams.values()); - for (FrameHandler handler : activeStreamsCopy) { - if (handler != null) { - try { - handler.handleError(e); - } catch (Throwable ignored) { - } + activeStreamsCopy = new ArrayList<>(activeStreams.values()); + } + + for (FrameHandler handler : activeStreamsCopy) { + if (handler != null) { + try { + handler.handleError(e); + } catch (Throwable ignored) { } } } diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java index b2f084f51..ce4fe70a3 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java @@ -185,15 +185,18 @@ final void doOnDispose() { requestHandler.dispose(); } - private synchronized void cleanUpSendingSubscriptions() { - final IntObjectMap activeStreams = this.activeStreams; - final Collection activeStreamsCopy = new ArrayList<>(activeStreams.values()); + private void cleanUpSendingSubscriptions() { + final Collection activeStreamsCopy; + synchronized (this) { + final IntObjectMap activeStreams = this.activeStreams; + activeStreamsCopy = new ArrayList<>(activeStreams.values()); + } + for (FrameHandler handler : activeStreamsCopy) { if (handler != null) { handler.handleCancel(); } } - activeStreams.clear(); } final void handleFrame(ByteBuf frame) { diff --git a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java index 809125402..aab491793 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java @@ -86,6 +86,8 @@ final class RequestChannelRequesterFlux extends Flux Context cachedContext; CoreSubscriber inboundSubscriber; boolean inboundDone; + long requested; + long produced; CompositeByteBuf frames; @@ -138,6 +140,8 @@ public final void request(long n) { return; } + this.requested = Operators.addCap(this.requested, n); + long previousState = addRequestN(STATE, this, n, this.requesterLeaseTracker == null); if (isTerminated(previousState)) { return; @@ -706,6 +710,27 @@ public final void handlePayload(Payload value) { return; } + final long produced = this.produced; + if (this.requested == produced) { + value.release(); + if (!tryCancel()) { + return; + } + + final Throwable cause = + Exceptions.failWithOverflow( + "The number of messages received exceeds the number requested"); + final RequestInterceptor requestInterceptor = this.requestInterceptor; + if (requestInterceptor != null) { + requestInterceptor.onTerminate(streamId, FrameType.REQUEST_CHANNEL, cause); + } + + this.inboundSubscriber.onError(cause); + return; + } + + this.produced = produced + 1; + this.inboundSubscriber.onNext(value); } } diff --git a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelResponderSubscriber.java b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelResponderSubscriber.java index 8dac9858d..32128fee4 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelResponderSubscriber.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelResponderSubscriber.java @@ -88,6 +88,8 @@ final class RequestChannelResponderSubscriber extends Flux boolean inboundDone; boolean outboundDone; + long requested; + long produced; public RequestChannelResponderSubscriber( int streamId, @@ -179,6 +181,8 @@ public void request(long n) { return; } + this.requested = Operators.addCap(this.requested, n); + long previousState = StateUtils.addRequestN(STATE, this, n); if (isTerminated(previousState)) { // full termination can be the result of both sides completion / cancelFrame / remote or local @@ -196,6 +200,9 @@ public void request(long n) { Payload firstPayload = this.firstPayload; if (firstPayload != null) { this.firstPayload = null; + + this.produced++; + inboundSubscriber.onNext(firstPayload); } @@ -216,6 +223,8 @@ public void request(long n) { final Payload firstPayload = this.firstPayload; this.firstPayload = null; + this.produced++; + inboundSubscriber.onNext(firstPayload); inboundSubscriber.onComplete(); @@ -238,6 +247,9 @@ public void request(long n) { final Payload firstPayload = this.firstPayload; this.firstPayload = null; + + this.produced++; + inboundSubscriber.onNext(firstPayload); previousState = markFirstFrameSent(STATE, this); @@ -416,6 +428,58 @@ final void handlePayload(Payload p) { return; } + final long produced = this.produced; + if (this.requested == produced) { + p.release(); + + this.inboundDone = true; + + final Throwable cause = + Exceptions.failWithOverflow( + "The number of messages received exceeds the number requested"); + boolean wasThrowableAdded = Exceptions.addThrowable(INBOUND_ERROR, this, cause); + + long previousState = markTerminated(STATE, this); + if (isTerminated(previousState)) { + if (!wasThrowableAdded) { + Operators.onErrorDropped(cause, this.inboundSubscriber.currentContext()); + } + return; + } + + this.requesterResponderSupport.remove(this.streamId, this); + + this.connection.sendFrame( + streamId, + ErrorFrameCodec.encode( + this.allocator, streamId, new CanceledException(cause.getMessage()))); + + if (!isSubscribed(previousState)) { + final Payload firstPayload = this.firstPayload; + this.firstPayload = null; + firstPayload.release(); + } else if (isFirstFrameSent(previousState) && !isInboundTerminated(previousState)) { + Throwable inboundError = Exceptions.terminate(INBOUND_ERROR, this); + if (inboundError != TERMINATED) { + //noinspection ConstantConditions + this.inboundSubscriber.onError(inboundError); + } + } + + // this is downstream subscription so need to cancel it just in case error signal has not + // reached it + // needs for disconnected upstream and downstream case + this.outboundSubscription.cancel(); + + final RequestInterceptor interceptor = requestInterceptor; + if (interceptor != null) { + interceptor.onTerminate(this.streamId, FrameType.REQUEST_CHANNEL, cause); + } + return; + } + + this.produced = produced + 1; + this.inboundSubscriber.onNext(p); } } diff --git a/rsocket-core/src/main/java/io/rsocket/core/RequestStreamRequesterFlux.java b/rsocket-core/src/main/java/io/rsocket/core/RequestStreamRequesterFlux.java index 424451a58..55ec43feb 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RequestStreamRequesterFlux.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RequestStreamRequesterFlux.java @@ -65,6 +65,8 @@ final class RequestStreamRequesterFlux extends Flux CoreSubscriber inboundSubscriber; CompositeByteBuf frames; boolean done; + long requested; + long produced; RequestStreamRequesterFlux(Payload payload, RequesterResponderSupport requesterResponderSupport) { this.allocator = requesterResponderSupport.getAllocator(); @@ -134,6 +136,8 @@ public final void request(long n) { return; } + this.requested = Operators.addCap(this.requested, n); + final RequesterLeaseTracker requesterLeaseTracker = this.requesterLeaseTracker; final boolean leaseEnabled = requesterLeaseTracker != null; final long previousState = addRequestN(STATE, this, n, !leaseEnabled); @@ -295,6 +299,34 @@ public final void handlePayload(Payload p) { return; } + final long produced = this.produced; + if (this.requested == produced) { + p.release(); + + long previousState = markTerminated(STATE, this); + if (isTerminated(previousState)) { + return; + } + + final int streamId = this.streamId; + this.requesterResponderSupport.remove(streamId, this); + + final IllegalStateException cause = + Exceptions.failWithOverflow( + "The number of messages received exceeds the number requested"); + this.connection.sendFrame(streamId, CancelFrameCodec.encode(this.allocator, streamId)); + + final RequestInterceptor requestInterceptor = this.requestInterceptor; + if (requestInterceptor != null) { + requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, cause); + } + + this.inboundSubscriber.onError(cause); + return; + } + + this.produced = produced + 1; + this.inboundSubscriber.onNext(p); } diff --git a/rsocket-core/src/main/java/io/rsocket/core/ResolvingOperator.java b/rsocket-core/src/main/java/io/rsocket/core/ResolvingOperator.java index 85c4a17a7..50bef5b70 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/ResolvingOperator.java +++ b/rsocket-core/src/main/java/io/rsocket/core/ResolvingOperator.java @@ -331,6 +331,30 @@ protected void doOnDispose() { // no ops } + public final boolean connect() { + for (; ; ) { + final BiConsumer[] a = this.subscribers; + + if (a == TERMINATED) { + return false; + } + + if (a == READY) { + return true; + } + + if (a != EMPTY_UNSUBSCRIBED) { + // do nothing if already started + return true; + } + + if (SUBSCRIBERS.compareAndSet(this, a, EMPTY_SUBSCRIBED)) { + this.doSubscribe(); + return true; + } + } + } + final int add(BiConsumer ps) { for (; ; ) { BiConsumer[] a = this.subscribers; diff --git a/rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java index 98bed7ba7..09026356f 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java @@ -22,13 +22,10 @@ import reactor.core.publisher.Sinks; public abstract class BaseDuplexConnection implements DuplexConnection { - protected Sinks.Empty onClose = Sinks.empty(); + protected final Sinks.Empty onClose = Sinks.empty(); + protected final UnboundedProcessor sender = new UnboundedProcessor(onClose::tryEmitEmpty); - protected UnboundedProcessor sender = new UnboundedProcessor(); - - public BaseDuplexConnection() { - onClose().doFinally(s -> doOnClose()).subscribe(); - } + public BaseDuplexConnection() {} @Override public void sendFrame(int streamId, ByteBuf frame) { @@ -42,13 +39,13 @@ public void sendFrame(int streamId, ByteBuf frame) { protected abstract void doOnClose(); @Override - public final Mono onClose() { + public Mono onClose() { return onClose.asMono(); } @Override public final void dispose() { - onClose.tryEmitEmpty(); + doOnClose(); } @Override diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/LoadbalanceRSocketClient.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/LoadbalanceRSocketClient.java index 1b677edba..21ea3d836 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/LoadbalanceRSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/LoadbalanceRSocketClient.java @@ -40,6 +40,16 @@ private LoadbalanceRSocketClient(RSocketPool rSocketPool) { this.rSocketPool = rSocketPool; } + @Override + public Mono onClose() { + return rSocketPool.onClose(); + } + + @Override + public boolean connect() { + return rSocketPool.connect(); + } + /** Return {@code Mono} that selects an RSocket from the underlying pool. */ @Override public Mono source() { diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/PooledRSocket.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/PooledRSocket.java index 1e7f09ec4..a77329d31 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/PooledRSocket.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/PooledRSocket.java @@ -26,6 +26,7 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Operators; +import reactor.core.publisher.Sinks; import reactor.util.context.Context; /** Default implementation of {@link RSocket} stored in {@link RSocketPool} */ @@ -35,6 +36,7 @@ final class PooledRSocket extends ResolvingOperator final RSocketPool parent; final Mono rSocketSource; final LoadbalanceTarget loadbalanceTarget; + final Sinks.Empty onCloseSink; volatile Subscription s; @@ -46,6 +48,7 @@ final class PooledRSocket extends ResolvingOperator this.parent = parent; this.rSocketSource = rSocketSource; this.loadbalanceTarget = loadbalanceTarget; + this.onCloseSink = Sinks.unsafe().empty(); } @Override @@ -155,6 +158,12 @@ void doCleanup(Throwable t) { break; } } + + if (t == ON_DISPOSE) { + this.onCloseSink.tryEmitEmpty(); + } else { + this.onCloseSink.tryEmitError(t); + } } @Override @@ -165,6 +174,13 @@ protected void doOnValueExpired(RSocket value) { @Override protected void doOnDispose() { Operators.terminate(S, this); + + final RSocket value = this.value; + if (value != null) { + value.onClose().subscribe(null, onCloseSink::tryEmitError, onCloseSink::tryEmitEmpty); + } else { + onCloseSink.tryEmitEmpty(); + } } @Override @@ -193,7 +209,12 @@ public Mono metadataPush(Payload payload) { } LoadbalanceTarget target() { - return loadbalanceTarget; + return this.loadbalanceTarget; + } + + @Override + public Mono onClose() { + return this.onCloseSink.asMono(); } @Override diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/RSocketPool.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/RSocketPool.java index bf6f53830..59d9678d0 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/RSocketPool.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/RSocketPool.java @@ -16,6 +16,7 @@ package io.rsocket.loadbalance; import io.netty.util.ReferenceCountUtil; +import io.rsocket.Closeable; import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.core.RSocketConnector; @@ -28,16 +29,18 @@ import java.util.ListIterator; import java.util.concurrent.CancellationException; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import java.util.stream.Collectors; import org.reactivestreams.Publisher; import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Operators; +import reactor.core.publisher.Sinks; import reactor.util.annotation.Nullable; class RSocketPool extends ResolvingOperator - implements CoreSubscriber> { + implements CoreSubscriber>, Closeable { static final AtomicReferenceFieldUpdater ACTIVE_SOCKETS = AtomicReferenceFieldUpdater.newUpdater( @@ -49,6 +52,7 @@ class RSocketPool extends ResolvingOperator final DeferredResolutionRSocket deferredResolutionRSocket = new DeferredResolutionRSocket(this); final RSocketConnector connector; final LoadbalanceStrategy loadbalanceStrategy; + final Sinks.Empty onAllClosedSink = Sinks.unsafe().empty(); volatile PooledRSocket[] activeSockets; volatile Subscription s; @@ -64,6 +68,11 @@ public RSocketPool( targetPublisher.subscribe(this); } + @Override + public Mono onClose() { + return onAllClosedSink.asMono(); + } + @Override protected void doOnDispose() { Operators.terminate(S, this); @@ -72,6 +81,14 @@ protected void doOnDispose() { for (RSocket rSocket : activeSockets) { rSocket.dispose(); } + + if (activeSockets.length > 0) { + Mono.whenDelayError( + Arrays.stream(activeSockets).map(RSocket::onClose).collect(Collectors.toList())) + .subscribe(null, onAllClosedSink::tryEmitError, onAllClosedSink::tryEmitEmpty); + } else { + onAllClosedSink.tryEmitEmpty(); + } } @Override diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/ResolvingOperator.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/ResolvingOperator.java index a25bcc584..52f16e166 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/ResolvingOperator.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/ResolvingOperator.java @@ -327,6 +327,30 @@ protected void doOnDispose() { // no ops } + public final boolean connect() { + for (; ; ) { + final BiConsumer[] a = this.subscribers; + + if (a == TERMINATED) { + return false; + } + + if (a == READY) { + return true; + } + + if (a != EMPTY_UNSUBSCRIBED) { + // do nothing if already started + return true; + } + + if (SUBSCRIBERS.compareAndSet(this, a, EMPTY_SUBSCRIBED)) { + this.doSubscribe(); + return true; + } + } + } + final int add(BiConsumer ps) { for (; ; ) { BiConsumer[] a = this.subscribers; diff --git a/rsocket-core/src/main/resources/META-INF/native-image/io.rsocket/rsocket-core/reflect-config.json b/rsocket-core/src/main/resources/META-INF/native-image/io.rsocket/rsocket-core/reflect-config.json new file mode 100644 index 000000000..0a3844451 --- /dev/null +++ b/rsocket-core/src/main/resources/META-INF/native-image/io.rsocket/rsocket-core/reflect-config.json @@ -0,0 +1,130 @@ +[ + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.BaseLinkedQueueConsumerNodeRef" + }, + "name": "io.rsocket.internal.jctools.queues.BaseLinkedQueueConsumerNodeRef", + "fields": [ + { + "name": "consumerNode" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.BaseLinkedQueueProducerNodeRef" + }, + "name": "io.rsocket.internal.jctools.queues.BaseLinkedQueueProducerNodeRef", + "fields": [ + { + "name": "producerNode" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields" + }, + "name": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueColdProducerFields", + "fields": [ + { + "name": "producerLimit" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields" + }, + "name": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueConsumerFields", + "fields": [ + { + "name": "consumerIndex" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueProducerFields" + }, + "name": "io.rsocket.internal.jctools.queues.BaseMpscLinkedArrayQueueProducerFields", + "fields": [ + { + "name": "producerIndex" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.LinkedQueueNode" + }, + "name": "io.rsocket.internal.jctools.queues.LinkedQueueNode", + "fields": [ + { + "name": "next" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.MpscArrayQueueConsumerIndexField" + }, + "name": "io.rsocket.internal.jctools.queues.MpscArrayQueueConsumerIndexField", + "fields": [ + { + "name": "consumerIndex" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.MpscArrayQueueProducerIndexField" + }, + "name": "io.rsocket.internal.jctools.queues.MpscArrayQueueProducerIndexField", + "fields": [ + { + "name": "producerIndex" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.MpscArrayQueueProducerLimitField" + }, + "name": "io.rsocket.internal.jctools.queues.MpscArrayQueueProducerLimitField", + "fields": [ + { + "name": "producerLimit" + } + ] + }, + { + "condition": { + "typeReachable": "io.rsocket.internal.jctools.queues.UnsafeAccess" + }, + "name": "sun.misc.Unsafe", + "fields": [ + { + "name": "theUnsafe" + } + ], + "queriedMethods": [ + { + "name": "getAndAddLong", + "parameterTypes": [ + "java.lang.Object", + "long", + "long" + ] + }, + { + "name": "getAndSetObject", + "parameterTypes": [ + "java.lang.Object", + "long", + "java.lang.Object" + ] + } + ] + } +] \ No newline at end of file diff --git a/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java b/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java index 9085f1d8f..3c95fd65c 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java +++ b/rsocket-core/src/test/java/io/rsocket/core/DefaultRSocketClientTests.java @@ -28,7 +28,9 @@ import io.rsocket.frame.PayloadFrameCodec; import io.rsocket.frame.decoder.PayloadDecoder; import io.rsocket.internal.subscriber.AssertSubscriber; +import io.rsocket.test.util.TestDuplexConnection; import io.rsocket.util.ByteBufPayload; +import io.rsocket.util.RSocketProxy; import java.time.Duration; import java.util.ArrayList; import java.util.Collection; @@ -457,6 +459,125 @@ public void shouldDisposeOriginalSource() { Assertions.assertThat(rule.socket.isDisposed()).isTrue(); } + @Test + public void shouldReceiveOnCloseNotificationOnDisposeOriginalSource() { + Sinks.Empty onCloseDelayer = Sinks.empty(); + ClientSocketRule rule = + new ClientSocketRule() { + @Override + protected RSocket newRSocket() { + return new RSocketProxy(super.newRSocket()) { + @Override + public Mono onClose() { + return super.onClose().and(onCloseDelayer.asMono()); + } + }; + } + }; + rule.init(); + AssertSubscriber assertSubscriber = AssertSubscriber.create(); + rule.client.source().subscribe(assertSubscriber); + rule.delayer.run(); + assertSubscriber.assertTerminated().assertValueCount(1); + + rule.client.dispose(); + + Assertions.assertThat(rule.client.isDisposed()).isTrue(); + + AssertSubscriber onCloseSubscriber = AssertSubscriber.create(); + + rule.client.onClose().subscribe(onCloseSubscriber); + onCloseSubscriber.assertNotTerminated(); + + onCloseDelayer.tryEmitEmpty(); + + onCloseSubscriber.assertTerminated().assertComplete(); + + Assertions.assertThat(rule.socket.isDisposed()).isTrue(); + } + + @Test + public void shouldResolveOnStartSource() { + AssertSubscriber assertSubscriber = AssertSubscriber.create(); + Assertions.assertThat(rule.client.connect()).isTrue(); + rule.client.source().subscribe(assertSubscriber); + rule.delayer.run(); + assertSubscriber.assertTerminated().assertValueCount(1); + + rule.client.dispose(); + + Assertions.assertThat(rule.client.isDisposed()).isTrue(); + + AssertSubscriber assertSubscriber1 = AssertSubscriber.create(); + + rule.client.onClose().subscribe(assertSubscriber1); + + assertSubscriber1.assertTerminated().assertComplete(); + + Assertions.assertThat(rule.socket.isDisposed()).isTrue(); + } + + @Test + public void shouldNotStartIfAlreadyDisposed() { + Assertions.assertThat(rule.client.connect()).isTrue(); + Assertions.assertThat(rule.client.connect()).isTrue(); + rule.delayer.run(); + + rule.client.dispose(); + + Assertions.assertThat(rule.client.connect()).isFalse(); + + Assertions.assertThat(rule.client.isDisposed()).isTrue(); + + AssertSubscriber assertSubscriber1 = AssertSubscriber.create(); + + rule.client.onClose().subscribe(assertSubscriber1); + + assertSubscriber1.assertTerminated().assertComplete(); + + Assertions.assertThat(rule.socket.isDisposed()).isTrue(); + } + + @Test + public void shouldBeRestartedIfSourceWasClosed() { + AssertSubscriber assertSubscriber = AssertSubscriber.create(); + AssertSubscriber terminateSubscriber = AssertSubscriber.create(); + + Assertions.assertThat(rule.client.connect()).isTrue(); + rule.client.source().subscribe(assertSubscriber); + rule.client.onClose().subscribe(terminateSubscriber); + + rule.delayer.run(); + + assertSubscriber.assertTerminated().assertValueCount(1); + + rule.socket.dispose(); + + terminateSubscriber.assertNotTerminated(); + Assertions.assertThat(rule.client.isDisposed()).isFalse(); + + rule.connection = new TestDuplexConnection(rule.allocator); + rule.socket = rule.newRSocket(); + rule.producer = Sinks.one(); + + AssertSubscriber assertSubscriber2 = AssertSubscriber.create(); + + Assertions.assertThat(rule.client.connect()).isTrue(); + rule.client.source().subscribe(assertSubscriber2); + + rule.delayer.run(); + + assertSubscriber2.assertTerminated().assertValueCount(1); + + rule.client.dispose(); + + terminateSubscriber.assertTerminated().assertComplete(); + + Assertions.assertThat(rule.client.connect()).isFalse(); + + Assertions.assertThat(rule.socket.isDisposed()).isTrue(); + } + @Test public void shouldDisposeOriginalSourceIfRacing() { for (int i = 0; i < RaceTestConstants.REPEATS; i++) { @@ -485,7 +606,38 @@ public void shouldDisposeOriginalSourceIfRacing() { } } - public static class ClientSocketRule extends AbstractSocketRule { + @Test + public void shouldStartOriginalSourceOnceIfRacing() { + for (int i = 0; i < RaceTestConstants.REPEATS; i++) { + ClientSocketRule rule = new ClientSocketRule(); + + rule.init(); + + AssertSubscriber assertSubscriber = AssertSubscriber.create(); + + RaceTestUtils.race( + () -> rule.client.source().subscribe(assertSubscriber), () -> rule.client.connect()); + + Assertions.assertThat(rule.producer.currentSubscriberCount()).isOne(); + + rule.delayer.run(); + + assertSubscriber.assertTerminated(); + + rule.client.dispose(); + + Assertions.assertThat(rule.client.isDisposed()).isTrue(); + Assertions.assertThat(rule.socket.isDisposed()).isTrue(); + + AssertSubscriber assertSubscriber1 = AssertSubscriber.create(); + + rule.client.onClose().subscribe(assertSubscriber1); + + assertSubscriber1.assertTerminated().assertComplete(); + } + } + + public static class ClientSocketRule extends AbstractSocketRule { protected RSocketClient client; protected Runnable delayer; @@ -498,14 +650,16 @@ protected void doInit() { producer = Sinks.one(); client = new DefaultRSocketClient( - producer - .asMono() - .doOnCancel(() -> socket.dispose()) - .doOnDiscard(Disposable.class, Disposable::dispose)); + Mono.defer( + () -> + producer + .asMono() + .doOnCancel(() -> socket.dispose()) + .doOnDiscard(Disposable.class, Disposable::dispose))); } @Override - protected RSocketRequester newRSocket() { + protected RSocket newRSocket() { return new RSocketRequester( connection, PayloadDecoder.ZERO_COPY, diff --git a/rsocket-core/src/test/java/io/rsocket/core/RequestChannelRequesterFluxTest.java b/rsocket-core/src/test/java/io/rsocket/core/RequestChannelRequesterFluxTest.java index e39b0d690..c1e0a6876 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RequestChannelRequesterFluxTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RequestChannelRequesterFluxTest.java @@ -16,6 +16,7 @@ package io.rsocket.core; import static io.rsocket.frame.FrameLengthCodec.FRAME_LENGTH_MASK; +import static io.rsocket.frame.FrameType.CANCEL; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; @@ -40,6 +41,7 @@ import java.util.stream.Stream; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; @@ -513,6 +515,77 @@ public void errorShouldTerminateExecution(String terminationMode) { stateAssert.isTerminated(); } + @Test + public void failOnOverflow() { + final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); + final LeaksTrackingByteBufAllocator allocator = activeStreams.getAllocator(); + final TestDuplexConnection sender = activeStreams.getDuplexConnection(); + final TestPublisher publisher = TestPublisher.create(); + + final RequestChannelRequesterFlux requestChannelRequesterFlux = + new RequestChannelRequesterFlux(publisher, activeStreams); + final StateAssert stateAssert = + StateAssert.assertThat(requestChannelRequesterFlux); + + // state machine check + + stateAssert.isUnsubscribed(); + activeStreams.assertNoActiveStreams(); + + final AssertSubscriber assertSubscriber = + requestChannelRequesterFlux.subscribeWith(AssertSubscriber.create(0)); + activeStreams.assertNoActiveStreams(); + + // state machine check + stateAssert.hasSubscribedFlagOnly(); + + assertSubscriber.request(1); + stateAssert.hasSubscribedFlag().hasRequestN(1).hasNoFirstFrameSentFlag(); + activeStreams.assertNoActiveStreams(); + + Payload payload1 = TestRequesterResponderSupport.randomPayload(allocator); + + publisher.next(payload1.retain()); + + FrameAssert.assertThat(sender.awaitFrame()) + .typeOf(FrameType.REQUEST_CHANNEL) + .hasPayload(payload1) + .hasRequestN(1) + .hasNoLeaks(); + payload1.release(); + + stateAssert.hasSubscribedFlag().hasRequestN(1).hasFirstFrameSentFlag(); + activeStreams.assertHasStream(1, requestChannelRequesterFlux); + + publisher.assertMaxRequested(1); + + Payload nextPayload = TestRequesterResponderSupport.genericPayload(allocator); + requestChannelRequesterFlux.handlePayload(nextPayload); + + Payload unrequestedPayload = TestRequesterResponderSupport.genericPayload(allocator); + requestChannelRequesterFlux.handlePayload(unrequestedPayload); + + final ByteBuf cancelFrame = sender.awaitFrame(); + FrameAssert.assertThat(cancelFrame) + .isNotNull() + .typeOf(CANCEL) + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + assertSubscriber + .assertValuesWith(p -> PayloadAssert.assertThat(p).isSameAs(nextPayload).hasNoLeaks()) + .assertError() + .assertErrorMessage("The number of messages received exceeds the number requested"); + + publisher.assertWasCancelled(); + + activeStreams.assertNoActiveStreams(); + // state machine check + stateAssert.isTerminated(); + Assertions.assertThat(sender.isEmpty()).isTrue(); + } + /* * +--------------------------------+ * | Racing Test Cases | diff --git a/rsocket-core/src/test/java/io/rsocket/core/RequestChannelResponderSubscriberTest.java b/rsocket-core/src/test/java/io/rsocket/core/RequestChannelResponderSubscriberTest.java index 32af4e3b6..890458caf 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RequestChannelResponderSubscriberTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RequestChannelResponderSubscriberTest.java @@ -263,6 +263,143 @@ public void requestNFrameShouldBeSentOnSubscriptionAndThenSeparately(String comp allocator.assertHasNoLeaks(); } + @Test + public void failOnOverflow() { + final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); + final LeaksTrackingByteBufAllocator allocator = activeStreams.getAllocator(); + final TestDuplexConnection sender = activeStreams.getDuplexConnection(); + final Payload firstPayload = TestRequesterResponderSupport.genericPayload(allocator); + final TestPublisher publisher = TestPublisher.create(); + + final RequestChannelResponderSubscriber requestChannelResponderSubscriber = + new RequestChannelResponderSubscriber(1, 1, firstPayload, activeStreams); + final StateAssert stateAssert = + StateAssert.assertThat(requestChannelResponderSubscriber); + activeStreams.activeStreams.put(1, requestChannelResponderSubscriber); + + // state machine check + stateAssert.isUnsubscribed().hasRequestN(0); + activeStreams.assertHasStream(1, requestChannelResponderSubscriber); + + publisher.subscribe(requestChannelResponderSubscriber); + publisher.assertMaxRequested(1); + // state machine check + stateAssert.isUnsubscribed().hasRequestN(0); + + final AssertSubscriber assertSubscriber = + requestChannelResponderSubscriber.subscribeWith(AssertSubscriber.create(0)); + Assertions.assertThat(firstPayload.refCnt()).isOne(); + + // state machine check + stateAssert.hasSubscribedFlagOnly().hasRequestN(0); + + assertSubscriber.request(1); + + // state machine check + stateAssert.hasSubscribedFlag().hasFirstFrameSentFlag().hasRequestN(1); + + // should not send requestN since 1 is remaining + Assertions.assertThat(sender.isEmpty()).isTrue(); + + assertSubscriber.request(1); + + stateAssert.hasSubscribedFlag().hasRequestN(2).hasFirstFrameSentFlag(); + + // should not send requestN since 1 is remaining + FrameAssert.assertThat(sender.awaitFrame()) + .typeOf(REQUEST_N) + .hasStreamId(1) + .hasRequestN(1) + .hasNoLeaks(); + + Payload nextPayload = TestRequesterResponderSupport.genericPayload(allocator); + requestChannelResponderSubscriber.handlePayload(nextPayload); + + Payload unrequestedPayload = TestRequesterResponderSupport.genericPayload(allocator); + requestChannelResponderSubscriber.handlePayload(unrequestedPayload); + + final ByteBuf cancelErrorFrame = sender.awaitFrame(); + FrameAssert.assertThat(cancelErrorFrame) + .isNotNull() + .typeOf(ERROR) + .hasData("The number of messages received exceeds the number requested") + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + assertSubscriber + .assertValuesWith( + p -> PayloadAssert.assertThat(p).isSameAs(firstPayload).hasNoLeaks(), + p -> PayloadAssert.assertThat(p).isSameAs(nextPayload).hasNoLeaks()) + .assertErrorMessage("The number of messages received exceeds the number requested"); + + Assertions.assertThat(firstPayload.refCnt()).isZero(); + Assertions.assertThat(nextPayload.refCnt()).isZero(); + Assertions.assertThat(unrequestedPayload.refCnt()).isZero(); + stateAssert.isTerminated(); + activeStreams.assertNoActiveStreams(); + + Assertions.assertThat(sender.isEmpty()).isTrue(); + allocator.assertHasNoLeaks(); + } + + @Test + public void failOnOverflowBeforeFirstPayloadIsSent() { + final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); + final LeaksTrackingByteBufAllocator allocator = activeStreams.getAllocator(); + final TestDuplexConnection sender = activeStreams.getDuplexConnection(); + final Payload firstPayload = TestRequesterResponderSupport.genericPayload(allocator); + final TestPublisher publisher = TestPublisher.create(); + + final RequestChannelResponderSubscriber requestChannelResponderSubscriber = + new RequestChannelResponderSubscriber(1, 1, firstPayload, activeStreams); + final StateAssert stateAssert = + StateAssert.assertThat(requestChannelResponderSubscriber); + activeStreams.activeStreams.put(1, requestChannelResponderSubscriber); + + // state machine check + stateAssert.isUnsubscribed().hasRequestN(0); + activeStreams.assertHasStream(1, requestChannelResponderSubscriber); + + publisher.subscribe(requestChannelResponderSubscriber); + publisher.assertMaxRequested(1); + // state machine check + stateAssert.isUnsubscribed().hasRequestN(0); + + final AssertSubscriber assertSubscriber = + requestChannelResponderSubscriber.subscribeWith(AssertSubscriber.create(0)); + Assertions.assertThat(firstPayload.refCnt()).isOne(); + + // state machine check + stateAssert.hasSubscribedFlagOnly().hasRequestN(0); + + Payload unrequestedPayload = TestRequesterResponderSupport.genericPayload(allocator); + requestChannelResponderSubscriber.handlePayload(unrequestedPayload); + + final ByteBuf cancelErrorFrame = sender.awaitFrame(); + FrameAssert.assertThat(cancelErrorFrame) + .isNotNull() + .typeOf(ERROR) + .hasData("The number of messages received exceeds the number requested") + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + assertSubscriber.request(1); + + assertSubscriber + .assertValuesWith(p -> PayloadAssert.assertThat(p).isSameAs(firstPayload).hasNoLeaks()) + .assertErrorMessage("The number of messages received exceeds the number requested"); + + Assertions.assertThat(firstPayload.refCnt()).isZero(); + Assertions.assertThat(unrequestedPayload.refCnt()).isZero(); + stateAssert.isTerminated(); + activeStreams.assertNoActiveStreams(); + + Assertions.assertThat(sender.isEmpty()).isTrue(); + allocator.assertHasNoLeaks(); + } + /* * +--------------------------------+ * | Racing Test Cases | @@ -664,7 +801,7 @@ public void shouldHaveNoLeaksOnReassemblyAndCancelRacing(String terminationMode) ; final TestPublisher publisher = TestPublisher.createNoncompliant(DEFER_CANCELLATION, CLEANUP_ON_TERMINATE); - final AssertSubscriber assertSubscriber = new AssertSubscriber<>(1); + final AssertSubscriber assertSubscriber = new AssertSubscriber<>(2); Payload firstPayload = TestRequesterResponderSupport.genericPayload(allocator); final RequestChannelResponderSubscriber requestOperator = @@ -725,8 +862,17 @@ public void shouldHaveNoLeaksOnReassemblyAndCancelRacing(String terminationMode) assertSubscriber.assertTerminated().assertError(); } - final ByteBuf frame = sender.awaitFrame(); - FrameAssert.assertThat(frame) + final ByteBuf requstFrame = sender.awaitFrame(); + FrameAssert.assertThat(requstFrame) + .isNotNull() + .typeOf(REQUEST_N) + .hasRequestN(1) + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + final ByteBuf terminalFrame = sender.awaitFrame(); + FrameAssert.assertThat(terminalFrame) .isNotNull() .typeOf(terminationMode.equals("cancel") ? CANCEL : ERROR) .hasClientSideStreamId() diff --git a/rsocket-core/src/test/java/io/rsocket/core/RequestStreamRequesterFluxTest.java b/rsocket-core/src/test/java/io/rsocket/core/RequestStreamRequesterFluxTest.java index 88dd5441e..8702d1a80 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RequestStreamRequesterFluxTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RequestStreamRequesterFluxTest.java @@ -926,7 +926,7 @@ public void shouldErrorOnIncorrectRefCntInGivenPayloadLatePhase() { final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); final LeaksTrackingByteBufAllocator allocator = activeStreams.getAllocator(); final TestDuplexConnection sender = activeStreams.getDuplexConnection(); - ; + final Payload payload = ByteBufPayload.create(""); final RequestStreamRequesterFlux requestStreamRequesterFlux = @@ -1129,6 +1129,87 @@ static Stream> shouldErrorIfNoAvailabilityS .isInstanceOf(RuntimeException.class)); } + @Test + public void failOnOverflow() { + final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); + final LeaksTrackingByteBufAllocator allocator = activeStreams.getAllocator(); + final TestDuplexConnection sender = activeStreams.getDuplexConnection(); + final Payload payload = TestRequesterResponderSupport.genericPayload(allocator); + + final RequestStreamRequesterFlux requestStreamRequesterFlux = + new RequestStreamRequesterFlux(payload, activeStreams); + final StateAssert stateAssert = + StateAssert.assertThat(requestStreamRequesterFlux); + + // state machine check + + stateAssert.isUnsubscribed(); + activeStreams.assertNoActiveStreams(); + + final AssertSubscriber assertSubscriber = + requestStreamRequesterFlux.subscribeWith(AssertSubscriber.create(0)); + Assertions.assertThat(payload.refCnt()).isOne(); + activeStreams.assertNoActiveStreams(); + // state machine check + stateAssert.hasSubscribedFlagOnly(); + + assertSubscriber.request(1); + + Assertions.assertThat(payload.refCnt()).isZero(); + activeStreams.assertHasStream(1, requestStreamRequesterFlux); + + // state machine check + stateAssert.hasSubscribedFlag().hasRequestN(1).hasFirstFrameSentFlag(); + + final ByteBuf frame = sender.awaitFrame(); + FrameAssert.assertThat(frame) + .isNotNull() + .hasPayloadSize( + "testData".getBytes(CharsetUtil.UTF_8).length + + "testMetadata".getBytes(CharsetUtil.UTF_8).length) + .hasMetadata("testMetadata") + .hasData("testData") + .hasNoFragmentsFollow() + .hasRequestN(1) + .typeOf(FrameType.REQUEST_STREAM) + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + Assertions.assertThat(sender.isEmpty()).isTrue(); + + Payload requestedPayload = TestRequesterResponderSupport.randomPayload(allocator); + requestStreamRequesterFlux.handlePayload(requestedPayload); + + Payload unrequestedPayload = TestRequesterResponderSupport.randomPayload(allocator); + requestStreamRequesterFlux.handlePayload(unrequestedPayload); + + final ByteBuf cancelFrame = sender.awaitFrame(); + FrameAssert.assertThat(cancelFrame) + .isNotNull() + .typeOf(FrameType.CANCEL) + .hasClientSideStreamId() + .hasStreamId(1) + .hasNoLeaks(); + + assertSubscriber + .assertValuesWith(p -> PayloadAssert.assertThat(p).isEqualTo(requestedPayload).hasNoLeaks()) + .assertError() + .assertErrorMessage("The number of messages received exceeds the number requested"); + + PayloadAssert.assertThat(requestedPayload).isReleased(); + PayloadAssert.assertThat(unrequestedPayload).isReleased(); + + Assertions.assertThat(payload.refCnt()).isZero(); + activeStreams.assertNoActiveStreams(); + + Assertions.assertThat(sender.isEmpty()).isTrue(); + + // state machine check + stateAssert.isTerminated(); + allocator.assertHasNoLeaks(); + } + @Test public void checkName() { final TestRequesterResponderSupport activeStreams = TestRequesterResponderSupport.client(); diff --git a/rsocket-core/src/test/java/io/rsocket/loadbalance/LoadbalanceTest.java b/rsocket-core/src/test/java/io/rsocket/loadbalance/LoadbalanceTest.java index 5780737cc..fcd3ae4a9 100644 --- a/rsocket-core/src/test/java/io/rsocket/loadbalance/LoadbalanceTest.java +++ b/rsocket-core/src/test/java/io/rsocket/loadbalance/LoadbalanceTest.java @@ -19,6 +19,7 @@ import io.rsocket.RSocket; import io.rsocket.RaceTestConstants; import io.rsocket.core.RSocketConnector; +import io.rsocket.internal.subscriber.AssertSubscriber; import io.rsocket.plugins.RSocketInterceptor; import io.rsocket.test.util.TestClientTransport; import io.rsocket.transport.ClientTransport; @@ -319,6 +320,85 @@ public Flux requestChannel(Publisher source) { Assertions.assertThat(counter.get()).isEqualTo(3); } + @Test + public void shouldNotifyOnCloseWhenAllTheActiveSubscribersAreClosed() { + final AtomicInteger counter = new AtomicInteger(); + final ClientTransport mockTransport = Mockito.mock(ClientTransport.class); + final RSocketConnector rSocketConnectorMock = Mockito.mock(RSocketConnector.class); + + Sinks.Empty onCloseSocket1 = Sinks.empty(); + Sinks.Empty onCloseSocket2 = Sinks.empty(); + + RSocket socket1 = + new RSocket() { + @Override + public Mono onClose() { + return onCloseSocket1.asMono(); + } + + @Override + public Mono fireAndForget(Payload payload) { + return Mono.empty(); + } + }; + RSocket socket2 = + new RSocket() { + @Override + public Mono onClose() { + return onCloseSocket2.asMono(); + } + + @Override + public Mono fireAndForget(Payload payload) { + return Mono.empty(); + } + }; + + Mockito.when(rSocketConnectorMock.connect(Mockito.any(ClientTransport.class))) + .then(im -> Mono.just(socket1)) + .then(im -> Mono.just(socket2)) + .then(im -> Mono.never().doOnCancel(() -> counter.incrementAndGet())); + + final TestPublisher> source = TestPublisher.create(); + final RSocketPool rSocketPool = + new RSocketPool(rSocketConnectorMock, source, new RoundRobinLoadbalanceStrategy()); + + source.next( + Arrays.asList( + LoadbalanceTarget.from("1", mockTransport), + LoadbalanceTarget.from("2", mockTransport), + LoadbalanceTarget.from("3", mockTransport))); + + StepVerifier.create(rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE)) + .expectSubscription() + .expectComplete() + .verify(Duration.ofSeconds(2)); + + StepVerifier.create(rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE)) + .expectSubscription() + .expectComplete() + .verify(Duration.ofSeconds(2)); + + rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE).subscribe(); + + rSocketPool.dispose(); + + AssertSubscriber onCloseSubscriber = + rSocketPool.onClose().subscribeWith(AssertSubscriber.create()); + + onCloseSubscriber.assertNotTerminated(); + + onCloseSocket1.tryEmitEmpty(); + + onCloseSubscriber.assertNotTerminated(); + + onCloseSocket2.tryEmitEmpty(); + + onCloseSubscriber.assertTerminated().assertComplete(); + + Assertions.assertThat(counter.get()).isOne(); + } + static class TestRSocket extends RSocketProxy { final Sinks.Empty sink = Sinks.empty(); diff --git a/rsocket-test/src/main/java/io/rsocket/test/PingClient.java b/rsocket-test/src/main/java/io/rsocket/test/PingClient.java index 9017e854b..14740950a 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/PingClient.java +++ b/rsocket-test/src/main/java/io/rsocket/test/PingClient.java @@ -63,8 +63,8 @@ Flux pingPong( BiFunction> interaction, int count, final Recorder histogram) { - return client - .flatMapMany( + return Flux.usingWhen( + client, rsocket -> Flux.range(1, count) .flatMap( @@ -78,7 +78,11 @@ Flux pingPong( histogram.recordValue(diff); }); }, - 64)) + 64), + rsocket -> { + rsocket.dispose(); + return rsocket.onClose(); + }) .doOnError(Throwable::printStackTrace); } } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java index 85874f44d..901d1ba9a 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java @@ -26,6 +26,7 @@ import java.net.SocketAddress; import java.util.Objects; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.netty.Connection; /** An implementation of {@link DuplexConnection} that connects via TCP. */ @@ -67,24 +68,19 @@ protected void doOnClose() { connection.dispose(); } + @Override + public Mono onClose() { + return super.onClose().and(connection.onDispose()); + } + @Override public void sendErrorAndClose(RSocketErrorException e) { final ByteBuf errorFrame = ErrorFrameCodec.encode(alloc(), 0, e); connection .outbound() .sendObject(FrameLengthCodec.encode(alloc(), errorFrame.readableBytes(), errorFrame)) - .then() - .subscribe( - null, - t -> onClose.tryEmitError(t), - () -> { - final Throwable cause = e.getCause(); - if (cause == null) { - onClose.tryEmitEmpty(); - } else { - onClose.tryEmitError(cause); - } - }); + .subscribe(connection.disposeSubscriber()); + sender.onComplete(); } @Override diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java index 140cfc59f..542ff3599 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java @@ -25,6 +25,7 @@ import java.net.SocketAddress; import java.util.Objects; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.netty.Connection; /** @@ -72,6 +73,11 @@ protected void doOnClose() { connection.dispose(); } + @Override + public Mono onClose() { + return super.onClose().and(connection.onDispose()); + } + @Override public Flux receive() { return connection.inbound().receive(); @@ -83,17 +89,7 @@ public void sendErrorAndClose(RSocketErrorException e) { connection .outbound() .sendObject(new BinaryWebSocketFrame(errorFrame)) - .then() - .subscribe( - null, - t -> onClose.tryEmitError(t), - () -> { - final Throwable cause = e.getCause(); - if (cause == null) { - onClose.tryEmitEmpty(); - } else { - onClose.tryEmitError(cause); - } - }); + .subscribe(connection.disposeSubscriber()); + sender.onComplete(); } } diff --git a/rsocket-transport-netty/src/main/resources/META-INF/native-image/io.rsocket/rsocket-transport-netty/reflect-config.json b/rsocket-transport-netty/src/main/resources/META-INF/native-image/io.rsocket/rsocket-transport-netty/reflect-config.json new file mode 100644 index 000000000..3a2baa440 --- /dev/null +++ b/rsocket-transport-netty/src/main/resources/META-INF/native-image/io.rsocket/rsocket-transport-netty/reflect-config.json @@ -0,0 +1,16 @@ +[ + { + "condition": { + "typeReachable": "io.rsocket.transport.netty.RSocketLengthCodec" + }, + "name": "io.rsocket.transport.netty.RSocketLengthCodec", + "queryAllPublicMethods": true + }, + { + "condition": { + "typeReachable": "io.rsocket.transport.netty.server.BaseWebsocketServerTransport$PongHandler" + }, + "name": "io.rsocket.transport.netty.server.BaseWebsocketServerTransport$PongHandler", + "queryAllPublicMethods": true + } +] \ No newline at end of file