diff --git a/README.md b/README.md index 51f02fc2d..2ab9b9935 100644 --- a/README.md +++ b/README.md @@ -61,7 +61,7 @@ import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.RSocketFactory; import io.rsocket.transport.netty.client.WebsocketClientTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import reactor.core.publisher.Flux; import java.net.URI; @@ -72,7 +72,7 @@ public class ExampleClient { RSocket client = RSocketFactory.connect().keepAlive().transport(ws).start().block(); try { - Flux s = client.requestStream(PayloadImpl.textPayload("peace")); + Flux s = client.requestStream(DefaultPayload.textPayload("peace")); s.take(10).doOnNext(p -> System.out.println(p.getDataUtf8())).blockLast(); } finally { diff --git a/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java b/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java index 22ee60095..9d1510682 100644 --- a/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java +++ b/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java @@ -2,7 +2,7 @@ import io.rsocket.fragmentation.FrameFragmenter; import io.rsocket.fragmentation.FrameReassembler; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.nio.ByteBuffer; import java.util.concurrent.ThreadLocalRandom; import java.util.stream.Collectors; @@ -43,13 +43,13 @@ public void setup(Blackhole bh) { ByteBuffer data = createRandomBytes(1 << 18); ByteBuffer metadata = createRandomBytes(1 << 18); largeFrame = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); largeFrameFragmenter = new FrameFragmenter(1024); data = createRandomBytes(16); metadata = createRandomBytes(16); smallFrame = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); smallFrameFragmenter = new FrameFragmenter(2); smallFramesIterable = smallFrameFragmenter diff --git a/rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java b/rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java index 215826e8e..39b8bf44d 100644 --- a/rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java +++ b/rsocket-core/src/jmh/java/io/rsocket/RSocketPerf.java @@ -18,7 +18,8 @@ import io.rsocket.RSocketFactory.Start; import io.rsocket.perfutil.TestDuplexConnection; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import org.openjdk.jmh.annotations.Benchmark; @@ -79,7 +80,7 @@ public static class Input { static final ByteBuffer HELLO = ByteBuffer.wrap("HELLO".getBytes(StandardCharsets.UTF_8)); - static final Payload HELLO_PAYLOAD = new PayloadImpl(HELLO); + static final Payload HELLO_PAYLOAD = new DefaultPayload(HELLO); static final DirectProcessor clientReceive = DirectProcessor.create(); static final DirectProcessor serverReceive = DirectProcessor.create(); diff --git a/rsocket-core/src/main/java/io/rsocket/ConnectionSetupPayload.java b/rsocket-core/src/main/java/io/rsocket/ConnectionSetupPayload.java index fa4f7ff71..35feedbcf 100644 --- a/rsocket-core/src/main/java/io/rsocket/ConnectionSetupPayload.java +++ b/rsocket-core/src/main/java/io/rsocket/ConnectionSetupPayload.java @@ -17,47 +17,50 @@ import static io.rsocket.frame.FrameHeaderFlyweight.FLAGS_M; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.netty.util.AbstractReferenceCounted; +import io.rsocket.Frame.Setup; import io.rsocket.frame.SetupFrameFlyweight; -import java.nio.ByteBuffer; /** * Exposed to server for determination of RequestHandler based on mime types and SETUP metadata/data */ -public abstract class ConnectionSetupPayload implements Payload { +public abstract class ConnectionSetupPayload extends AbstractReferenceCounted implements Payload { public static final int NO_FLAGS = 0; public static final int HONOR_LEASE = SetupFrameFlyweight.FLAGS_WILL_HONOR_LEASE; public static final int STRICT_INTERPRETATION = SetupFrameFlyweight.FLAGS_STRICT_INTERPRETATION; public static ConnectionSetupPayload create(String metadataMimeType, String dataMimeType) { - return new ConnectionSetupPayloadImpl( - metadataMimeType, dataMimeType, Frame.NULL_BYTEBUFFER, Frame.NULL_BYTEBUFFER, NO_FLAGS); + return new DefaultConnectionSetupPayload( + metadataMimeType, dataMimeType, Unpooled.EMPTY_BUFFER, Unpooled.EMPTY_BUFFER, NO_FLAGS); } public static ConnectionSetupPayload create( String metadataMimeType, String dataMimeType, Payload payload) { - return new ConnectionSetupPayloadImpl( + return new DefaultConnectionSetupPayload( metadataMimeType, dataMimeType, - payload.getData(), - payload.getMetadata(), + payload.sliceData(), + payload.sliceMetadata(), payload.hasMetadata() ? FLAGS_M : 0); } public static ConnectionSetupPayload create( String metadataMimeType, String dataMimeType, int flags) { - return new ConnectionSetupPayloadImpl( - metadataMimeType, dataMimeType, Frame.NULL_BYTEBUFFER, Frame.NULL_BYTEBUFFER, flags); + return new DefaultConnectionSetupPayload( + metadataMimeType, dataMimeType, Unpooled.EMPTY_BUFFER, Unpooled.EMPTY_BUFFER, flags); } public static ConnectionSetupPayload create(final Frame setupFrame) { Frame.ensureFrameType(FrameType.SETUP, setupFrame); - return new ConnectionSetupPayloadImpl( - Frame.Setup.metadataMimeType(setupFrame), - Frame.Setup.dataMimeType(setupFrame), - setupFrame.getData(), - setupFrame.getMetadata(), - Frame.Setup.getFlags(setupFrame)); + return new DefaultConnectionSetupPayload( + Setup.metadataMimeType(setupFrame), + Setup.dataMimeType(setupFrame), + setupFrame.sliceData(), + setupFrame.sliceMetadata(), + Setup.getFlags(setupFrame)); } public abstract String metadataMimeType(); @@ -79,19 +82,34 @@ public boolean hasMetadata() { return Frame.isFlagSet(getFlags(), FLAGS_M); } - private static final class ConnectionSetupPayloadImpl extends ConnectionSetupPayload { + @Override + public ConnectionSetupPayload retain() { + super.retain(); + return this; + } + + @Override + public ConnectionSetupPayload retain(int increment) { + super.retain(increment); + return this; + } + + public abstract ConnectionSetupPayload touch(); + public abstract ConnectionSetupPayload touch(Object hint); + + private static final class DefaultConnectionSetupPayload extends ConnectionSetupPayload { private final String metadataMimeType; private final String dataMimeType; - private final ByteBuffer data; - private final ByteBuffer metadata; + private final ByteBuf data; + private final ByteBuf metadata; private final int flags; - public ConnectionSetupPayloadImpl( + public DefaultConnectionSetupPayload( String metadataMimeType, String dataMimeType, - ByteBuffer data, - ByteBuffer metadata, + ByteBuf data, + ByteBuf metadata, int flags) { this.metadataMimeType = metadataMimeType; this.dataMimeType = dataMimeType; @@ -99,7 +117,7 @@ public ConnectionSetupPayloadImpl( this.metadata = metadata; this.flags = flags; - if (!hasMetadata() && metadata.remaining() > 0) { + if (!hasMetadata() && metadata.readableBytes() > 0) { throw new IllegalArgumentException("metadata flag incorrect"); } } @@ -115,12 +133,12 @@ public String dataMimeType() { } @Override - public ByteBuffer getData() { + public ByteBuf sliceData() { return data; } @Override - public ByteBuffer getMetadata() { + public ByteBuf sliceMetadata() { return metadata; } @@ -128,5 +146,25 @@ public ByteBuffer getMetadata() { public int getFlags() { return flags; } + + @Override + public ConnectionSetupPayload touch() { + data.touch(); + metadata.touch(); + return this; + } + + @Override + public ConnectionSetupPayload touch(Object hint) { + data.touch(hint); + metadata.touch(hint); + return this; + } + + @Override + protected void deallocate() { + data.release(); + metadata.release(); + } } } diff --git a/rsocket-core/src/main/java/io/rsocket/Frame.java b/rsocket-core/src/main/java/io/rsocket/Frame.java index 1642e942d..7dc8572e2 100644 --- a/rsocket-core/src/main/java/io/rsocket/Frame.java +++ b/rsocket-core/src/main/java/io/rsocket/Frame.java @@ -30,7 +30,6 @@ import io.rsocket.frame.RequestNFrameFlyweight; import io.rsocket.frame.SetupFrameFlyweight; import io.rsocket.frame.VersionFlyweight; -import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import javax.annotation.Nullable; import org.slf4j.Logger; @@ -41,9 +40,7 @@ * *

This provides encoding, decoding and field accessors. */ -public class Frame implements ByteBufHolder { - public static final ByteBuffer NULL_BYTEBUFFER = ByteBuffer.allocateDirect(0); - +public class Frame implements Payload, ByteBufHolder { private static final Recycler RECYCLER = new Recycler() { protected Frame newObject(Handle handle) { @@ -52,7 +49,7 @@ protected Frame newObject(Handle handle) { }; private final Handle handle; - private @Nullable ByteBuf content; + private ByteBuf content; private Frame(final Handle handle) { this.handle = handle; @@ -183,43 +180,25 @@ public boolean release(int decrement) { } /** - * Return {@link ByteBuffer} that is a {@link ByteBuffer#slice()} for the frame metadata + * Return {@link ByteBuf} that is a {@link ByteBuf#slice()} for the frame metadata * - *

If no metadata is present, the ByteBuffer will have 0 capacity. + *

If no metadata is present, the ByteBuf will have 0 capacity. * - * @return ByteBuffer containing the content + * @return ByteBuf containing the content */ - public ByteBuffer getMetadata() { - final ByteBuf metadata = FrameHeaderFlyweight.sliceFrameMetadata(content); - if (metadata == null) { - return NULL_BYTEBUFFER; - } else if (metadata.readableBytes() > 0) { - final ByteBuffer buffer = ByteBuffer.allocateDirect(metadata.readableBytes()); - metadata.readBytes(buffer); - buffer.flip(); - return buffer; - } else { - return NULL_BYTEBUFFER; - } + public ByteBuf sliceMetadata() { + return hasMetadata() ? FrameHeaderFlyweight.sliceFrameMetadata(content) : Unpooled.EMPTY_BUFFER; } /** - * Return {@link ByteBuffer} that is a {@link ByteBuffer#slice()} for the frame data + * Return {@link ByteBuf} that is a {@link ByteBuf#slice()} for the frame data * - *

If no data is present, the ByteBuffer will have 0 capacity. + *

If no data is present, the ByteBuf will have 0 capacity. * - * @return ByteBuffer containing the data + * @return ByteBuf containing the data */ - public ByteBuffer getData() { - final ByteBuf data = FrameHeaderFlyweight.sliceFrameData(content); - if (data.readableBytes() > 0) { - final ByteBuffer buffer = ByteBuffer.allocateDirect(data.readableBytes()); - data.readBytes(buffer); - buffer.flip(); - return buffer; - } else { - return NULL_BYTEBUFFER; - } + public ByteBuf sliceData() { + return FrameHeaderFlyweight.sliceFrameData(content); } /** @@ -270,14 +249,11 @@ public static int setFlag(int current, int toSet) { return current | toSet; } + @Override public boolean hasMetadata() { return Frame.isFlagSet(this.flags(), FLAGS_M); } - public String getDataUtf8() { - return StandardCharsets.UTF_8.decode(getData()).toString(); - } - /* TODO: * * fromRequest(type, id, payload) @@ -297,14 +273,8 @@ public static Frame from( String metadataMimeType, String dataMimeType, Payload payload) { - final ByteBuf metadata = - payload.hasMetadata() - ? Unpooled.wrappedBuffer(payload.getMetadata()) - : Unpooled.EMPTY_BUFFER; - final ByteBuf data = - payload.getData() != null - ? Unpooled.wrappedBuffer(payload.getData()) - : Unpooled.EMPTY_BUFFER; + final ByteBuf metadata = payload.hasMetadata() ? payload.sliceMetadata() : Unpooled.EMPTY_BUFFER; + final ByteBuf data = payload.sliceData(); final Frame frame = RECYCLER.get(); frame.content = @@ -460,9 +430,8 @@ public static Frame from(int streamId, FrameType type, Payload payload, int init if (initialRequestN < 1) { throw new IllegalStateException("initial request n must be greater than 0"); } - final @Nullable ByteBuf metadata = - payload.hasMetadata() ? Unpooled.wrappedBuffer(payload.getMetadata()) : null; - final ByteBuf data = Unpooled.wrappedBuffer(payload.getData()); + final @Nullable ByteBuf metadata = payload.hasMetadata() ? payload.sliceMetadata() : null; + final ByteBuf data = payload.sliceData(); final Frame frame = RECYCLER.get(); frame.content = @@ -561,9 +530,8 @@ public static Frame from(int streamId, FrameType type, Payload payload) { } public static Frame from(int streamId, FrameType type, Payload payload, int flags) { - final ByteBuf metadata = - payload.hasMetadata() ? Unpooled.wrappedBuffer(payload.getMetadata()) : null; - final ByteBuf data = Unpooled.wrappedBuffer(payload.getData()); + final ByteBuf metadata = payload.hasMetadata() ? payload.sliceMetadata() : null; + final ByteBuf data = payload.sliceData(); return from(streamId, type, metadata, data, flags); } diff --git a/rsocket-core/src/main/java/io/rsocket/Payload.java b/rsocket-core/src/main/java/io/rsocket/Payload.java index 00e0e360b..0b8f5ee8a 100644 --- a/rsocket-core/src/main/java/io/rsocket/Payload.java +++ b/rsocket-core/src/main/java/io/rsocket/Payload.java @@ -15,11 +15,16 @@ */ package io.rsocket; +import io.netty.buffer.ByteBuf; + import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import io.netty.util.ReferenceCounted; +import io.netty.util.ResourceLeakDetector; + /** Payload of a {@link Frame}. */ -public interface Payload { +public interface Payload extends ReferenceCounted { /** * Returns whether the payload has metadata, useful for tell if metadata is empty or not present. * @@ -33,20 +38,56 @@ public interface Payload { * * @return payload metadata. */ - ByteBuffer getMetadata(); + ByteBuf sliceMetadata(); /** * Returns the Payload data. Always non-null. * * @return payload data. */ - ByteBuffer getData(); + ByteBuf sliceData(); + + /** + * Increases the reference count by {@code 1}. + */ + @Override + Payload retain(); + + /** + * Increases the reference count by the specified {@code increment}. + */ + @Override + Payload retain(int increment); + + /** + * Records the current access location of this object for debugging purposes. + * If this object is determined to be leaked, the information recorded by this operation will be provided to you + * via {@link ResourceLeakDetector}. This method is a shortcut to {@link #touch(Object) touch(null)}. + */ + @Override + Payload touch(); + + /** + * Records the current access location of this object with an additional arbitrary information for debugging + * purposes. If this object is determined to be leaked, the information recorded by this operation will be + * provided to you via {@link ResourceLeakDetector}. + */ + @Override + Payload touch(Object hint); + + default ByteBuffer getMetadata() { + return sliceMetadata().nioBuffer(); + } + + default ByteBuffer getData() { + return sliceData().nioBuffer(); + } default String getMetadataUtf8() { - return StandardCharsets.UTF_8.decode(getMetadata()).toString(); + return sliceMetadata().toString(StandardCharsets.UTF_8); } default String getDataUtf8() { - return StandardCharsets.UTF_8.decode(getData()).toString(); + return sliceData().toString(StandardCharsets.UTF_8); } -} +} \ No newline at end of file diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java index a8b3e0b50..13eb75952 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java @@ -24,7 +24,7 @@ import io.rsocket.exceptions.Exceptions; import io.rsocket.internal.LimitableRequestPublisher; import io.rsocket.internal.UnboundedProcessor; -import io.rsocket.util.PayloadImpl; + import java.nio.channels.ClosedChannelException; import java.time.Duration; import java.util.Collection; @@ -46,6 +46,7 @@ class RSocketClient implements RSocket { noStacktrace(new ClosedChannelException()); private final DuplexConnection connection; + private final Function frameDecoder; private final Consumer errorConsumer; private final StreamIdSupplier streamIdSupplier; private final MonoProcessor started; @@ -60,19 +61,22 @@ class RSocketClient implements RSocket { RSocketClient( DuplexConnection connection, + Function frameDecoder, Consumer errorConsumer, StreamIdSupplier streamIdSupplier) { - this(connection, errorConsumer, streamIdSupplier, Duration.ZERO, Duration.ZERO, 0); + this(connection, frameDecoder, errorConsumer, streamIdSupplier, Duration.ZERO, Duration.ZERO, 0); } RSocketClient( DuplexConnection connection, + Function frameDecoder, Consumer errorConsumer, StreamIdSupplier streamIdSupplier, Duration tickPeriod, Duration ackTimeout, int missedAcks) { this.connection = connection; + this.frameDecoder = frameDecoder; this.errorConsumer = errorConsumer; this.streamIdSupplier = streamIdSupplier; this.started = MonoProcessor.create(); @@ -194,6 +198,7 @@ public Mono fireAndForget(Payload payload) { final int streamId = streamIdSupplier.nextStreamId(); final Frame requestFrame = Frame.Request.from(streamId, FrameType.FIRE_AND_FORGET, payload, 1); + payload.release(); sendProcessor.onNext(requestFrame); }); @@ -218,6 +223,7 @@ public Flux requestChannel(Publisher payloads) { @Override public Mono metadataPush(Payload payload) { final Frame requestFrame = Frame.Request.from(0, FrameType.METADATA_PUSH, payload, 1); + payload.release(); sendProcessor.onNext(requestFrame); return Mono.empty(); } @@ -257,7 +263,7 @@ public Flux handleRequestStream(final Payload payload) { if (first.compareAndSet(false, true) && !receiver.isTerminated()) { final Frame requestFrame = Frame.Request.from(streamId, FrameType.REQUEST_STREAM, payload, l); - + payload.release(); sendProcessor.onNext(requestFrame); } else if (contains(streamId) && !receiver.isTerminated()) { sendProcessor.onNext(Frame.RequestN.from(streamId, l)); @@ -290,6 +296,7 @@ private Mono handleRequestResponse(final Payload payload) { int streamId = streamIdSupplier.nextStreamId(); final Frame requestFrame = Frame.Request.from(streamId, FrameType.REQUEST_RESPONSE, payload, 1); + payload.release(); MonoProcessor receiver = MonoProcessor.create(); @@ -363,13 +370,16 @@ public Flux get() { @Override public Frame apply(Payload payload) { + final Frame requestFrame; if (firstPayload.compareAndSet(true, false)) { - return Frame.Request.from( + requestFrame = Frame.Request.from( streamId, requestType, payload, l); } else { - return Frame.PayloadFrame.from( + requestFrame = Frame.PayloadFrame.from( streamId, FrameType.NEXT, payload); } + payload.release(); + return requestFrame; } }) .doOnComplete( @@ -504,7 +514,7 @@ private void handleFrame(int streamId, FrameType type, Frame frame) { removeReceiver(streamId); break; case NEXT_COMPLETE: - receiver.onNext(new PayloadImpl(frame)); + receiver.onNext(frameDecoder.apply(frame)); receiver.onComplete(); break; case CANCEL: @@ -520,7 +530,7 @@ private void handleFrame(int streamId, FrameType type, Frame frame) { break; } case NEXT: - receiver.onNext(new PayloadImpl(frame)); + receiver.onNext(frameDecoder.apply(frame)); break; case REQUEST_N: { diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java b/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java index 21553f228..2bd3d842b 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketFactory.java @@ -27,11 +27,13 @@ import io.rsocket.plugins.RSocketInterceptor; import io.rsocket.transport.ClientTransport; import io.rsocket.transport.ServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.time.Duration; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; + +import io.rsocket.util.EmptyPayload; import reactor.core.publisher.Mono; /** Factory for creating RSocket clients and servers. */ @@ -58,18 +60,6 @@ public interface Start { Mono start(); } - public interface SetupPayload { - T setupPayload(Payload payload); - } - - public interface Acceptor { - T acceptor(Supplier acceptor); - - default T acceptor(A acceptor) { - return acceptor(() -> acceptor); - } - } - public interface ClientTransportAcceptor { Start transport(Supplier transport); @@ -86,43 +76,7 @@ default Start transport(ServerTransport transport) { } } - public interface Fragmentation { - T fragment(int mtu); - } - - public interface ErrorConsumer { - T errorConsumer(Consumer errorConsumer); - } - - public interface KeepAlive { - T keepAlive(); - - T keepAlive(Duration tickPeriod, Duration ackTimeout, int missedAcks); - - T keepAliveTickPeriod(Duration tickPeriod); - - T keepAliveAckTimeout(Duration ackTimeout); - - T keepAliveMissedAcks(int missedAcks); - } - - public interface MimeType { - T mimeType(String metadataMimeType, String dataMimeType); - - T dataMimeType(String dataMimeType); - - T metadataMimeType(String metadataMimeType); - } - - public static class ClientRSocketFactory - implements Acceptor>, - ClientTransportAcceptor, - KeepAlive, - MimeType, - Fragmentation, - ErrorConsumer, - SetupPayload { - + public static class ClientRSocketFactory implements ClientTransportAcceptor { private Supplier> acceptor = () -> rSocket -> new AbstractRSocket() {}; @@ -131,7 +85,8 @@ public static class ClientRSocketFactory private PluginRegistry plugins = new PluginRegistry(Plugins.defaultPlugins()); private int flags = SetupFrameFlyweight.FLAGS_STRICT_INTERPRETATION; - private Payload setupPayload = PayloadImpl.EMPTY; + private Payload setupPayload = EmptyPayload.INSTANCE; + private Function frameDecoder = DefaultPayload::create; private Duration tickPeriod = Duration.ZERO; private Duration ackTimeout = Duration.ofSeconds(30); @@ -155,13 +110,11 @@ public ClientRSocketFactory addServerPlugin(RSocketInterceptor interceptor) { return this; } - @Override public ClientRSocketFactory keepAlive() { tickPeriod = Duration.ofSeconds(20); return this; } - @Override public ClientRSocketFactory keepAlive( Duration tickPeriod, Duration ackTimeout, int missedAcks) { this.tickPeriod = tickPeriod; @@ -170,38 +123,32 @@ public ClientRSocketFactory keepAlive( return this; } - @Override public ClientRSocketFactory keepAliveTickPeriod(Duration tickPeriod) { this.tickPeriod = tickPeriod; return this; } - @Override public ClientRSocketFactory keepAliveAckTimeout(Duration ackTimeout) { this.ackTimeout = ackTimeout; return this; } - @Override public ClientRSocketFactory keepAliveMissedAcks(int missedAcks) { this.missedAcks = missedAcks; return this; } - @Override public ClientRSocketFactory mimeType(String metadataMimeType, String dataMimeType) { this.dataMimeType = dataMimeType; this.metadataMimeType = metadataMimeType; return this; } - @Override public ClientRSocketFactory dataMimeType(String dataMimeType) { this.dataMimeType = dataMimeType; return this; } - @Override public ClientRSocketFactory metadataMimeType(String metadataMimeType) { this.metadataMimeType = metadataMimeType; return this; @@ -212,31 +159,37 @@ public Start transport(Supplier transportClient) { return new StartClient(transportClient); } - @Override + public ClientTransportAcceptor acceptor(Function acceptor) { + this.acceptor = () -> acceptor; + return StartClient::new; + } + public ClientTransportAcceptor acceptor(Supplier> acceptor) { this.acceptor = acceptor; return StartClient::new; } - @Override public ClientRSocketFactory fragment(int mtu) { this.mtu = mtu; return this; } - @Override public ClientRSocketFactory errorConsumer(Consumer errorConsumer) { this.errorConsumer = errorConsumer; return this; } - @Override public ClientRSocketFactory setupPayload(Payload payload) { this.setupPayload = payload; return this; } - protected class StartClient implements Start { + public ClientRSocketFactory frameDecoder(Function frameDecoder) { + this.frameDecoder = frameDecoder; + return this; + } + + private class StartClient implements Start { private final Supplier transportClient; StartClient(Supplier transportClient) { @@ -269,6 +222,7 @@ public Mono start() { RSocketClient rSocketClient = new RSocketClient( multiplexer.asClientConnection(), + frameDecoder, errorConsumer, StreamIdSupplier.clientSupplier(), tickPeriod, @@ -290,7 +244,7 @@ public Mono start() { .doOnNext( rSocket -> new RSocketServer( - multiplexer.asServerConnection(), rSocket, errorConsumer)) + multiplexer.asServerConnection(), rSocket, frameDecoder, errorConsumer)) .then(finalConnection.sendOne(setupFrame)) .then(wrappedRSocketClient); }); @@ -299,12 +253,9 @@ public Mono start() { } } - public static class ServerRSocketFactory - implements Acceptor, - Fragmentation, - ErrorConsumer { - + public static class ServerRSocketFactory { private Supplier acceptor; + private Function frameDecoder = DefaultPayload::create; private Consumer errorConsumer = Throwable::printStackTrace; private int mtu = 0; private PluginRegistry plugins = new PluginRegistry(Plugins.defaultPlugins()); @@ -326,19 +277,26 @@ public ServerRSocketFactory addServerPlugin(RSocketInterceptor interceptor) { return this; } - @Override + public ServerTransportAcceptor acceptor(SocketAcceptor acceptor) { + this.acceptor = () -> acceptor; + return ServerStart::new; + } + public ServerTransportAcceptor acceptor(Supplier acceptor) { this.acceptor = acceptor; return ServerStart::new; } - @Override + public ServerRSocketFactory frameDecoder(Function frameDecoder) { + this.frameDecoder = frameDecoder; + return this; + } + public ServerRSocketFactory fragment(int mtu) { this.mtu = mtu; return this; } - @Override public ServerRSocketFactory errorConsumer(Consumer errorConsumer) { this.errorConsumer = errorConsumer; return this; @@ -389,7 +347,10 @@ private Mono processSetupFrame( RSocketClient rSocketClient = new RSocketClient( - multiplexer.asServerConnection(), errorConsumer, StreamIdSupplier.serverSupplier()); + multiplexer.asServerConnection(), + frameDecoder, + errorConsumer, + StreamIdSupplier.serverSupplier()); Mono wrappedRSocketClient = Mono.just(rSocketClient).map(plugins::applyClient); @@ -398,7 +359,7 @@ private Mono processSetupFrame( sender -> acceptor.get().accept(setupPayload, sender).map(plugins::applyServer)) .map( handler -> - new RSocketServer(multiplexer.asClientConnection(), handler, errorConsumer)) + new RSocketServer(multiplexer.asClientConnection(), handler, frameDecoder, errorConsumer)) .then(); } } diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java index c9f567873..88a5702df 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java @@ -26,9 +26,10 @@ import io.rsocket.exceptions.ApplicationException; 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 java.util.function.Function; import javax.annotation.Nullable; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; @@ -44,6 +45,7 @@ class RSocketServer implements RSocket { private final DuplexConnection connection; private final RSocket requestHandler; + private final Function frameDecoder; private final Consumer errorConsumer; private final IntObjectHashMap sendingSubscriptions; @@ -53,9 +55,13 @@ class RSocketServer implements RSocket { private Disposable receiveDisposable; RSocketServer( - DuplexConnection connection, RSocket requestHandler, Consumer errorConsumer) { + DuplexConnection connection, + RSocket requestHandler, + Function frameDecoder, + Consumer errorConsumer) { this.connection = connection; this.requestHandler = requestHandler; + this.frameDecoder = frameDecoder; this.errorConsumer = errorConsumer; this.sendingSubscriptions = new IntObjectHashMap<>(); this.channelProcessors = new IntObjectHashMap<>(); @@ -227,9 +233,9 @@ private Mono handleFrame(Frame frame) { Subscriber receiver; switch (frame.getType()) { case FIRE_AND_FORGET: - return handleFireAndForget(streamId, fireAndForget(new PayloadImpl(frame))); + return handleFireAndForget(streamId, fireAndForget(frameDecoder.apply(frame))); case REQUEST_RESPONSE: - return handleRequestResponse(streamId, requestResponse(new PayloadImpl(frame))); + return handleRequestResponse(streamId, requestResponse(frameDecoder.apply(frame))); case CANCEL: return handleCancelFrame(streamId); case KEEPALIVE: @@ -238,14 +244,14 @@ private Mono handleFrame(Frame frame) { return handleRequestN(streamId, frame); case REQUEST_STREAM: return handleStream( - streamId, requestStream(new PayloadImpl(frame)), initialRequestN(frame)); + streamId, requestStream(frameDecoder.apply(frame)), initialRequestN(frame)); case REQUEST_CHANNEL: return handleChannel(streamId, frame); case PAYLOAD: // TODO: Hook in receiving socket. return Mono.empty(); case METADATA_PUSH: - return metadataPush(new PayloadImpl(frame)); + return metadataPush(frameDecoder.apply(frame)); case LEASE: // Lease must not be received here as this is the server end of the socket which sends // leases. @@ -253,7 +259,7 @@ private Mono handleFrame(Frame frame) { case NEXT: receiver = getChannelProcessor(streamId); if (receiver != null) { - receiver.onNext(new PayloadImpl(frame)); + receiver.onNext(frameDecoder.apply(frame)); } return Mono.empty(); case COMPLETE: @@ -271,7 +277,7 @@ private Mono handleFrame(Frame frame) { case NEXT_COMPLETE: receiver = getChannelProcessor(streamId); if (receiver != null) { - receiver.onNext(new PayloadImpl(frame)); + receiver.onNext(frameDecoder.apply(frame)); receiver.onComplete(); } @@ -308,7 +314,9 @@ private Mono handleRequestResponse(int streamId, Mono response) { if (payload.hasMetadata()) { flags = Frame.setFlag(flags, FLAGS_M); } - return Frame.PayloadFrame.from(streamId, FrameType.NEXT_COMPLETE, payload, flags); + final Frame frame = Frame.PayloadFrame.from(streamId, FrameType.NEXT_COMPLETE, payload, flags); + payload.release(); + return frame; }) .doOnError(errorConsumer) .onErrorResume(t -> Mono.just(Frame.Error.from(streamId, t))) @@ -322,7 +330,11 @@ private Mono handleRequestResponse(int streamId, Mono response) { private Mono handleStream(int streamId, Flux response, int initialRequestN) { response - .map(payload -> Frame.PayloadFrame.from(streamId, FrameType.NEXT, payload)) + .map(payload -> { + final Frame frame = Frame.PayloadFrame.from(streamId, FrameType.NEXT, payload); + payload.release(); + return frame; + }) .transform( frameFlux -> { LimitableRequestPublisher frames = LimitableRequestPublisher.wrap(frameFlux); @@ -364,7 +376,7 @@ private Mono handleChannel(int streamId, Frame firstFrame) { // not chained, as the payload should be enqueued in the Unicast processor before this method // returns // and any later payload can be processed - frames.onNext(new PayloadImpl(firstFrame)); + frames.onNext(frameDecoder.apply(firstFrame)); return handleStream(streamId, requestChannel(payloads), initialRequestN(firstFrame)); } diff --git a/rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java b/rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java index 57ebd3ebe..326e0bc1f 100644 --- a/rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java +++ b/rsocket-core/src/main/java/io/rsocket/fragmentation/FrameReassembler.java @@ -51,11 +51,11 @@ public synchronized void append(Frame frame) { if (FrameHeaderFlyweight.hasMetadataLengthField(frameType)) { metadataOffset += FrameHeaderFlyweight.FRAME_LENGTH_SIZE; } - metadataBuffer.addComponent(true, byteBuf.retainedSlice(metadataOffset, metadataLength)); + metadataBuffer.addComponent(true, byteBuf.slice(metadataOffset, metadataLength)); } if (0 < dataLength) { final int dataOffset = FrameHeaderFlyweight.dataOffset(byteBuf, frameType, frameLength); - dataBuffer.addComponent(true, byteBuf.retainedSlice(dataOffset, dataLength)); + dataBuffer.addComponent(true, byteBuf.slice(dataOffset, dataLength)); } } diff --git a/rsocket-core/src/main/java/io/rsocket/lease/Lease.java b/rsocket-core/src/main/java/io/rsocket/lease/Lease.java index 416bc1998..24f075a95 100644 --- a/rsocket-core/src/main/java/io/rsocket/lease/Lease.java +++ b/rsocket-core/src/main/java/io/rsocket/lease/Lease.java @@ -16,8 +16,8 @@ package io.rsocket.lease; -import java.nio.ByteBuffer; import javax.annotation.Nullable; +import java.nio.ByteBuffer; /** A contract for RSocket lease, which is sent by a request acceptor and is time bound. */ public interface Lease { diff --git a/rsocket-core/src/main/java/io/rsocket/util/ByteBufPayload.java b/rsocket-core/src/main/java/io/rsocket/util/ByteBufPayload.java new file mode 100644 index 000000000..b88f4c9a3 --- /dev/null +++ b/rsocket-core/src/main/java/io/rsocket/util/ByteBufPayload.java @@ -0,0 +1,161 @@ +/* + * Copyright 2016 Netflix, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.rsocket.util; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.Unpooled; +import io.netty.util.AbstractReferenceCounted; +import io.rsocket.Payload; + +import javax.annotation.Nullable; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.Charset; + +public final class ByteBufPayload extends AbstractReferenceCounted implements Payload { + private final ByteBuf data; + private final ByteBuf metadata; + + private ByteBufPayload(ByteBuf data) { + this.data = data.asReadOnly(); + this.metadata = null; + } + + private ByteBufPayload(ByteBuf data, @Nullable ByteBuf metadata) { + this.data = data.asReadOnly(); + this.metadata = metadata == null ? null : metadata.asReadOnly(); + } + + @Override + public boolean hasMetadata() { + return metadata != null; + } + + @Override + public ByteBuf sliceMetadata() { + return metadata == null ? Unpooled.EMPTY_BUFFER : metadata.duplicate(); + } + + @Override + public ByteBuf sliceData() { + return data.duplicate(); + } + + @Override + public ByteBufPayload retain() { + super.retain(); + return this; + } + + @Override + public ByteBufPayload retain(int increment) { + super.retain(increment); + return this; + } + + @Override + public ByteBufPayload touch() { + data.touch(); + if (metadata != null) { + metadata.touch(); + } + return this; + } + + @Override + public ByteBufPayload touch(Object hint) { + data.touch(hint); + if (metadata != null) { + metadata.touch(hint); + } + return this; + } + + @Override + protected void deallocate() { + data.release(); + if (metadata != null) { + metadata.release(); + } + } + + /** + * Static factory method for a text payload. Mainly looks better than "new ByteBufPayload(data)" + * + * @param data the data of the payload. + * @return a payload. + */ + public static Payload create(String data) { + return new ByteBufPayload(ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, data)); + } + + /** + * Static factory method for a text payload. Mainly looks better than "new ByteBufPayload(data, + * metadata)" + * + * @param data the data of the payload. + * @param metadata the metadata for the payload. + * @return a payload. + */ + public static Payload create(String data, @Nullable String metadata) { + return new ByteBufPayload( + ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, data), + metadata == null ? null : ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, metadata) + ); + } + + public static Payload create(CharSequence data, Charset dataCharset) { + return new ByteBufPayload(ByteBufUtil.encodeString(ByteBufAllocator.DEFAULT, CharBuffer.wrap(data), dataCharset)); + } + + public static Payload create(CharSequence data, Charset dataCharset, @Nullable CharSequence metadata, Charset metadataCharset) { + return new ByteBufPayload( + ByteBufUtil.encodeString(ByteBufAllocator.DEFAULT, CharBuffer.wrap(data), dataCharset), + metadata == null ? null : ByteBufUtil.encodeString(ByteBufAllocator.DEFAULT, CharBuffer.wrap(metadata), metadataCharset) + ); + } + + public static Payload create(byte[] data) { + return new ByteBufPayload(Unpooled.wrappedBuffer(data)); + } + + public static Payload create(byte[] data, @Nullable byte[] metadata) { + return new ByteBufPayload(Unpooled.wrappedBuffer(data), metadata == null ? null : Unpooled.wrappedBuffer(metadata)); + } + + public static Payload create(ByteBuffer data) { + return new ByteBufPayload(Unpooled.wrappedBuffer(data)); + } + + public static Payload create(ByteBuffer data, @Nullable ByteBuffer metadata) { + return new ByteBufPayload(Unpooled.wrappedBuffer(data), metadata == null ? null : Unpooled.wrappedBuffer(metadata)); + } + + public static Payload create(ByteBuf data) { + return new ByteBufPayload(data); + } + + public static Payload create(ByteBuf data, @Nullable ByteBuf metadata) { + return new ByteBufPayload(data, metadata); + } + + public static Payload create(Payload payload) { + return new ByteBufPayload(payload.sliceData(), payload.hasMetadata() ? payload.sliceMetadata() : null); + } +} diff --git a/rsocket-core/src/main/java/io/rsocket/util/DefaultPayload.java b/rsocket-core/src/main/java/io/rsocket/util/DefaultPayload.java new file mode 100644 index 000000000..df7fd2662 --- /dev/null +++ b/rsocket-core/src/main/java/io/rsocket/util/DefaultPayload.java @@ -0,0 +1,163 @@ +/* + * Copyright 2016 Netflix, Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.rsocket.util; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.rsocket.Payload; +import java.nio.ByteBuffer; +import java.nio.CharBuffer; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import javax.annotation.Nullable; + +/** + * An implementation of {@link Payload}. This implementation is not thread-safe, and hence + * any method can not be invoked concurrently. + */ +public final class DefaultPayload implements Payload { + public static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocateDirect(0); + + private final ByteBuffer data; + private final ByteBuffer metadata; + + private DefaultPayload(ByteBuffer data) { + this.data = data.asReadOnlyBuffer(); + this.metadata = null; + } + + private DefaultPayload(ByteBuffer data, @Nullable ByteBuffer metadata) { + this.data = data.asReadOnlyBuffer(); + this.metadata = metadata == null ? null : metadata.asReadOnlyBuffer(); + } + + @Override + public boolean hasMetadata() { + return metadata != null; + } + + @Override + public ByteBuf sliceMetadata() { + return metadata == null ? Unpooled.EMPTY_BUFFER : Unpooled.wrappedBuffer(metadata); + } + + @Override + public ByteBuf sliceData() { + return Unpooled.wrappedBuffer(data); + } + + @Override + public ByteBuffer getMetadata() { + return metadata == null ? DefaultPayload.EMPTY_BUFFER : metadata.duplicate(); + } + + @Override + public ByteBuffer getData() { + return data.duplicate(); + } + + @Override + public int refCnt() { + return 1; + } + + @Override + public DefaultPayload retain() { + return this; + } + + @Override + public DefaultPayload retain(int increment) { + return this; + } + + @Override + public DefaultPayload touch() { + return this; + } + + @Override + public DefaultPayload touch(Object hint) { + return this; + } + + @Override + public boolean release() { + return false; + } + + @Override + public boolean release(int decrement) { + return false; + } + + /** + * Static factory method for a text payload. Mainly looks better than "new DefaultPayload(data)" + * + * @param data the data of the payload. + * @return a payload. + */ + public static Payload create(CharSequence data) { + return new DefaultPayload(StandardCharsets.UTF_8.encode(CharBuffer.wrap(data))); + } + + /** + * Static factory method for a text payload. Mainly looks better than "new DefaultPayload(data, + * metadata)" + * + * @param data the data of the payload. + * @param metadata the metadata for the payload. + * @return a payload. + */ + public static Payload create(CharSequence data, @Nullable CharSequence metadata) { + return new DefaultPayload( + StandardCharsets.UTF_8.encode(CharBuffer.wrap(data)), + metadata == null ? null : StandardCharsets.UTF_8.encode(CharBuffer.wrap(metadata)) + ); + } + + public static Payload create(CharSequence data, Charset dataCharset) { + return new DefaultPayload(dataCharset.encode(CharBuffer.wrap(data))); + } + + public static Payload create(CharSequence data, Charset dataCharset, @Nullable CharSequence metadata, Charset metadataCharset) { + return new DefaultPayload( + dataCharset.encode(CharBuffer.wrap(data)), + metadata == null ? null : metadataCharset.encode(CharBuffer.wrap(metadata)) + ); + } + + public static Payload create(byte[] data) { + return new DefaultPayload(ByteBuffer.wrap(data)); + } + + public static Payload create(byte[] data, @Nullable byte[] metadata) { + return new DefaultPayload(ByteBuffer.wrap(data), metadata == null ? null : ByteBuffer.wrap(metadata)); + } + + public static Payload create(ByteBuffer data) { + return new DefaultPayload(data); + } + + public static Payload create(ByteBuffer data, @Nullable ByteBuffer metadata) { + return new DefaultPayload(data, metadata); + } + + public static Payload create(Payload payload) { + return new DefaultPayload(payload.getData(), payload.hasMetadata() ? payload.getMetadata() : null); + } +} diff --git a/rsocket-core/src/main/java/io/rsocket/util/EmptyPayload.java b/rsocket-core/src/main/java/io/rsocket/util/EmptyPayload.java new file mode 100644 index 000000000..d0d8e8434 --- /dev/null +++ b/rsocket-core/src/main/java/io/rsocket/util/EmptyPayload.java @@ -0,0 +1,62 @@ +package io.rsocket.util; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.Unpooled; +import io.rsocket.Payload; + +public class EmptyPayload implements Payload { + public static final EmptyPayload INSTANCE = new EmptyPayload(); + + private EmptyPayload() { + } + + @Override + public boolean hasMetadata() { + return false; + } + + @Override + public ByteBuf sliceMetadata() { + return Unpooled.EMPTY_BUFFER; + } + + @Override + public ByteBuf sliceData() { + return Unpooled.EMPTY_BUFFER; + } + + @Override + public int refCnt() { + return 1; + } + + @Override + public EmptyPayload retain() { + return this; + } + + @Override + public EmptyPayload retain(int increment) { + return this; + } + + @Override + public EmptyPayload touch() { + return this; + } + + @Override + public EmptyPayload touch(Object hint) { + return this; + } + + @Override + public boolean release() { + return false; + } + + @Override + public boolean release(int decrement) { + return false; + } +} diff --git a/rsocket-core/src/main/java/io/rsocket/util/PayloadImpl.java b/rsocket-core/src/main/java/io/rsocket/util/PayloadImpl.java deleted file mode 100644 index e86dd270e..000000000 --- a/rsocket-core/src/main/java/io/rsocket/util/PayloadImpl.java +++ /dev/null @@ -1,131 +0,0 @@ -/* - * Copyright 2016 Netflix, Inc. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package io.rsocket.util; - -import io.rsocket.Frame; -import io.rsocket.Payload; -import java.nio.ByteBuffer; -import java.nio.charset.Charset; -import java.nio.charset.StandardCharsets; -import javax.annotation.Nullable; - -/** - * An implementation of {@link Payload}. This implementation is not thread-safe, and hence - * any method can not be invoked concurrently. - */ -public class PayloadImpl implements Payload { - - public static final PayloadImpl EMPTY = - new PayloadImpl(Frame.NULL_BYTEBUFFER, Frame.NULL_BYTEBUFFER, false); - - private final ByteBuffer data; - private final ByteBuffer metadata; - private final int dataStartPosition; - private final int metadataStartPosition; - private final boolean reusable; - - public PayloadImpl(Frame frame) { - this(frame.getData(), frame.hasMetadata() ? frame.getMetadata() : null); - } - - public PayloadImpl(String data) { - this(data, Charset.defaultCharset()); - } - - public PayloadImpl(String data, @Nullable String metadata) { - this(data, StandardCharsets.UTF_8, metadata, StandardCharsets.UTF_8); - } - - public PayloadImpl(String data, Charset dataCharset) { - this(dataCharset.encode(data), null); - } - - public PayloadImpl( - String data, Charset dataCharset, @Nullable String metadata, Charset metaDataCharset) { - this(dataCharset.encode(data), metadata == null ? null : metaDataCharset.encode(metadata)); - } - - public PayloadImpl(byte[] data) { - this(ByteBuffer.wrap(data), Frame.NULL_BYTEBUFFER); - } - - public PayloadImpl(byte[] data, @Nullable byte[] metadata) { - this(ByteBuffer.wrap(data), metadata == null ? null : ByteBuffer.wrap(metadata)); - } - - public PayloadImpl(ByteBuffer data) { - this(data, Frame.NULL_BYTEBUFFER); - } - - public PayloadImpl(ByteBuffer data, @Nullable ByteBuffer metadata) { - this(data, metadata, true); - } - - public PayloadImpl(ByteBuffer data, ByteBuffer metadata, boolean reusable) { - this.data = data; - this.metadata = metadata; - this.reusable = reusable; - this.dataStartPosition = reusable ? this.data.position() : 0; - this.metadataStartPosition = (reusable && metadata != null) ? this.metadata.position() : 0; - } - - @Override - public ByteBuffer getData() { - if (reusable) { - data.position(dataStartPosition); - } - return data; - } - - @Override - public ByteBuffer getMetadata() { - if (metadata == null) { - return Frame.NULL_BYTEBUFFER; - } - if (reusable) { - metadata.position(metadataStartPosition); - } - return metadata; - } - - @Override - public boolean hasMetadata() { - return metadata != null; - } - - /** - * Static factory method for a text payload. Mainly looks better than "new PayloadImpl(data)" - * - * @param data the data of the payload. - * @return a payload. - */ - public static Payload textPayload(String data) { - return new PayloadImpl(data); - } - - /** - * Static factory method for a text payload. Mainly looks better than "new PayloadImpl(data, - * metadata)" - * - * @param data the data of the payload. - * @param metadata the metadata for the payload. - * @return a payload. - */ - public static Payload textPayload(String data, @Nullable String metadata) { - return new PayloadImpl(data, metadata); - } -} diff --git a/rsocket-core/src/test/java/io/rsocket/FrameTest.java b/rsocket-core/src/test/java/io/rsocket/FrameTest.java index 7850914ce..2518ac610 100644 --- a/rsocket-core/src/test/java/io/rsocket/FrameTest.java +++ b/rsocket-core/src/test/java/io/rsocket/FrameTest.java @@ -3,14 +3,14 @@ import static org.junit.Assert.assertEquals; import io.rsocket.frame.FrameHeaderFlyweight; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import org.junit.Test; public class FrameTest { @Test public void testFrameToString() { final Frame requestFrame = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl("streaming in -> 0"), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create("streaming in -> 0"), 1); assertEquals( "Frame => Stream ID: 1 Type: REQUEST_RESPONSE Payload: data: \"streaming in -> 0\" ", requestFrame.toString()); @@ -20,7 +20,7 @@ public void testFrameToString() { public void testFrameWithMetadataToString() { final Frame requestFrame = Frame.Request.from( - 1, FrameType.REQUEST_RESPONSE, new PayloadImpl("streaming in -> 0", "metadata"), 1); + 1, FrameType.REQUEST_RESPONSE, DefaultPayload.create("streaming in -> 0", "metadata"), 1); assertEquals( "Frame => Stream ID: 1 Type: REQUEST_RESPONSE Payload: metadata: \"metadata\" data: \"streaming in -> 0\" ", requestFrame.toString()); @@ -30,7 +30,7 @@ public void testFrameWithMetadataToString() { public void testPayload() { Frame frame = Frame.PayloadFrame.from( - 1, FrameType.NEXT_COMPLETE, new PayloadImpl("Hello"), FrameHeaderFlyweight.FLAGS_C); + 1, FrameType.NEXT_COMPLETE, DefaultPayload.create("Hello"), FrameHeaderFlyweight.FLAGS_C); frame.toString(); } } diff --git a/rsocket-core/src/test/java/io/rsocket/RSocketClientTest.java b/rsocket-core/src/test/java/io/rsocket/RSocketClientTest.java index 7ad6e4f10..5ff0db8b6 100644 --- a/rsocket-core/src/test/java/io/rsocket/RSocketClientTest.java +++ b/rsocket-core/src/test/java/io/rsocket/RSocketClientTest.java @@ -33,11 +33,15 @@ import io.rsocket.exceptions.RejectedSetupException; import io.rsocket.frame.RequestFrameFlyweight; import io.rsocket.test.util.TestSubscriber; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; + +import io.rsocket.util.EmptyPayload; +import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; import org.reactivestreams.Publisher; @@ -68,7 +72,7 @@ public void testInvalidFrameOnStream0() { @Test(timeout = 2_000) public void testStreamInitialN() { - Flux stream = rule.socket.requestStream(PayloadImpl.EMPTY); + Flux stream = rule.socket.requestStream(EmptyPayload.INSTANCE); BaseSubscriber subscriber = new BaseSubscriber() { @@ -110,7 +114,7 @@ public void testHandleSetupException() { @Test(timeout = 2_000) public void testHandleApplicationException() { rule.connection.clearSendReceiveBuffers(); - Publisher response = rule.socket.requestResponse(PayloadImpl.EMPTY); + Publisher response = rule.socket.requestResponse(EmptyPayload.INSTANCE); Subscriber responseSub = TestSubscriber.create(); response.subscribe(responseSub); @@ -123,13 +127,13 @@ public void testHandleApplicationException() { @Test(timeout = 2_000) public void testHandleValidFrame() { - Publisher response = rule.socket.requestResponse(PayloadImpl.EMPTY); + Publisher response = rule.socket.requestResponse(EmptyPayload.INSTANCE); Subscriber sub = TestSubscriber.create(); response.subscribe(sub); int streamId = rule.getStreamIdForRequestType(REQUEST_RESPONSE); rule.connection.addToReceivedBuffer( - Frame.PayloadFrame.from(streamId, NEXT_COMPLETE, PayloadImpl.EMPTY)); + Frame.PayloadFrame.from(streamId, NEXT_COMPLETE, EmptyPayload.INSTANCE)); verify(sub).onNext(anyPayload()); verify(sub).onComplete(); @@ -137,7 +141,7 @@ public void testHandleValidFrame() { @Test(timeout = 2_000) public void testRequestReplyWithCancel() { - Mono response = rule.socket.requestResponse(PayloadImpl.EMPTY); + Mono response = rule.socket.requestResponse(EmptyPayload.INSTANCE); try { response.block(Duration.ofMillis(100)); @@ -159,7 +163,7 @@ public void testRequestReplyWithCancel() { @Test(timeout = 2_000) public void testRequestReplyErrorOnSend() { rule.connection.setAvailability(0); // Fails send - Mono response = rule.socket.requestResponse(PayloadImpl.EMPTY); + Mono response = rule.socket.requestResponse(EmptyPayload.INSTANCE); Subscriber responseSub = TestSubscriber.create(10); response.subscribe(responseSub); @@ -173,7 +177,7 @@ public void testRequestReplyErrorOnSend() { @Test(timeout = 2_000) public void testLazyRequestResponse() { - Publisher response = rule.socket.requestResponse(PayloadImpl.EMPTY); + Publisher response = rule.socket.requestResponse(EmptyPayload.INSTANCE); int streamId = sendRequestResponse(response); rule.connection.clearSendReceiveBuffers(); int streamId2 = sendRequestResponse(response); @@ -185,7 +189,7 @@ public int sendRequestResponse(Publisher response) { response.subscribe(sub); int streamId = rule.getStreamIdForRequestType(REQUEST_RESPONSE); rule.connection.addToReceivedBuffer( - Frame.PayloadFrame.from(streamId, NEXT_COMPLETE, PayloadImpl.EMPTY)); + Frame.PayloadFrame.from(streamId, NEXT_COMPLETE, EmptyPayload.INSTANCE)); verify(sub).onNext(anyPayload()); verify(sub).onComplete(); return streamId; @@ -196,6 +200,7 @@ public static class ClientSocketRule extends AbstractSocketRule { protected RSocketClient newRSocket() { return new RSocketClient( connection, + DefaultPayload::create, throwable -> errors.add(throwable), StreamIdSupplier.clientSupplier(), Duration.ofMillis(100), diff --git a/rsocket-core/src/test/java/io/rsocket/RSocketServerTest.java b/rsocket-core/src/test/java/io/rsocket/RSocketServerTest.java index bd6e18792..a0396282d 100644 --- a/rsocket-core/src/test/java/io/rsocket/RSocketServerTest.java +++ b/rsocket-core/src/test/java/io/rsocket/RSocketServerTest.java @@ -25,10 +25,13 @@ import io.netty.buffer.Unpooled; import io.rsocket.test.util.TestDuplexConnection; import io.rsocket.test.util.TestSubscriber; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.util.Collection; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; + +import io.rsocket.util.EmptyPayload; import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; @@ -127,11 +130,11 @@ public void setAcceptingSocket(RSocket acceptingSocket) { @Override protected RSocketServer newRSocket() { - return new RSocketServer(connection, acceptingSocket, throwable -> errors.add(throwable)); + return new RSocketServer(connection, acceptingSocket, DefaultPayload::create, throwable -> errors.add(throwable)); } private void sendRequest(int streamId, FrameType frameType) { - Frame request = Frame.Request.from(streamId, frameType, PayloadImpl.EMPTY, 1); + Frame request = Frame.Request.from(streamId, frameType, EmptyPayload.INSTANCE, 1); connection.addToReceivedBuffer(request); connection.addToReceivedBuffer(Frame.RequestN.from(streamId, 2)); } diff --git a/rsocket-core/src/test/java/io/rsocket/RSocketTest.java b/rsocket-core/src/test/java/io/rsocket/RSocketTest.java index 00974154a..3c87cafae 100644 --- a/rsocket-core/src/test/java/io/rsocket/RSocketTest.java +++ b/rsocket-core/src/test/java/io/rsocket/RSocketTest.java @@ -24,9 +24,11 @@ import io.rsocket.exceptions.ApplicationException; import io.rsocket.test.util.LocalDuplexConnection; import io.rsocket.test.util.TestSubscriber; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.util.ArrayList; import java.util.concurrent.CountDownLatch; + +import io.rsocket.util.EmptyPayload; import org.hamcrest.MatcherAssert; import org.junit.Assert; import org.junit.Rule; @@ -47,7 +49,7 @@ public class RSocketTest { @Test(timeout = 2_000) public void testRequestReplyNoError() { Subscriber subscriber = TestSubscriber.create(); - rule.crs.requestResponse(new PayloadImpl("hello")).subscribe(subscriber); + rule.crs.requestResponse(DefaultPayload.create("hello")).subscribe(subscriber); verify(subscriber).onNext(TestSubscriber.anyPayload()); verify(subscriber).onComplete(); rule.assertNoErrors(); @@ -63,7 +65,7 @@ public Mono requestResponse(Payload payload) { } }); Subscriber subscriber = TestSubscriber.create(); - rule.crs.requestResponse(PayloadImpl.EMPTY).subscribe(subscriber); + rule.crs.requestResponse(EmptyPayload.INSTANCE).subscribe(subscriber); verify(subscriber).onError(any(ApplicationException.class)); // Client sees error through normal API @@ -75,7 +77,7 @@ public Mono requestResponse(Payload payload) { @Test(timeout = 2000) public void testChannel() throws Exception { CountDownLatch latch = new CountDownLatch(10); - Flux requests = Flux.range(0, 10).map(i -> new PayloadImpl("streaming in -> " + i)); + Flux requests = Flux.range(0, 10).map(i -> DefaultPayload.create("streaming in -> " + i)); Flux responses = rule.crs.requestChannel(requests); @@ -126,22 +128,26 @@ public Mono requestResponse(Payload payload) { @Override public Flux requestChannel(Publisher payloads) { Flux.from(payloads) - .map(payload -> new PayloadImpl("server got -> [" + payload.toString() + "]")) + .map(payload -> DefaultPayload.create("server got -> [" + payload.toString() + "]")) .subscribe(); return Flux.range(1, 10) .map( - payload -> new PayloadImpl("server got -> [" + payload.toString() + "]")); + payload -> DefaultPayload.create("server got -> [" + payload.toString() + "]")); } }; srs = new RSocketServer( - serverConnection, requestAcceptor, throwable -> serverErrors.add(throwable)); + serverConnection, + requestAcceptor, + DefaultPayload::create, + throwable -> serverErrors.add(throwable)); crs = new RSocketClient( clientConnection, + DefaultPayload::create, throwable -> clientErrors.add(throwable), StreamIdSupplier.clientSupplier()); } diff --git a/rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationDuplexConnectionTest.java b/rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationDuplexConnectionTest.java index 2228851d7..1c06513de 100644 --- a/rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationDuplexConnectionTest.java +++ b/rsocket-core/src/test/java/io/rsocket/fragmentation/FragmentationDuplexConnectionTest.java @@ -25,7 +25,8 @@ import io.rsocket.DuplexConnection; import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.nio.ByteBuffer; import java.util.concurrent.ThreadLocalRandom; import org.junit.Test; @@ -55,7 +56,7 @@ public void testSendOneWithFragmentation() { ByteBuffer metadata = createRandomBytes(16); Frame frame = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FragmentationDuplexConnection duplexConnection = new FragmentationDuplexConnection(mockConnection, 2); @@ -99,11 +100,11 @@ public void testShouldFragmentMultiple() { ByteBuffer metadata = createRandomBytes(16); Frame frame1 = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); Frame frame2 = - Frame.Request.from(2, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(2, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); Frame frame3 = - Frame.Request.from(3, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(3, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FragmentationDuplexConnection duplexConnection = new FragmentationDuplexConnection(mockConnection, 2); @@ -118,7 +119,7 @@ public void testReassembleFragmentFrame() { ByteBuffer data = createRandomBytes(16); ByteBuffer metadata = createRandomBytes(16); Frame frame = - Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); Flux fragmentedFrames = frameFragmenter.fragment(frame); EmitterProcessor processor = EmitterProcessor.create(128); diff --git a/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameFragmenterTest.java b/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameFragmenterTest.java index e6119d294..2e5a24685 100644 --- a/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameFragmenterTest.java +++ b/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameFragmenterTest.java @@ -18,7 +18,8 @@ import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.nio.ByteBuffer; import java.util.concurrent.ThreadLocalRandom; import org.junit.Test; @@ -31,7 +32,7 @@ public void testFragmentWithMetadataAndData() { ByteBuffer metadata = createRandomBytes(16); Frame from = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); @@ -44,7 +45,7 @@ public void testFragmentWithMetadataAndDataWithOddData() { ByteBuffer metadata = createRandomBytes(17); Frame from = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); @@ -57,7 +58,7 @@ public void testFragmentWithMetadataOnly() { ByteBuffer metadata = createRandomBytes(16); Frame from = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); @@ -70,7 +71,7 @@ public void testFragmentWithDataOnly() { ByteBuffer metadata = ByteBuffer.allocate(0); Frame from = - Frame.Request.from(1, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); diff --git a/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameReassemblerTest.java b/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameReassemblerTest.java index 278c7fece..f95f8f0e6 100644 --- a/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameReassemblerTest.java +++ b/rsocket-core/src/test/java/io/rsocket/fragmentation/FrameReassemblerTest.java @@ -18,7 +18,8 @@ import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.nio.ByteBuffer; import java.util.concurrent.ThreadLocalRandom; import org.junit.Test; @@ -31,7 +32,7 @@ public void testAppend() { ByteBuffer metadata = createRandomBytes(16); Frame from = - Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); FrameReassembler reassembler = new FrameReassembler(from); frameFragmenter.fragment(from).subscribe(reassembler::append); @@ -46,7 +47,7 @@ private ByteBuffer createRandomBytes(int size) { ByteBuffer data = createRandomBytes(16); ByteBuffer metadata = createRandomBytes(16); - Frame from = Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame from = Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, DefaultPayload.create(data, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); @@ -83,7 +84,7 @@ public void testReassmembleAndClear() { ByteBuffer data = createRandomBytes(16); ByteBuffer metadata = createRandomBytes(16); - Frame request = Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, new PayloadImpl(data, metadata), 1); + Frame request = Frame.Request.from(1024, FrameType.REQUEST_RESPONSE, DefaultPayload.createdata, metadata), 1); FrameFragmenter frameFragmenter = new FrameFragmenter(2); diff --git a/rsocket-core/src/test/java/io/rsocket/frame/RequestFrameFlyweightTest.java b/rsocket-core/src/test/java/io/rsocket/frame/RequestFrameFlyweightTest.java index 828a75b79..21895c469 100644 --- a/rsocket-core/src/test/java/io/rsocket/frame/RequestFrameFlyweightTest.java +++ b/rsocket-core/src/test/java/io/rsocket/frame/RequestFrameFlyweightTest.java @@ -24,7 +24,8 @@ import io.netty.buffer.Unpooled; import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.Payload; +import io.rsocket.util.DefaultPayload; import java.nio.charset.StandardCharsets; import org.junit.Test; @@ -45,8 +46,8 @@ public void testEncoding() { assertEquals( "000010000000011900000000010000026d6464", ByteBufUtil.hexDump(byteBuf, 0, encoded)); - PayloadImpl payload = - new PayloadImpl(Frame.from(stringToBuf("000010000000011900000000010000026d6464"))); + Payload payload = + DefaultPayload.create(Frame.from(stringToBuf("000010000000011900000000010000026d6464"))); assertEquals("md", StandardCharsets.UTF_8.decode(payload.getMetadata()).toString()); } @@ -64,8 +65,8 @@ public void testEncodingWithEmptyMetadata() { Unpooled.copiedBuffer("d", StandardCharsets.UTF_8)); assertEquals("00000e0000000119000000000100000064", ByteBufUtil.hexDump(byteBuf, 0, encoded)); - PayloadImpl payload = - new PayloadImpl(Frame.from(stringToBuf("00000e0000000119000000000100000064"))); + Payload payload = + DefaultPayload.create(Frame.from(stringToBuf("00000e0000000119000000000100000064"))); assertEquals("", StandardCharsets.UTF_8.decode(payload.getMetadata()).toString()); } @@ -83,7 +84,7 @@ public void testEncodingWithNullMetadata() { Unpooled.copiedBuffer("d", StandardCharsets.UTF_8)); assertEquals("00000b0000000118000000000164", ByteBufUtil.hexDump(byteBuf, 0, encoded)); - PayloadImpl payload = new PayloadImpl(Frame.from(stringToBuf("00000b0000000118000000000164"))); + Payload payload = DefaultPayload.create(Frame.from(stringToBuf("00000b0000000118000000000164"))); assertFalse(payload.hasMetadata()); } diff --git a/rsocket-core/src/test/java/io/rsocket/resume/ResumeCacheTest.java b/rsocket-core/src/test/java/io/rsocket/resume/ResumeCacheTest.java index 33bbde9b8..524d7b13d 100644 --- a/rsocket-core/src/test/java/io/rsocket/resume/ResumeCacheTest.java +++ b/rsocket-core/src/test/java/io/rsocket/resume/ResumeCacheTest.java @@ -4,14 +4,14 @@ import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import org.junit.Test; import reactor.core.publisher.Flux; public class ResumeCacheTest { private Frame CANCEL = Frame.Cancel.from(1); private Frame STREAM = - Frame.Request.from(1, FrameType.REQUEST_STREAM, new PayloadImpl("Test"), 100); + Frame.Request.from(1, FrameType.REQUEST_STREAM, DefaultPayload.create("Test"), 100); private ResumeCache cache = new ResumeCache(ResumePositionCounter.frames(), 2); diff --git a/rsocket-core/src/test/java/io/rsocket/resume/ResumeUtilTest.java b/rsocket-core/src/test/java/io/rsocket/resume/ResumeUtilTest.java index 68f64c1ba..cbc13d18f 100644 --- a/rsocket-core/src/test/java/io/rsocket/resume/ResumeUtilTest.java +++ b/rsocket-core/src/test/java/io/rsocket/resume/ResumeUtilTest.java @@ -6,13 +6,13 @@ import io.rsocket.Frame; import io.rsocket.FrameType; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import org.junit.Test; public class ResumeUtilTest { private Frame CANCEL = Frame.Cancel.from(1); private Frame STREAM = - Frame.Request.from(1, FrameType.REQUEST_STREAM, new PayloadImpl("Test"), 100); + Frame.Request.from(1, FrameType.REQUEST_STREAM, DefaultPayload.create("Test"), 100); @Test public void testSupportedTypes() { diff --git a/rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java b/rsocket-core/src/test/java/io/rsocket/util/DefaultPayloadTest.java similarity index 81% rename from rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java rename to rsocket-core/src/test/java/io/rsocket/util/DefaultPayloadTest.java index 9099efa9a..d97682850 100644 --- a/rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java +++ b/rsocket-core/src/test/java/io/rsocket/util/DefaultPayloadTest.java @@ -13,7 +13,7 @@ package io.rsocket.util; -import static io.rsocket.util.PayloadImpl.textPayload; +import static io.rsocket.util.DefaultPayload.create; import static org.hamcrest.MatcherAssert.*; import static org.hamcrest.Matchers.*; @@ -21,20 +21,20 @@ import javax.annotation.Nullable; import org.junit.Test; -public class PayloadImplTest { +public class DefaultPayloadTest { public static final String DATA_VAL = "data"; public static final String METADATA_VAL = "metadata"; @Test public void testReuse() { - PayloadImpl p = new PayloadImpl(DATA_VAL, METADATA_VAL); + Payload p = DefaultPayload.create(DATA_VAL, METADATA_VAL); assertDataAndMetadata(p, DATA_VAL, METADATA_VAL); assertDataAndMetadata(p, DATA_VAL, METADATA_VAL); } @Test public void testReuseWithExternalMark() { - PayloadImpl p = new PayloadImpl(DATA_VAL, METADATA_VAL); + Payload p = DefaultPayload.create(DATA_VAL, METADATA_VAL); assertDataAndMetadata(p, DATA_VAL, METADATA_VAL); p.getData().position(2).mark(); assertDataAndMetadata(p, DATA_VAL, METADATA_VAL); @@ -52,7 +52,7 @@ public void assertDataAndMetadata(Payload p, String dataVal, @Nullable String me @Test public void staticMethods() { - assertDataAndMetadata(textPayload(DATA_VAL, METADATA_VAL), DATA_VAL, METADATA_VAL); - assertDataAndMetadata(textPayload(DATA_VAL), DATA_VAL, null); + assertDataAndMetadata(DefaultPayload.create(DATA_VAL, METADATA_VAL), DATA_VAL, METADATA_VAL); + assertDataAndMetadata(DefaultPayload.create(DATA_VAL), DATA_VAL, null); } } diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java index 4ee9d33c2..923eadb1b 100644 --- a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java @@ -21,7 +21,8 @@ import io.rsocket.SocketAcceptor; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.time.Duration; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; @@ -43,7 +44,7 @@ public static void main(String[] args) { .block(); socket - .requestChannel(Flux.interval(Duration.ofMillis(1000)).map(i -> new PayloadImpl("Hello"))) + .requestChannel(Flux.interval(Duration.ofMillis(1000)).map(i -> DefaultPayload.create("Hello"))) .map(Payload::getDataUtf8) .doOnNext(System.out::println) .take(10) @@ -61,7 +62,7 @@ public Flux requestChannel(Publisher payloads) { return Flux.from(payloads) .map(Payload::getDataUtf8) .map(s -> "Echo: " + s) - .map(PayloadImpl::new); + .map(DefaultPayload::create); } }); } diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/duplex/DuplexClient.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/duplex/DuplexClient.java index 59ceb6e35..051ddc65e 100644 --- a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/duplex/DuplexClient.java +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/duplex/DuplexClient.java @@ -19,7 +19,8 @@ import io.rsocket.RSocketFactory; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.time.Duration; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -31,7 +32,7 @@ public static void main(String[] args) { .acceptor( (setup, reactiveSocket) -> { reactiveSocket - .requestStream(new PayloadImpl("Hello-Bidi")) + .requestStream(DefaultPayload.create("Hello-Bidi")) .map(Payload::getDataUtf8) .log() .subscribe(); @@ -50,7 +51,7 @@ public static void main(String[] args) { @Override public Flux requestStream(Payload payload) { return Flux.interval(Duration.ofSeconds(1)) - .map(aLong -> new PayloadImpl("Bi-di Response => " + aLong)); + .map(aLong -> DefaultPayload.create("Bi-di Response => " + aLong)); } }) .transport(TcpClientTransport.create("localhost", 7000)) diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/requestresponse/HelloWorldClient.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/requestresponse/HelloWorldClient.java index 258a997bd..3282a72e5 100644 --- a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/requestresponse/HelloWorldClient.java +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/requestresponse/HelloWorldClient.java @@ -22,7 +22,7 @@ import io.rsocket.RSocketFactory; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import reactor.core.publisher.Mono; public final class HelloWorldClient { @@ -56,21 +56,21 @@ public Mono requestResponse(Payload p) { .block(); socket - .requestResponse(new PayloadImpl("Hello")) + .requestResponse(DefaultPayload.create("Hello")) .map(Payload::getDataUtf8) .onErrorReturn("error") .doOnNext(System.out::println) .block(); socket - .requestResponse(new PayloadImpl("Hello")) + .requestResponse(DefaultPayload.create("Hello")) .map(Payload::getDataUtf8) .onErrorReturn("error") .doOnNext(System.out::println) .block(); socket - .requestResponse(new PayloadImpl("Hello")) + .requestResponse(DefaultPayload.create("Hello")) .map(Payload::getDataUtf8) .onErrorReturn("error") .doOnNext(System.out::println) diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/StreamingClient.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/StreamingClient.java index 0df129a67..b33ae5626 100644 --- a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/StreamingClient.java +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/StreamingClient.java @@ -16,7 +16,7 @@ import io.rsocket.*; import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.time.Duration; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -37,7 +37,7 @@ public static void main(String[] args) { .block(); socket - .requestStream(new PayloadImpl("Hello")) + .requestStream(DefaultPayload.create("Hello")) .map(Payload::getDataUtf8) .doOnNext(System.out::println) .take(10) @@ -53,7 +53,7 @@ public Mono accept(ConnectionSetupPayload setupPayload, RSocket reactiv @Override public Flux requestStream(Payload payload) { return Flux.interval(Duration.ofMillis(100)) - .map(aLong -> new PayloadImpl("Interval: " + aLong)); + .map(aLong -> DefaultPayload.create("Interval: " + aLong)); } }); } diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java index 935bdd6cc..94652c810 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/IntegrationTest.java @@ -26,7 +26,7 @@ import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.NettyContextCloseable; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import io.rsocket.util.RSocketProxy; import org.junit.After; import org.junit.Assert; @@ -116,13 +116,13 @@ public void startup() { new AbstractRSocket() { @Override public Mono requestResponse(Payload payload) { - return Mono.just(new PayloadImpl("RESPONSE", "METADATA")) + return Mono.just(DefaultPayload.create("RESPONSE", "METADATA")) .doOnSubscribe(s -> requestCount.incrementAndGet()); } @Override public Flux requestStream(Payload payload) { - return Flux.range(1, 10_000).map(i -> new PayloadImpl("data -> " + i)); + return Flux.range(1, 10_000).map(i -> DefaultPayload.create("data -> " + i)); } @Override @@ -151,7 +151,7 @@ public void teardown() { @Test(timeout = 5_000L) public void testRequest() { - client.requestResponse(new PayloadImpl("REQUEST", "META")).block(); + client.requestResponse(DefaultPayload.create("REQUEST", "META")).block(); assertThat("Server did not see the request.", requestCount.get(), is(1)); assertTrue(calledClient); assertTrue(calledServer); @@ -161,7 +161,7 @@ public void testRequest() { @Test(timeout = 5_000L) public void testStream() { Subscriber subscriber = TestSubscriber.createCancelling(); - client.requestStream(new PayloadImpl("start")).subscribe(subscriber); + client.requestStream(DefaultPayload.create("start")).subscribe(subscriber); verify(subscriber).onSubscribe(any()); verifyNoMoreInteractions(subscriber); diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java index 3a0067c12..3c3dacd37 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java @@ -26,7 +26,8 @@ import io.rsocket.transport.netty.client.TcpClientTransport; import io.rsocket.transport.netty.server.NettyContextCloseable; import io.rsocket.transport.netty.server.TcpServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; +import io.rsocket.util.EmptyPayload; import io.rsocket.util.RSocketProxy; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; @@ -77,7 +78,7 @@ public Flux requestStream(Payload payload) { }; RSocket client = buildClient(); Boolean hasElements = - client.requestStream(new PayloadImpl("REQUEST", "META")).log().hasElements().block(); + client.requestStream(DefaultPayload.create("REQUEST", "META")).log().hasElements().block(); assertFalse(hasElements); } @@ -88,13 +89,13 @@ public void testSingleStream() { new AbstractRSocket() { @Override public Flux requestStream(Payload payload) { - return Flux.just(new PayloadImpl("RESPONSE", "METADATA")); + return Flux.just(DefaultPayload.create("RESPONSE", "METADATA")); } }; RSocket client = buildClient(); - Payload result = client.requestStream(new PayloadImpl("REQUEST", "META")).blockLast(); + Payload result = client.requestStream(DefaultPayload.create("REQUEST", "META")).blockLast(); assertEquals("RESPONSE", result.getDataUtf8()); } @@ -105,13 +106,13 @@ public void testZeroPayload() { new AbstractRSocket() { @Override public Flux requestStream(Payload payload) { - return Flux.just(PayloadImpl.EMPTY); + return Flux.just(EmptyPayload.INSTANCE); } }; RSocket client = buildClient(); - Payload result = client.requestStream(new PayloadImpl("REQUEST", "META")).blockFirst(); + Payload result = client.requestStream(DefaultPayload.create("REQUEST", "META")).blockFirst(); assertEquals("", result.getDataUtf8()); } @@ -128,7 +129,7 @@ public Mono requestResponse(Payload payload) { first = false; return Mono.error(new RuntimeException("EX")); } else { - return Mono.just(new PayloadImpl("SUCCESS")); + return Mono.just(DefaultPayload.create("SUCCESS")); } } }; @@ -137,13 +138,13 @@ public Mono requestResponse(Payload payload) { Payload response1 = client - .requestResponse(new PayloadImpl("REQUEST", "META")) - .onErrorReturn(new PayloadImpl("ERROR")) + .requestResponse(DefaultPayload.create("REQUEST", "META")) + .onErrorReturn(DefaultPayload.create("ERROR")) .block(); Payload response2 = client - .requestResponse(new PayloadImpl("REQUEST", "META")) - .onErrorReturn(new PayloadImpl("ERROR")) + .requestResponse(DefaultPayload.create("REQUEST", "META")) + .onErrorReturn(DefaultPayload.create("ERROR")) .block(); assertEquals("ERROR", response1.getDataUtf8()); @@ -168,8 +169,8 @@ public Flux requestStream(Payload payload) { RSocket client = buildClient(); - Flux response1 = client.requestStream(new PayloadImpl("REQUEST1")); - Flux response2 = client.requestStream(new PayloadImpl("REQUEST2")); + Flux response1 = client.requestStream(DefaultPayload.create("REQUEST1")); + Flux response2 = client.requestStream(DefaultPayload.create("REQUEST2")); CountDownLatch nextCountdown = new CountDownLatch(2); CountDownLatch completeCountdown = new CountDownLatch(2); @@ -182,8 +183,8 @@ public Flux requestStream(Payload payload) { .subscribeOn(Schedulers.newSingle("2")) .subscribe(c -> nextCountdown.countDown(), t -> {}, completeCountdown::countDown); - processor1.onNext(new PayloadImpl("RESPONSE1A")); - processor2.onNext(new PayloadImpl("RESPONSE2A")); + processor1.onNext(DefaultPayload.create("RESPONSE1A")); + processor2.onNext(DefaultPayload.create("RESPONSE2A")); nextCountdown.await(); diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java b/rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java index c41af77d2..31051566e 100644 --- a/rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java +++ b/rsocket-examples/src/test/java/io/rsocket/integration/TestingStreaming.java @@ -6,7 +6,8 @@ import io.rsocket.transport.ServerTransport; import io.rsocket.transport.local.LocalClientTransport; import io.rsocket.transport.local.LocalServerTransport; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; @@ -45,7 +46,7 @@ public Flux requestStream(Payload payload) { throw new RuntimeException("BOOM!"); } }) - .map(l -> new PayloadImpl("l -> " + l)) + .map(l -> DefaultPayload.create("l -> " + l)) .cast(Payload.class); } }; @@ -83,7 +84,7 @@ public double availability() { @Override public Flux requestStream(Payload payload) { return Flux.range(1, 1000) - .map(l -> new PayloadImpl("l -> " + l)) + .map(l -> DefaultPayload.create("l -> " + l)) .cast(Payload.class); } }; @@ -111,7 +112,7 @@ private Flux consumer(String s) { rSocket -> { AtomicInteger count = new AtomicInteger(); return Flux.range(1, 100) - .flatMap(i -> rSocket.requestStream(new PayloadImpl("i -> " + i)).take(100), 1); + .flatMap(i -> rSocket.requestStream(DefaultPayload.create("i -> " + i)).take(100), 1); }); } @@ -134,7 +135,7 @@ public double availability() { @Override public Flux requestStream(Payload payload) { return Flux.range(1, 10_000) - .map(l -> new PayloadImpl("l -> " + l)) + .map(l -> DefaultPayload.create("l -> " + l)) .cast(Payload.class); } }; diff --git a/rsocket-load-balancer/src/test/java/io/rsocket/client/LoadBalancedRSocketMonoTest.java b/rsocket-load-balancer/src/test/java/io/rsocket/client/LoadBalancedRSocketMonoTest.java index d1a98775d..e0c1788b4 100644 --- a/rsocket-load-balancer/src/test/java/io/rsocket/client/LoadBalancedRSocketMonoTest.java +++ b/rsocket-load-balancer/src/test/java/io/rsocket/client/LoadBalancedRSocketMonoTest.java @@ -19,13 +19,15 @@ import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.client.filter.RSocketSupplier; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.Arrays; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.function.Function; + +import io.rsocket.util.EmptyPayload; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -96,7 +98,7 @@ private void makeAcall(RSocket balancer) throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); balancer - .requestResponse(PayloadImpl.EMPTY) + .requestResponse(EmptyPayload.INSTANCE) .subscribe( new Subscriber() { @Override diff --git a/rsocket-load-balancer/src/test/java/io/rsocket/client/RSocketSupplierTest.java b/rsocket-load-balancer/src/test/java/io/rsocket/client/RSocketSupplierTest.java index 7fd439ed1..40334d553 100644 --- a/rsocket-load-balancer/src/test/java/io/rsocket/client/RSocketSupplierTest.java +++ b/rsocket-load-balancer/src/test/java/io/rsocket/client/RSocketSupplierTest.java @@ -25,11 +25,14 @@ import io.rsocket.RSocket; import io.rsocket.client.filter.RSocketSupplier; import io.rsocket.test.TestSubscriber; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; + +import io.rsocket.util.EmptyPayload; import org.junit.Test; import org.mockito.Mockito; import org.reactivestreams.Publisher; @@ -44,7 +47,7 @@ public void testError() throws InterruptedException { testRSocket( (latch, socket) -> { assertEquals(1.0, socket.availability(), 0.0); - Publisher payloadPublisher = socket.requestResponse(PayloadImpl.EMPTY); + Publisher payloadPublisher = socket.requestResponse(EmptyPayload.INSTANCE); Subscriber subscriber = TestSubscriber.create(); payloadPublisher.subscribe(subscriber); @@ -73,7 +76,7 @@ public void testWidowReset() throws InterruptedException { testRSocket( (latch, socket) -> { assertEquals(1.0, socket.availability(), 0.0); - Publisher payloadPublisher = socket.requestResponse(PayloadImpl.EMPTY); + Publisher payloadPublisher = socket.requestResponse(EmptyPayload.INSTANCE); Subscriber subscriber = TestSubscriber.create(); payloadPublisher.subscribe(subscriber); @@ -106,7 +109,7 @@ private void testRSocket(BiConsumer f) throws Interrupt new TestingRSocket( input -> { if (count.getAndIncrement() < 1) { - return PayloadImpl.EMPTY; + return EmptyPayload.INSTANCE; } else { throw new RuntimeException(); } diff --git a/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java b/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java index 0210341d9..3d73e8edc 100644 --- a/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java +++ b/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java @@ -22,8 +22,11 @@ import io.rsocket.RSocket; import io.rsocket.client.filter.RSockets; import io.rsocket.exceptions.TimeoutException; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; + import java.time.Duration; + +import io.rsocket.util.EmptyPayload; import org.hamcrest.MatcherAssert; import org.junit.Test; import org.reactivestreams.Subscriber; @@ -36,7 +39,7 @@ public void testTimeoutSocket() { RSocket timeout = RSockets.timeout(Duration.ofMillis(50)).apply(socket); timeout - .requestResponse(PayloadImpl.EMPTY) + .requestResponse(EmptyPayload.INSTANCE) .subscribe( new Subscriber() { @Override 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 7a1d79e0e..0126b07c4 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java +++ b/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java @@ -19,7 +19,7 @@ import static org.junit.Assert.assertEquals; import io.rsocket.Payload; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Ignore; import org.junit.Rule; @@ -47,7 +47,7 @@ public void testFireNForget10() { public void testPushMetadata10() { long outputCount = Flux.range(1, 10) - .flatMap(i -> setup.getRSocket().metadataPush(new PayloadImpl("", "metadata"))) + .flatMap(i -> setup.getRSocket().metadataPush(DefaultPayload.create("", "metadata"))) .doOnError(Throwable::printStackTrace) .count() .block(); @@ -94,7 +94,7 @@ private Payload testPayload(int metadataPresent) { metadata = "metadata"; break; } - return new PayloadImpl("hello", metadata); + return DefaultPayload.create("hello", metadata); } @Test(timeout = 10000) @@ -208,7 +208,7 @@ public void testChannel3() { @Test(timeout = 10000) public void testChannel512() { - Flux payloads = Flux.range(1, 512).map(i -> new PayloadImpl("hello " + i)); + Flux payloads = Flux.range(1, 512).map(i -> DefaultPayload.create("hello " + i)); long count = setup.getRSocket().requestChannel(payloads).count().block(); @@ -217,7 +217,7 @@ public void testChannel512() { @Test(timeout = 30000) public void testChannel20_000() { - Flux payloads = Flux.range(1, 20_000).map(i -> new PayloadImpl("hello " + i)); + Flux payloads = Flux.range(1, 20_000).map(i -> DefaultPayload.create("hello " + i)); long count = setup.getRSocket().requestChannel(payloads).count().block(); @@ -226,7 +226,7 @@ public void testChannel20_000() { @Test(timeout = 60_000) public void testChannel200_000() { - Flux payloads = Flux.range(1, 200_000).map(i -> new PayloadImpl("hello " + i)); + Flux payloads = Flux.range(1, 200_000).map(i -> DefaultPayload.create("hello " + i)); long count = setup.getRSocket().requestChannel(payloads).count().block(); @@ -238,7 +238,7 @@ public void testChannel200_000() { public void testChannel2_000_000() { AtomicInteger counter = new AtomicInteger(0); - Flux payloads = Flux.range(1, 2_000_000).map(i -> new PayloadImpl("hello " + i)); + Flux payloads = Flux.range(1, 2_000_000).map(i -> DefaultPayload.create("hello " + i)); long count = setup.getRSocket().requestChannel(payloads).count().block(); assertEquals(2_000_000, count); 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 2d66fef6c..08ce2cc22 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/PingClient.java +++ b/rsocket-test/src/main/java/io/rsocket/test/PingClient.java @@ -18,7 +18,8 @@ import io.rsocket.Payload; import io.rsocket.RSocket; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.ByteBufPayload; + import java.time.Duration; import org.HdrHistogram.Recorder; import reactor.core.publisher.Flux; @@ -31,7 +32,7 @@ public class PingClient { public PingClient(Mono client) { this.client = client; - this.payload = new PayloadImpl("hello"); + this.payload = ByteBufPayload.create("hello"); } public Recorder startTracker(Duration interval) { @@ -58,7 +59,8 @@ public Flux startPingPong(int count, final Recorder histogram) { i -> { long start = System.nanoTime(); return rsocket - .requestResponse(payload) + .requestResponse(payload.retain()) + .doOnNext(Payload::release) .doFinally( signalType -> { long diff = System.nanoTime() - start; diff --git a/rsocket-test/src/main/java/io/rsocket/test/PingHandler.java b/rsocket-test/src/main/java/io/rsocket/test/PingHandler.java index f349a978c..38cdf6972 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/PingHandler.java +++ b/rsocket-test/src/main/java/io/rsocket/test/PingHandler.java @@ -21,7 +21,8 @@ import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.SocketAcceptor; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.ByteBufPayload; + import java.util.concurrent.ThreadLocalRandom; import reactor.core.publisher.Mono; @@ -32,20 +33,22 @@ public class PingHandler implements SocketAcceptor { public PingHandler() { byte[] data = new byte[1024]; ThreadLocalRandom.current().nextBytes(data); - pong = new PayloadImpl(data); + pong = ByteBufPayload.create(data); } public PingHandler(byte[] data) { - pong = new PayloadImpl(data); + pong = ByteBufPayload.create(data); } @Override public Mono accept(ConnectionSetupPayload setup, RSocket sendingSocket) { + setup.release(); return Mono.just( new AbstractRSocket() { @Override public Mono requestResponse(Payload payload) { - return Mono.just(pong); + payload.release(); + return Mono.just(pong.retain()); } }); } diff --git a/rsocket-test/src/main/java/io/rsocket/test/TestRSocket.java b/rsocket-test/src/main/java/io/rsocket/test/TestRSocket.java index 5f75400e8..a8287bd83 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/TestRSocket.java +++ b/rsocket-test/src/main/java/io/rsocket/test/TestRSocket.java @@ -18,7 +18,7 @@ import io.rsocket.AbstractRSocket; import io.rsocket.Payload; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -27,7 +27,7 @@ public class TestRSocket extends AbstractRSocket { @Override public Mono requestResponse(Payload payload) { - return Mono.just(new PayloadImpl("hello world", "metadata")); + return Mono.just(DefaultPayload.create("hello world", "metadata")); } @Override @@ -48,6 +48,6 @@ public Mono fireAndForget(Payload payload) { @Override public Flux requestChannel(Publisher payloads) { // TODO is defensive copy neccesary? - return Flux.from(payloads).map(p -> new PayloadImpl(p.getDataUtf8(), p.getMetadataUtf8())); + return Flux.from(payloads).map(DefaultPayload::create); } } diff --git a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/ClientServerTest.java b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/ClientServerTest.java index d0383a1b1..289284a17 100644 --- a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/ClientServerTest.java +++ b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/ClientServerTest.java @@ -19,7 +19,7 @@ import io.rsocket.Payload; import io.rsocket.test.ClientSetupRule; -import io.rsocket.util.PayloadImpl; +import io.rsocket.util.DefaultPayload; import org.junit.Ignore; import org.junit.Rule; import org.junit.Test; @@ -34,7 +34,7 @@ public class ClientServerTest { public void testFireNForget10() { long outputCount = Flux.range(1, 10) - .flatMap(i -> setup.getRSocket().fireAndForget(new PayloadImpl("hello", "metadata"))) + .flatMap(i -> setup.getRSocket().fireAndForget(DefaultPayload.create("hello", "metadata"))) .doOnError(Throwable::printStackTrace) .count() .block(); @@ -46,7 +46,7 @@ public void testFireNForget10() { public void testPushMetadata10() { long outputCount = Flux.range(1, 10) - .flatMap(i -> setup.getRSocket().metadataPush(new PayloadImpl("", "metadata"))) + .flatMap(i -> setup.getRSocket().metadataPush(DefaultPayload.create("", "metadata"))) .doOnError(Throwable::printStackTrace) .count() .block(); @@ -62,7 +62,7 @@ public void testRequestResponse1() { i -> setup .getRSocket() - .requestResponse(new PayloadImpl("hello", "metadata")) + .requestResponse(DefaultPayload.create("hello", "metadata")) .map(Payload::getDataUtf8)) .doOnError(Throwable::printStackTrace) .count() @@ -79,7 +79,7 @@ public void testRequestResponse10() { i -> setup .getRSocket() - .requestResponse(new PayloadImpl("hello", "metadata")) + .requestResponse(DefaultPayload.create("hello", "metadata")) .map(Payload::getDataUtf8)) .doOnError(Throwable::printStackTrace) .count() @@ -96,7 +96,7 @@ public void testRequestResponse100() { i -> setup .getRSocket() - .requestResponse(new PayloadImpl("hello", "metadata")) + .requestResponse(DefaultPayload.create("hello", "metadata")) .map(Payload::getDataUtf8)) .doOnError(Throwable::printStackTrace) .count() @@ -113,7 +113,7 @@ public void testRequestResponse10_000() { i -> setup .getRSocket() - .requestResponse(new PayloadImpl("hello", "metadata")) + .requestResponse(DefaultPayload.create("hello", "metadata")) .map(Payload::getDataUtf8)) .doOnError(Throwable::printStackTrace) .count() @@ -125,7 +125,7 @@ public void testRequestResponse10_000() { @Test(timeout = 10000) public void testRequestStream() { Flux publisher = - setup.getRSocket().requestStream(new PayloadImpl("hello", "metadata")); + setup.getRSocket().requestStream(DefaultPayload.create("hello", "metadata")); long count = publisher.take(5).count().block(); diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPing.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPing.java index 8663b8352..3f0122fd5 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPing.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPing.java @@ -15,11 +15,13 @@ */ package io.rsocket.transport.netty; +import io.rsocket.Frame; import io.rsocket.RSocket; import io.rsocket.RSocketFactory; import io.rsocket.test.PingClient; import io.rsocket.transport.netty.client.TcpClientTransport; import java.time.Duration; + import org.HdrHistogram.Recorder; import reactor.core.publisher.Mono; @@ -27,7 +29,10 @@ public final class TcpPing { public static void main(String... args) { Mono client = - RSocketFactory.connect().transport(TcpClientTransport.create(7878)).start(); + RSocketFactory.connect() + .frameDecoder(Frame::retain) + .transport(TcpClientTransport.create(7878)) + .start(); PingClient pingClient = new PingClient(client); Recorder recorder = pingClient.startTracker(Duration.ofSeconds(1)); diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPongServer.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPongServer.java index 5afbff908..2d701f462 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPongServer.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/TcpPongServer.java @@ -15,6 +15,7 @@ */ package io.rsocket.transport.netty; +import io.rsocket.Frame; import io.rsocket.RSocketFactory; import io.rsocket.test.PingHandler; import io.rsocket.transport.netty.server.TcpServerTransport; @@ -23,6 +24,7 @@ public final class TcpPongServer { public static void main(String... args) { RSocketFactory.receive() + .frameDecoder(Frame::retain) .acceptor(new PingHandler()) .transport(TcpServerTransport.create(7878)) .start()