rsocket · yschimke · Oct 7, 2017 · Oct 5, 2017 · Oct 5, 2017
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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<String> 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<Payload> 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<ParseChannelThread> mypct = new AtomicReference<>();
     Publisher<Payload> pub =
@@ -455,13 +458,13 @@ private void handleCancel(String[] args) {
 
   private void handleEOF() {
     MySubscriber<Void> 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<Void> 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
@@ -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
@@ -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
@@ -36,7 +36,8 @@ public class JavaServerDriver {
   private Map<Tuple<String, String>, String> requestResponseMarbles = new HashMap<>();
   private Map<Tuple<String, String>, String> requestStreamMarbles = new HashMap<>();
   private Map<Tuple<String, String>, 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<Tuple<String, String>, List<String>> requestChannelCommands = new HashMap<>();
   private Set<Tuple<String, String>> requestChannelFail = new HashSet<>();
   private Set<Tuple<String, String>> requestEchoChannel = new HashSet<>();
@@ -56,7 +57,8 @@ public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
                     try {
                       MySubscriber<Payload> 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<String, String> initpayload =
@@ -93,7 +95,8 @@ public Flux<Payload> requestChannel(Publisher<Payload> 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
@@ -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/...n/src/main/java/io/rsocket/aeron/internal/reactivestreams/messages/AckConnectDecoder.java b/...n/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());