From 8b5800adb37b2599c26c8a8985adb7f15abfed52 Mon Sep 17 00:00:00 2001 From: Robert Roeser Date: Tue, 31 Oct 2017 23:06:14 -0700 Subject: [PATCH] updated switch transform to use unbounded processor to prevent illegal state exception --- .../main/java/io/rsocket/RSocketClient.java | 17 ++- .../main/java/io/rsocket/RSocketServer.java | 15 +-- .../io/rsocket/internal/SwitchTransform.java | 121 +++++++++--------- .../rsocket/internal/UnboundedProcessor.java | 9 +- .../internal/UnboundedProcessorTest.java | 28 ++-- .../io/rsocket/test/BaseClientServerTest.java | 53 ++++---- .../netty/NettyDuplexConnection.java | 17 +-- .../netty/WebsocketDuplexConnection.java | 119 ++++++++--------- 8 files changed, 181 insertions(+), 198 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java index 04be7dac2..2b31f5130 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java @@ -16,6 +16,8 @@ package io.rsocket; +import static io.rsocket.util.ExceptionUtil.noStacktrace; + import io.netty.buffer.Unpooled; import io.netty.util.collection.IntObjectHashMap; import io.rsocket.exceptions.ConnectionException; @@ -23,12 +25,6 @@ import io.rsocket.internal.LimitableRequestPublisher; import io.rsocket.internal.UnboundedProcessor; import io.rsocket.util.PayloadImpl; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import reactor.core.Disposable; -import reactor.core.publisher.*; - -import javax.annotation.Nullable; import java.nio.channels.ClosedChannelException; import java.time.Duration; import java.util.Collection; @@ -37,8 +33,11 @@ import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; - -import static io.rsocket.util.ExceptionUtil.noStacktrace; +import javax.annotation.Nullable; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import reactor.core.Disposable; +import reactor.core.publisher.*; /** Client Side of a RSocket socket. Sends {@link Frame}s to a {@link RSocketServer} */ class RSocketClient implements RSocket { @@ -99,7 +98,7 @@ class RSocketClient implements RSocket { }) .subscribe(); } - + connection .onClose() .doFinally( diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java index 3edec97ac..731f5296d 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java @@ -16,6 +16,10 @@ package io.rsocket; +import static io.rsocket.Frame.Request.initialRequestN; +import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C; +import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M; + import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.util.collection.IntObjectHashMap; @@ -23,6 +27,9 @@ import io.rsocket.internal.LimitableRequestPublisher; import io.rsocket.internal.UnboundedProcessor; import io.rsocket.util.PayloadImpl; +import java.util.Collection; +import java.util.function.Consumer; +import javax.annotation.Nullable; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -32,14 +39,6 @@ import reactor.core.publisher.SignalType; import reactor.core.publisher.UnicastProcessor; -import javax.annotation.Nullable; -import java.util.Collection; -import java.util.function.Consumer; - -import static io.rsocket.Frame.Request.initialRequestN; -import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_C; -import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M; - /** Server side RSocket. Receives {@link Frame}s from a {@link RSocketClient} */ class RSocketServer implements RSocket { diff --git a/rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java b/rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java index 33a02c604..6204669d5 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java @@ -1,81 +1,80 @@ package io.rsocket.internal; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.function.BiFunction; import org.reactivestreams.Publisher; import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; -import reactor.core.publisher.DirectProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.Operators; -import java.util.Objects; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import java.util.function.BiFunction; - public final class SwitchTransform extends Flux { - final Publisher source; - final BiFunction, Publisher> transformer; - - public SwitchTransform(Publisher source, BiFunction, Publisher> transformer) { - this.source = Objects.requireNonNull(source, "source"); - this.transformer = Objects.requireNonNull(transformer, "transformer"); - } - - @Override - public void subscribe(CoreSubscriber actual) { - source.subscribe(new SwitchTransformSubscriber<>(actual, transformer)); - } + final Publisher source; + final BiFunction, Publisher> transformer; - static final class SwitchTransformSubscriber implements CoreSubscriber { - final CoreSubscriber actual; - final BiFunction, Publisher> transformer; - final DirectProcessor processor = DirectProcessor.create(); + public SwitchTransform( + Publisher source, BiFunction, Publisher> transformer) { + this.source = Objects.requireNonNull(source, "source"); + this.transformer = Objects.requireNonNull(transformer, "transformer"); + } - Subscription s; + @Override + public void subscribe(CoreSubscriber actual) { + Flux.from(source).subscribe(new SwitchTransformSubscriber<>(actual, transformer)); + } - volatile int once; - @SuppressWarnings("rawtypes") - static final AtomicIntegerFieldUpdater ONCE = - AtomicIntegerFieldUpdater.newUpdater(SwitchTransformSubscriber.class, "once"); + static final class SwitchTransformSubscriber implements CoreSubscriber { + @SuppressWarnings("rawtypes") + static final AtomicIntegerFieldUpdater ONCE = + AtomicIntegerFieldUpdater.newUpdater(SwitchTransformSubscriber.class, "once"); - SwitchTransformSubscriber(CoreSubscriber actual, BiFunction, Publisher> transformer) { - this.actual = actual; - this.transformer = transformer; - } + final CoreSubscriber actual; + final BiFunction, Publisher> transformer; + final UnboundedProcessor processor = new UnboundedProcessor<>(); + Subscription s; + volatile int once; - @Override - public void onSubscribe(Subscription s) { - if (Operators.validate(this.s, s)) { - this.s = s; + SwitchTransformSubscriber( + CoreSubscriber actual, + BiFunction, Publisher> transformer) { + this.actual = actual; + this.transformer = transformer; + } - processor.onSubscribe(s); - } - } + @Override + public void onSubscribe(Subscription s) { + if (Operators.validate(this.s, s)) { + this.s = s; + processor.onSubscribe(s); + } + } - @Override - public void onNext(T t) { - if (once == 0 && ONCE.compareAndSet(this, 0, 1)) { - try { - Publisher result = Objects.requireNonNull(transformer.apply(t, processor), - "The transformer returned a null value"); - result.subscribe(actual); - } - catch (Throwable e) { - onError(Operators.onOperatorError(s, e, t, actual.currentContext())); - return; - } - } - processor.onNext(t); - } + @Override + public void onNext(T t) { + if (once == 0 && ONCE.compareAndSet(this, 0, 1)) { + try { + Publisher result = + Objects.requireNonNull( + transformer.apply(t, processor), "The transformer returned a null value"); + Flux.from(result).subscribe(actual); + } catch (Throwable e) { + onError(Operators.onOperatorError(s, e, t, actual.currentContext())); + return; + } + } + processor.onNext(t); + } - @Override - public void onError(Throwable t) { - processor.onError(t); - } + @Override + public void onError(Throwable t) { + processor.onError(t); + } - @Override - public void onComplete() { - processor.onComplete(); - } - } + @Override + public void onComplete() { + processor.onComplete(); + } + } } diff --git a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java index 595e69927..a278fc5e0 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java @@ -16,6 +16,10 @@ package io.rsocket.internal; import io.netty.util.internal.shaded.org.jctools.queues.atomic.MpscGrowableAtomicArrayQueue; +import java.util.Objects; +import java.util.Queue; +import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.concurrent.atomic.AtomicLongFieldUpdater; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; @@ -27,11 +31,6 @@ import reactor.util.concurrent.Queues; import reactor.util.context.Context; -import java.util.Objects; -import java.util.Queue; -import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; - /** * A Processor implementation that takes a custom queue and allows only a single subscriber. * diff --git a/rsocket-core/src/test/java/io/rsocket/internal/UnboundedProcessorTest.java b/rsocket-core/src/test/java/io/rsocket/internal/UnboundedProcessorTest.java index 88f54b933..9a9699178 100644 --- a/rsocket-core/src/test/java/io/rsocket/internal/UnboundedProcessorTest.java +++ b/rsocket-core/src/test/java/io/rsocket/internal/UnboundedProcessorTest.java @@ -1,10 +1,9 @@ package io.rsocket.internal; +import java.util.concurrent.CountDownLatch; import org.junit.Assert; import org.junit.Test; -import java.util.concurrent.CountDownLatch; - public class UnboundedProcessorTest { @Test public void testOnNextBeforeSubscribe_10() { @@ -49,39 +48,34 @@ public void testOnNextBeforeSubscribeN(int n) { Assert.assertEquals(n, count); } - + @Test public void testOnNextAfterSubscribe_10() throws Exception { testOnNextAfterSubscribeN(10); } - + @Test public void testOnNextAfterSubscribe_100() throws Exception { testOnNextAfterSubscribeN(100); } - + @Test public void testOnNextAfterSubscribe_1000() throws Exception { testOnNextAfterSubscribeN(1000); } - - public void testOnNextAfterSubscribeN(int n) throws Exception { + + public void testOnNextAfterSubscribeN(int n) throws Exception { CountDownLatch latch = new CountDownLatch(n); - UnboundedProcessor processor = new UnboundedProcessor<>(); - processor - .log() - .doOnNext(integer -> - latch.countDown()) - .subscribe(); - + UnboundedProcessor processor = new UnboundedProcessor<>(); + processor.log().doOnNext(integer -> latch.countDown()).subscribe(); + for (int i = 0; i < n; i++) { System.out.println("onNexting -> " + i); processor.onNext(i); } - + processor.drain(); - + latch.await(); } - } diff --git a/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java b/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java index ba08661f3..7a1d79e0e 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java +++ b/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java @@ -16,17 +16,15 @@ package io.rsocket.test; +import static org.junit.Assert.assertEquals; + import io.rsocket.Payload; import io.rsocket.util.PayloadImpl; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import reactor.core.publisher.Flux; -import reactor.core.scheduler.Schedulers; - -import java.util.concurrent.atomic.AtomicInteger; - -import static org.junit.Assert.assertEquals; public abstract class BaseClientServerTest> { @Rule public final T setup = createClientServer(); @@ -57,7 +55,7 @@ public void testPushMetadata10() { assertEquals(0, outputCount); } - @Test//(timeout = 10000) + @Test // (timeout = 10000) public void testRequestResponse1() { long outputCount = Flux.range(1, 1) @@ -124,22 +122,22 @@ public void testRequestResponse10_000() { assertEquals(10_000, outputCount); } - + @Test(timeout = 10000) public void testRequestStream() { Flux publisher = setup.getRSocket().requestStream(testPayload(3)); - + long count = publisher.take(5).count().block(); - + assertEquals(5, count); } - + @Test(timeout = 10000) public void testRequestStreamAll() { Flux publisher = setup.getRSocket().requestStream(testPayload(3)); - + long count = publisher.count().block(); - + assertEquals(10000, count); } @@ -207,48 +205,41 @@ public void testChannel3() { assertEquals(3, count); } - + @Test(timeout = 10000) public void testChannel512() { Flux payloads = Flux.range(1, 512).map(i -> new PayloadImpl("hello " + i)); - + long count = setup.getRSocket().requestChannel(payloads).count().block(); - + assertEquals(512, count); } - + @Test(timeout = 30000) public void testChannel20_000() { Flux payloads = Flux.range(1, 20_000).map(i -> new PayloadImpl("hello " + i)); - + long count = setup.getRSocket().requestChannel(payloads).count().block(); - + assertEquals(20_000, count); } - + @Test(timeout = 60_000) public void testChannel200_000() { Flux payloads = Flux.range(1, 200_000).map(i -> new PayloadImpl("hello " + i)); - + long count = setup.getRSocket().requestChannel(payloads).count().block(); - + assertEquals(200_000, count); } - + @Test(timeout = 60_000) @Ignore public void testChannel2_000_000() { AtomicInteger counter = new AtomicInteger(0); - Flux payloads = - Flux.range(1, 2_000_000) - .map(i -> new PayloadImpl("hello " + i)); - long count = - setup - .getRSocket() - .requestChannel(payloads) - .count() - .block(); + Flux payloads = Flux.range(1, 2_000_000).map(i -> new PayloadImpl("hello " + i)); + long count = setup.getRSocket().requestChannel(payloads).count().block(); assertEquals(2_000_000, count); } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/NettyDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/NettyDuplexConnection.java index 0132253f9..bed01967b 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/NettyDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/NettyDuplexConnection.java @@ -14,6 +14,7 @@ * limitations under the License. */ package io.rsocket.transport.netty; + import io.rsocket.DuplexConnection; import io.rsocket.Frame; import org.reactivestreams.Publisher; @@ -29,13 +30,13 @@ public class NettyDuplexConnection implements DuplexConnection { private final NettyOutbound out; private final NettyContext context; private final MonoProcessor onClose; - + public NettyDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) { this.in = in; this.out = out; this.context = context; this.onClose = MonoProcessor.create(); - + context.onClose(onClose::onComplete); this.onClose .doFinally( @@ -45,32 +46,32 @@ public NettyDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext co }) .subscribe(); } - + @Override public Mono send(Publisher frames) { return Flux.from(frames).concatMap(this::sendOne).then(); } - + @Override public Mono sendOne(Frame frame) { return out.sendObject(frame.content()).then(); } - + @Override public Flux receive() { return in.receive().map(buf -> Frame.from(buf.retain())); } - + @Override public Mono close() { return Mono.fromRunnable(onClose::onComplete); } - + @Override public Mono onClose() { return onClose; } - + @Override public double availability() { return onClose.isTerminated() ? 0.0 : 1.0; 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 540f312a2..838fadf93 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 @@ -14,6 +14,7 @@ * limitations under the License. */ package io.rsocket.transport.netty; + import static io.netty.buffer.Unpooled.wrappedBuffer; import static io.rsocket.frame.FrameHeaderFlyweight.FRAME_LENGTH_SIZE; @@ -39,63 +40,63 @@ * back on for frames received. */ public class WebsocketDuplexConnection implements DuplexConnection { - private final NettyInbound in; - private final NettyOutbound out; - private final NettyContext context; - private final MonoProcessor onClose; - - public WebsocketDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) { - this.in = in; - this.out = out; - this.context = context; - this.onClose = MonoProcessor.create(); - - context.onClose(onClose::onComplete); - this.onClose - .doFinally( - s -> { - this.context.dispose(); - this.context.channel().close(); - }) - .subscribe(); - } - - @Override - public Mono send(Publisher frames) { - return Flux.from(frames).concatMap(this::sendOne).then(); - } - - @Override - public Mono sendOne(Frame frame) { - return out.sendObject(new BinaryWebSocketFrame(frame.content().skipBytes(FRAME_LENGTH_SIZE))) - .then(); - } - - @Override - public Flux receive() { - return in.receive() - .map( - buf -> { - CompositeByteBuf composite = context.channel().alloc().compositeBuffer(); - ByteBuf length = wrappedBuffer(new byte[FRAME_LENGTH_SIZE]); - FrameHeaderFlyweight.encodeLength(length, 0, buf.readableBytes()); - composite.addComponents(true, length, buf.retain()); - return Frame.from(composite); - }); - } - - @Override - public Mono close() { - return Mono.fromRunnable(onClose::onComplete); - } - - @Override - public Mono onClose() { - return onClose; - } - - @Override - public double availability() { - return onClose.isTerminated() ? 0.0 : 1.0; - } + private final NettyInbound in; + private final NettyOutbound out; + private final NettyContext context; + private final MonoProcessor onClose; + + public WebsocketDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) { + this.in = in; + this.out = out; + this.context = context; + this.onClose = MonoProcessor.create(); + + context.onClose(onClose::onComplete); + this.onClose + .doFinally( + s -> { + this.context.dispose(); + this.context.channel().close(); + }) + .subscribe(); + } + + @Override + public Mono send(Publisher frames) { + return Flux.from(frames).concatMap(this::sendOne).then(); + } + + @Override + public Mono sendOne(Frame frame) { + return out.sendObject(new BinaryWebSocketFrame(frame.content().skipBytes(FRAME_LENGTH_SIZE))) + .then(); + } + + @Override + public Flux receive() { + return in.receive() + .map( + buf -> { + CompositeByteBuf composite = context.channel().alloc().compositeBuffer(); + ByteBuf length = wrappedBuffer(new byte[FRAME_LENGTH_SIZE]); + FrameHeaderFlyweight.encodeLength(length, 0, buf.readableBytes()); + composite.addComponents(true, length, buf.retain()); + return Frame.from(composite); + }); + } + + @Override + public Mono close() { + return Mono.fromRunnable(onClose::onComplete); + } + + @Override + public Mono onClose() { + return onClose; + } + + @Override + public double availability() { + return onClose.isTerminated() ? 0.0 : 1.0; + } }