diff --git a/build.gradle b/build.gradle index 9bf0d5cfc..25ccef03a 100644 --- a/build.gradle +++ b/build.gradle @@ -30,6 +30,10 @@ description = 'RSocket: stream oriented messaging passing with Reactive Stream s buildScan { licenseAgreementUrl = 'https://gradle.com/terms-of-service'; licenseAgree = 'yes' } +googleJavaFormat { + toolVersion = '1.4' +} + subprojects { apply plugin: 'java' apply plugin: 'maven' diff --git a/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java b/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java index 1eefc3452..40e35e3e5 100644 --- a/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java +++ b/rsocket-core/src/jmh/java/io/rsocket/FragmentationPerf.java @@ -21,7 +21,7 @@ @Fork( value = 1, jvmArgsAppend = {"-XX:+UnlockCommercialFeatures", "-XX:+FlightRecorder"} -) //, "-Dio.netty.leakDetection.level=advanced"}) +) // , "-Dio.netty.leakDetection.level=advanced"}) @Warmup(iterations = 10) @Measurement(iterations = 10_000) @State(Scope.Thread) @@ -84,7 +84,7 @@ public void smallFragmentationFrameReassembler(Input input) { Frame frame = smallFragmentAssembler.reassemble(); input.bh.consume(frame); frame.release(); - //input.smallFragmentAssembler.clear(); + // input.smallFragmentAssembler.clear(); } private static ByteBuffer createRandomBytes(int size) { diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java index 60db853af..90d0778c5 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java @@ -79,7 +79,8 @@ class RSocketClient implements RSocket { this.receivers = new IntObjectHashMap<>(256, 0.9f); this.missedAckCounter = new AtomicInteger(); - // DO NOT Change the order here. The Send processor must be subscribed to before receiving connections + // DO NOT Change the order here. The Send processor must be subscribed to before receiving + // connections this.sendProcessor = EmitterProcessor.create(); if (!Duration.ZERO.equals(tickPeriod)) { diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java index 72db5f8d6..d6138aace 100644 --- a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java @@ -56,7 +56,8 @@ class RSocketServer implements RSocket { this.sendingSubscriptions = new IntObjectHashMap<>(); this.channelProcessors = new IntObjectHashMap<>(); - // DO NOT Change the order here. The Send processor must be subscribed to before receiving connections + // DO NOT Change the order here. The Send processor must be subscribed to before receiving + // connections this.sendProcessor = EmitterProcessor.create(); connection diff --git a/rsocket-core/src/main/java/io/rsocket/frame/FrameHeaderFlyweight.java b/rsocket-core/src/main/java/io/rsocket/frame/FrameHeaderFlyweight.java index 957487910..94412e8fc 100644 --- a/rsocket-core/src/main/java/io/rsocket/frame/FrameHeaderFlyweight.java +++ b/rsocket-core/src/main/java/io/rsocket/frame/FrameHeaderFlyweight.java @@ -285,7 +285,8 @@ public static void encodeLength(final ByteBuf byteBuf, final int offset, final i if ((length & ~FRAME_LENGTH_MASK) != 0) { throw new IllegalArgumentException("Length is larger than 24 bits"); } - // Write each byte separately in reverse order, this mean we can write 1 << 23 without overflowing. + // Write each byte separately in reverse order, this mean we can write 1 << 23 without + // overflowing. byteBuf.setByte(offset, length >> 16); byteBuf.setByte(offset + 1, length >> 8); byteBuf.setByte(offset + 2, length); diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumeUtil.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumeUtil.java index 95d84e6ed..093120357 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ResumeUtil.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ResumeUtil.java @@ -11,7 +11,7 @@ public static boolean isTracked(FrameType frameType) { case REQUEST_STREAM: case REQUEST_RESPONSE: case FIRE_AND_FORGET: - //case METADATA_PUSH: + // case METADATA_PUSH: case REQUEST_N: case CANCEL: case ERROR: diff --git a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/client/JavaClientDriver.java b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/client/JavaClientDriver.java index 8eb295720..acdef3d01 100644 --- a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/client/JavaClientDriver.java +++ b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/client/JavaClientDriver.java @@ -151,7 +151,8 @@ public void runTest(TckClientTest test) { handleCancel(args); break; default: - // the default behavior is to just skip the line, so we can accommodate slight changes to the TCK + // the default behavior is to just skip the line, so we can accommodate slight changes to + // the TCK break; } } @@ -227,12 +228,14 @@ private void handleChannel(String[] args, Iterator iter, String name, bo // set the initial payload Payload initialPayload = new PayloadImpl(args[2], args[3]); - // this is the subscriber that will request data from the server, like all the other test subscribers + // this is the subscriber that will request data from the server, like all the other test + // subscribers MySubscriber testsub = new MySubscriber<>(1L, AGENT); CountDownLatch c = new CountDownLatch(1); // we now create the publisher that the server will subscribe to with its own subscriber - // we want to give that subscriber a subscription that the client will use to send data to the server + // we want to give that subscriber a subscription that the client will use to send data to the + // server RSocket client = getClient(args[0]); AtomicReference mypct = new AtomicReference<>(); Publisher pub = @@ -455,13 +458,13 @@ private void handleCancel(String[] args) { private void handleEOF() { MySubscriber fnfsub = new MySubscriber<>(0L, AGENT); - //if (clientMap.size() > 0) { + // if (clientMap.size() > 0) { // // Use any Client to send shutdown msg to the server // RSocket fnfclient = clientMap.get(clientMap.keySet().toArray()[0]); // Publisher fnfpub = fnfclient.fireAndForget(new PayloadImpl("shutdown", "shutdown")); // fnfpub.subscribe(fnfsub); // fnfsub.request(1); - //} + // } } /** A subscription for channel, it handles request(n) by sort of faking an initial payload. */ diff --git a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/ConsoleUtils.java b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/ConsoleUtils.java index 63173bd4e..001c74299 100644 --- a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/ConsoleUtils.java +++ b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/ConsoleUtils.java @@ -19,11 +19,11 @@ /** This class handles everything that gets printed to the console */ public class ConsoleUtils { - private static final String ANSI_RESET = ""; //"\u001B[0m"; - private static final String ANSI_RED = ""; //\u001B[31m"; - private static final String ANSI_GREEN = ""; //\u001B[32m"; - private static final String ANSI_CYAN = ""; //\u001B[36m"; - private static final String ANSI_BLUE = ""; //\u001B[34m"; + private static final String ANSI_RESET = ""; // "\u001B[0m"; + private static final String ANSI_RED = ""; // \u001B[31m"; + private static final String ANSI_GREEN = ""; // \u001B[32m"; + private static final String ANSI_CYAN = ""; // \u001B[36m"; + private static final String ANSI_BLUE = ""; // \u001B[34m"; private String agent; private final boolean debugEnabled = false; diff --git a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/MySubscriber.java b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/MySubscriber.java index eb12f70b5..5eaebde24 100644 --- a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/MySubscriber.java +++ b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/MySubscriber.java @@ -1,4 +1,3 @@ - /* * Copyright 2016 Netflix, Inc. * @@ -165,7 +164,8 @@ private void myPass(String message) { consoleUtils.info("PASSED: " + message); } - // there might be a race condition with take, so this behavior is defined as: either wait until we have received n + // there might be a race condition with take, so this behavior is defined as: either wait until we + // have received n // myValues and then cancel, or cancel if we already have n myValues public final void take(long n) { if (values.size() >= n) { diff --git a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/server/JavaServerDriver.java b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/server/JavaServerDriver.java index 3d62c49e0..5d11865a8 100644 --- a/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/server/JavaServerDriver.java +++ b/rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/server/JavaServerDriver.java @@ -36,7 +36,8 @@ public class JavaServerDriver { private Map, String> requestResponseMarbles = new HashMap<>(); private Map, String> requestStreamMarbles = new HashMap<>(); private Map, String> requestSubscriptionMarbles = new HashMap<>(); - // channel doesn't have an initial payload, but maybe the first payload sent can be viewed as the "initial" + // channel doesn't have an initial payload, but maybe the first payload sent can be viewed as the + // "initial" private Map, List> requestChannelCommands = new HashMap<>(); private Set> requestChannelFail = new HashSet<>(); private Set> requestEchoChannel = new HashSet<>(); @@ -56,7 +57,8 @@ public Flux requestChannel(Publisher payloads) { try { MySubscriber sub = new MySubscriber<>(0L, "[SERVER]"); payloads.subscribe(sub); - // want to get equivalent of "initial payload" so we can route behavior, this might change in the future + // want to get equivalent of "initial payload" so we can route behavior, this + // might change in the future sub.request(1); sub.awaitAtLeast(1); Tuple initpayload = @@ -93,7 +95,8 @@ public Flux requestChannel(Publisher payloads) { s.onSubscribe(echoSubscription); sub.setEcho(echoSubscription); sub.request( - 10000); // request a large number, which basically means the client can send whatever + 10000); // request a large number, which basically means the client can + // send whatever } } catch (Exception e) { assertNull("interrupted ", e.getMessage()); diff --git a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/SingleThreadedEventLoop.java b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/SingleThreadedEventLoop.java index 0a2b1c2bc..ba8b8e24d 100644 --- a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/SingleThreadedEventLoop.java +++ b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/SingleThreadedEventLoop.java @@ -76,7 +76,7 @@ public void run() { while (true) { try { int count = drain(); - //if (count > 100) { + // if (count > 100) { // System.out.println(name + " drained..." + count); // } idleStrategy.idle(count); diff --git a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/AckConnectDecoder.java b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/AckConnectDecoder.java index d3691a63d..26c3b6651 100644 --- a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/AckConnectDecoder.java +++ b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/AckConnectDecoder.java @@ -176,13 +176,29 @@ public StringBuilder appendTo(final StringBuilder builder) { } builder.append(BLOCK_LENGTH); builder.append("):"); - //Token{signal=BEGIN_FIELD, name='channelId', description='The AeronChannel id', id=1, version=0, encodedLength=0, offset=0, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int64', description='The AeronChannel id', id=-1, version=0, encodedLength=8, offset=0, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT64, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='channelId', description='The AeronChannel id', id=1, + // version=0, encodedLength=0, offset=0, componentTokenCount=3, + // encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int64', description='The AeronChannel id', id=-1, version=0, + // encodedLength=8, offset=0, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, + // primitiveType=INT64, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, + // constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, + // semanticType='null'}} builder.append("channelId="); builder.append(channelId()); builder.append('|'); - //Token{signal=BEGIN_FIELD, name='serverSessionId', description='The session id for the server publication', id=2, version=0, encodedLength=0, offset=8, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int32', description='The session id for the server publication', id=-1, version=0, encodedLength=4, offset=8, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='serverSessionId', description='The session id for the server + // publication', id=2, version=0, encodedLength=0, offset=8, componentTokenCount=3, + // encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int32', description='The session id for the server publication', + // id=-1, version=0, encodedLength=4, offset=8, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("serverSessionId="); builder.append(serverSessionId()); diff --git a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/ConnectDecoder.java b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/ConnectDecoder.java index 423e8bd0d..79927a8f1 100644 --- a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/ConnectDecoder.java +++ b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/ConnectDecoder.java @@ -465,35 +465,79 @@ public StringBuilder appendTo(final StringBuilder builder) { } builder.append(BLOCK_LENGTH); builder.append("):"); - //Token{signal=BEGIN_FIELD, name='channelId', description='The AeronChannel id', id=1, version=0, encodedLength=0, offset=0, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int64', description='The AeronChannel id', id=-1, version=0, encodedLength=8, offset=0, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT64, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='channelId', description='The AeronChannel id', id=1, + // version=0, encodedLength=0, offset=0, componentTokenCount=3, + // encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int64', description='The AeronChannel id', id=-1, version=0, + // encodedLength=8, offset=0, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, + // primitiveType=INT64, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, + // constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, + // semanticType='null'}} builder.append("channelId="); builder.append(channelId()); builder.append('|'); - //Token{signal=BEGIN_FIELD, name='sendingStreamId', description='The stream id the connecting client will send traffic on', id=2, version=0, encodedLength=0, offset=8, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int32', description='The stream id the connecting client will send traffic on', id=-1, version=0, encodedLength=4, offset=8, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='sendingStreamId', description='The stream id the connecting + // client will send traffic on', id=2, version=0, encodedLength=0, offset=8, + // componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, + // byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, + // characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int32', description='The stream id the connecting client will + // send traffic on', id=-1, version=0, encodedLength=4, offset=8, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("sendingStreamId="); builder.append(sendingStreamId()); builder.append('|'); - //Token{signal=BEGIN_FIELD, name='receivingStreamId', description='The stream id the connecting client will receive data on', id=3, version=0, encodedLength=0, offset=12, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int32', description='The stream id the connecting client will receive data on', id=-1, version=0, encodedLength=4, offset=12, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='receivingStreamId', description='The stream id the connecting + // client will receive data on', id=3, version=0, encodedLength=0, offset=12, + // componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, + // byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, + // characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int32', description='The stream id the connecting client will + // receive data on', id=-1, version=0, encodedLength=4, offset=12, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("receivingStreamId="); builder.append(receivingStreamId()); builder.append('|'); - //Token{signal=BEGIN_FIELD, name='clientSessionId', description='The session id for the client publication', id=4, version=0, encodedLength=0, offset=16, componentTokenCount=3, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} - //Token{signal=ENCODING, name='int32', description='The session id for the client publication', id=-1, version=0, encodedLength=4, offset=16, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_FIELD, name='clientSessionId', description='The session id for the client + // publication', id=4, version=0, encodedLength=0, offset=16, componentTokenCount=3, + // encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='int32', description='The session id for the client publication', + // id=-1, version=0, encodedLength=4, offset=16, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=INT32, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("clientSessionId="); builder.append(clientSessionId()); builder.append('|'); - //Token{signal=BEGIN_VAR_DATA, name='sendingChannel', description='The Aeron channel the client will send data on', id=5, version=0, encodedLength=0, offset=20, componentTokenCount=6, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_VAR_DATA, name='sendingChannel', description='The Aeron channel the client + // will send data on', id=5, version=0, encodedLength=0, offset=20, componentTokenCount=6, + // encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("sendingChannel="); builder.append(sendingChannel()); builder.append('|'); - //Token{signal=BEGIN_VAR_DATA, name='receivingChannel', description='The Aeron channel the client will receive data on', id=6, version=0, encodedLength=0, offset=-1, componentTokenCount=6, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_VAR_DATA, name='receivingChannel', description='The Aeron channel the + // client will receive data on', id=6, version=0, encodedLength=0, offset=-1, + // componentTokenCount=6, encoding=Encoding{presence=REQUIRED, primitiveType=null, + // byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, + // characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("receivingChannel="); builder.append(receivingChannel()); builder.append('|'); - //Token{signal=BEGIN_VAR_DATA, name='clientManagementChannel', description='The channel the client listens for management data on', id=6, version=0, encodedLength=0, offset=-1, componentTokenCount=6, encoding=Encoding{presence=REQUIRED, primitiveType=null, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=BEGIN_VAR_DATA, name='clientManagementChannel', description='The channel the + // client listens for management data on', id=6, version=0, encodedLength=0, offset=-1, + // componentTokenCount=6, encoding=Encoding{presence=REQUIRED, primitiveType=null, + // byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, + // characterEncoding='null', epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("clientManagementChannel="); builder.append(clientManagementChannel()); diff --git a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/VarDataEncodingDecoder.java b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/VarDataEncodingDecoder.java index 7381bd45a..7eb0d9328 100644 --- a/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/VarDataEncodingDecoder.java +++ b/rsocket-transport-aeron/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/VarDataEncodingDecoder.java @@ -73,11 +73,19 @@ public String toString() { public StringBuilder appendTo(final StringBuilder builder) { builder.append('('); - //Token{signal=ENCODING, name='length', description='The channel the client listens for management data on', id=-1, version=0, encodedLength=4, offset=0, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=UINT32, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=1073741824, nullValue=null, constValue=null, characterEncoding='UTF-8', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='length', description='The channel the client listens for + // management data on', id=-1, version=0, encodedLength=4, offset=0, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=UINT32, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=1073741824, nullValue=null, constValue=null, + // characterEncoding='UTF-8', epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append("length="); builder.append(length()); builder.append('|'); - //Token{signal=ENCODING, name='varData', description='The channel the client listens for management data on', id=-1, version=0, encodedLength=-1, offset=4, componentTokenCount=1, encoding=Encoding{presence=REQUIRED, primitiveType=UINT8, byteOrder=LITTLE_ENDIAN, minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='UTF-8', epoch='unix', timeUnit=nanosecond, semanticType='null'}} + // Token{signal=ENCODING, name='varData', description='The channel the client listens for + // management data on', id=-1, version=0, encodedLength=-1, offset=4, componentTokenCount=1, + // encoding=Encoding{presence=REQUIRED, primitiveType=UINT8, byteOrder=LITTLE_ENDIAN, + // minValue=null, maxValue=null, nullValue=null, constValue=null, characterEncoding='UTF-8', + // epoch='unix', timeUnit=nanosecond, semanticType='null'}} builder.append(')'); return builder; diff --git a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelPongServer.java b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelPongServer.java index f70c93443..0ef151519 100644 --- a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelPongServer.java +++ b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelPongServer.java @@ -35,7 +35,7 @@ public static void main(String... args) { AeronChannelServer.AeronChannelConsumer consumer = aeronChannel -> { Flux receive = aeronChannel.receive(); - //.doOnNext(b -> System.out.println("server got => " + b.getInt(0))); + // .doOnNext(b -> System.out.println("server got => " + b.getInt(0))); aeronChannel.send(receive).doOnError(Throwable::printStackTrace).subscribe(); }; diff --git a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelTest.java b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelTest.java index 563a37397..9c9f7ace1 100644 --- a/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelTest.java +++ b/rsocket-transport-aeron/src/test/java/io/rsocket/aeron/internal/reactivestreams/AeronChannelTest.java @@ -129,8 +129,9 @@ public void testPing() { Flux unsafeBufferObservable = Flux.range(1, count) - //.doOnNext(i -> LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(50))) - // .doOnNext(i -> System.out.println(Thread.currentThread() + " => client sending => " + i)) + // .doOnNext(i -> LockSupport.parkNanos(TimeUnit.MICROSECONDS.toNanos(50))) + // .doOnNext(i -> System.out.println(Thread.currentThread() + " => client sending => " + + // i)) .map( i -> { UnsafeBuffer buffer = new UnsafeBuffer(new byte[BitUtil.SIZE_OF_INT]); @@ -236,7 +237,7 @@ private void pingPong(int count) { // System.out.println("\nGOT AERON ERROR => \n [" + writer.toString() + "]\n\n"); });*/ - //ctx.driverTimeoutMs(Integer.MAX_VALUE); + // ctx.driverTimeoutMs(Integer.MAX_VALUE); Aeron aeron = Aeron.connect(ctx); Subscription serverSubscription = @@ -282,7 +283,7 @@ private void pingPong(int count) { serverChannel .receive() // latch.countDown(); - //System.out.println("received -> " + f.getInt(0)); + // System.out.println("received -> " + f.getInt(0)); .flatMap(serverChannel::send, 32) .doOnError(Throwable::printStackTrace) .subscribe(); @@ -316,12 +317,12 @@ private void pingPong(int count) { ThreadLocalRandom.current().nextBytes(bytes); Flux.range(1, count) - //.doOnRequest(l -> System.out.println("requested => " + l)) + // .doOnRequest(l -> System.out.println("requested => " + l)) .flatMap( i -> { - //System.out.println("Sending -> " + i); + // System.out.println("Sending -> " + i); - //UnsafeBuffer b = new UnsafeBuffer(new byte[BitUtil.SIZE_OF_INT]); + // UnsafeBuffer b = new UnsafeBuffer(new byte[BitUtil.SIZE_OF_INT]); UnsafeBuffer b = new UnsafeBuffer(bytes); b.putInt(0, i);