utf8 helpers on payload (rsocket#355)

yschimke · web-flow · commit b94540030012 · 2017-07-17T07:57:33.000-07:00
diff --git a/rsocket-core/src/main/java/io/rsocket/Frame.java b/rsocket-core/src/main/java/io/rsocket/Frame.java
@@ -266,6 +266,10 @@ public static Frame from(final ByteBuf content) {
     return frame;
   }
 
+  public String getDataUtf8() {
+    return StandardCharsets.UTF_8.decode(getData()).toString();
+  }
+
   /* TODO:
    *
    * fromRequest(type, id, payload)
diff --git a/rsocket-core/src/main/java/io/rsocket/Payload.java b/rsocket-core/src/main/java/io/rsocket/Payload.java
@@ -16,10 +16,19 @@
 package io.rsocket;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
 
 /** Payload of a {@link Frame}. */
 public interface Payload {
   ByteBuffer getMetadata();
 
   ByteBuffer getData();
+
+  default String getMetadataUtf8() {
+    return StandardCharsets.UTF_8.decode(getMetadata()).toString();
+  }
+
+  default String getDataUtf8() {
+    return StandardCharsets.UTF_8.decode(getData()).toString();
+  }
 }
diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketClient.java b/rsocket-core/src/main/java/io/rsocket/RSocketClient.java
@@ -472,7 +472,7 @@ private void handleMissingResponseProcessor(int streamId, FrameType type, Frame
       if (type == FrameType.ERROR) {
         // message for stream that has never existed, we have a problem with
         // the overall connection and must tear down
-        String errorMessage = StandardCharsets.UTF_8.decode(frame.getData()).toString();
+        String errorMessage = frame.getDataUtf8();
 
         throw new IllegalStateException(
             "Client received error for non-existent stream: "
diff --git a/rsocket-core/src/main/java/io/rsocket/exceptions/Exceptions.java b/rsocket-core/src/main/java/io/rsocket/exceptions/Exceptions.java
@@ -27,7 +27,7 @@ private Exceptions() {}
   public static RuntimeException from(Frame frame) {
     final int errorCode = Frame.Error.errorCode(frame);
 
-    String message = StandardCharsets.UTF_8.decode(frame.getData()).toString();
+    String message = frame.getDataUtf8();
     switch (errorCode) {
       case APPLICATION_ERROR:
         return new ApplicationException(message);
diff --git a/rsocket-core/src/test/java/io/rsocket/TestUtil.java b/rsocket-core/src/test/java/io/rsocket/TestUtil.java
diff --git a/rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java b/rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java
@@ -16,7 +16,6 @@
 import static org.hamcrest.MatcherAssert.*;
 import static org.hamcrest.Matchers.*;
 
-import io.rsocket.TestUtil;
 import org.junit.Test;
 
 public class PayloadImplTest {
@@ -39,8 +38,8 @@ public void testReuseWithExternalMark() throws Exception {
   }
 
   public void assertDataAndMetadata(PayloadImpl p) {
-    assertThat("Unexpected data.", TestUtil.byteToString(p.getData()), equalTo(DATA_VAL));
+    assertThat("Unexpected data.", p.getDataUtf8(), equalTo(DATA_VAL));
     assertThat(
-        "Unexpected metadata.", TestUtil.byteToString(p.getMetadata()), equalTo(METADATA_VAL));
+        "Unexpected metadata.", p.getMetadataUtf8(), equalTo(METADATA_VAL));
   }
 }
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
@@ -45,7 +45,7 @@ public static void main(String[] args) {
 
     socket
         .requestChannel(Flux.interval(Duration.ofMillis(1000)).map(i -> new PayloadImpl("Hello")))
-        .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+        .map(Payload::getDataUtf8)
         .doOnNext(System.out::println)
         .take(10)
         .thenEmpty(socket.close())
@@ -60,7 +60,7 @@ public Mono<RSocket> accept(ConnectionSetupPayload setupPayload, RSocket reactiv
             @Override
             public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
               return Flux.from(payloads)
-                  .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+                  .map(Payload::getDataUtf8)
                   .map(s -> "Echo: " + s)
                   .map(PayloadImpl::new);
             }
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
@@ -33,7 +33,7 @@ public static void main(String[] args) {
             (setup, reactiveSocket) -> {
               reactiveSocket
                   .requestStream(new PayloadImpl("Hello-Bidi"))
-                  .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+                  .map(Payload::getDataUtf8)
                   .log()
                   .subscribe();
 
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
@@ -58,21 +58,21 @@ public Mono<Payload> requestResponse(Payload p) {
 
     socket
         .requestResponse(new PayloadImpl("Hello"))
-        .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+        .map(Payload::getDataUtf8)
         .onErrorReturn("error")
         .doOnNext(System.out::println)
         .block();
 
     socket
         .requestResponse(new PayloadImpl("Hello"))
-        .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+        .map(Payload::getDataUtf8)
         .onErrorReturn("error")
         .doOnNext(System.out::println)
         .block();
 
     socket
         .requestResponse(new PayloadImpl("Hello"))
-        .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+        .map(Payload::getDataUtf8)
         .onErrorReturn("error")
         .doOnNext(System.out::println)
         .block();
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
@@ -39,7 +39,7 @@ public static void main(String[] args) {
 
     socket
         .requestStream(new PayloadImpl("Hello"))
-        .map(payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString())
+        .map(Payload::getDataUtf8)
         .doOnNext(System.out::println)
         .take(10)
         .thenEmpty(socket.close())
diff --git a/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java b/rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java
@@ -102,7 +102,7 @@ public Flux<Payload> requestStream(Payload payload) {
 
     Payload result = client.requestStream(new PayloadImpl("REQUEST", "META")).blockLast();
 
-    assertEquals("RESPONSE", StandardCharsets.UTF_8.decode(result.getData()).toString());
+    assertEquals("RESPONSE", result.getDataUtf8());
   }
 
   @Test(timeout = 5_000L)
@@ -119,7 +119,7 @@ public Flux<Payload> requestStream(Payload payload) {
 
     Payload result = client.requestStream(new PayloadImpl("REQUEST", "META")).blockFirst();
 
-    assertEquals("", StandardCharsets.UTF_8.decode(result.getData()).toString());
+    assertEquals("", result.getDataUtf8());
   }
 
   @Test(timeout = 5_000L)
@@ -152,8 +152,8 @@ public Mono<Payload> requestResponse(Payload payload) {
             .onErrorReturn(new PayloadImpl("ERROR"))
             .block();
 
-    assertEquals("ERROR", StandardCharsets.UTF_8.decode(response1.getData()).toString());
-    assertEquals("SUCCESS", StandardCharsets.UTF_8.decode(response2.getData()).toString());
+    assertEquals("ERROR", response1.getDataUtf8());
+    assertEquals("SUCCESS", response2.getDataUtf8());
   }
 
   @Test(timeout = 5_000L)
@@ -168,8 +168,7 @@ public void testTwoConcurrentStreams() throws InterruptedException {
         new AbstractRSocket() {
           @Override
           public Flux<Payload> requestStream(Payload payload) {
-            String s = StandardCharsets.UTF_8.decode(payload.getData()).toString();
-            return map.get(s);
+            return map.get(payload.getDataUtf8());
           }
         };
 
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
@@ -64,10 +64,7 @@ public void onSubscribe(Subscription s) {
   @Override
   public void onNext(T t) {
     Payload p = (Payload) t;
-    Tuple<String, String> tup =
-        new Tuple<>(
-            StandardCharsets.UTF_8.decode(p.getData()).toString(),
-            StandardCharsets.UTF_8.decode(p.getMetadata()).toString());
+    Tuple<String, String> tup = new Tuple<>(p.getDataUtf8(), p.getMetadataUtf8());
     consoleUtils.info("On NEXT got : " + tup.getK() + " " + tup.getV());
     if (isEcho) {
       echosub.add(tup);
@@ -129,9 +126,7 @@ public final void assertValues(List<Tuple<String, String>> values) {
     for (int i = 0; i < values.size(); i++) {
       Payload p = (Payload) this.values().get(i);
       Tuple<String, String> v =
-          new Tuple<>(
-              StandardCharsets.UTF_8.decode(p.getData()).toString(),
-              StandardCharsets.UTF_8.decode(p.getMetadata()).toString());
+          new Tuple<>(p.getDataUtf8(), p.getMetadataUtf8());
       Tuple<String, String> u = values.get(i);
       String msg = prefix + "Values at position %d differ; ";
       assertEquals(String.format(msg, i), u, v);
@@ -149,10 +144,7 @@ public final void assertValue(Tuple<String, String> value) {
     assertEquals(msg, 1, this.values.size());
 
     Payload p = (Payload) values().get(0);
-    Tuple<String, String> v =
-        new Tuple<>(
-            StandardCharsets.UTF_8.decode(p.getData()).toString(),
-            StandardCharsets.UTF_8.decode(p.getMetadata()).toString());
+    Tuple<String, String> v = new Tuple<>(p.getDataUtf8(), p.getMetadataUtf8());
     msg = prefix;
     assertEquals(msg, valueAndClass(value), valueAndClass(v));
 
@@ -202,10 +194,7 @@ public final void take(long n) {
   public Tuple<String, String> getElement(int n) {
     assert (n < values.size());
     Payload p = (Payload) values().get(n);
-    Tuple<String, String> tup =
-        new Tuple<>(
-            StandardCharsets.UTF_8.decode(p.getData()).toString(),
-            StandardCharsets.UTF_8.decode(p.getMetadata()).toString());
+    Tuple<String, String> tup = new Tuple<>(p.getDataUtf8(), p.getMetadataUtf8());
     return tup;
   }
 
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
@@ -165,9 +165,7 @@ public final Mono<Void> fireAndForget(Payload payload) {
               return Mono.from(
                   s -> {
                     Tuple<String, String> initialPayload =
-                        new Tuple<>(
-                            StandardCharsets.UTF_8.decode(payload.getData()).toString(),
-                            StandardCharsets.UTF_8.decode(payload.getMetadata()).toString());
+                        new Tuple<>(payload.getDataUtf8(), payload.getMetadataUtf8());
                     consoleUtils.initialPayload(
                         "Received firenforget "
                             + initialPayload.getK()
@@ -190,9 +188,7 @@ public Mono<Payload> requestResponse(Payload payload) {
               return Mono.from(
                   s -> {
                     Tuple<String, String> initialPayload =
-                        new Tuple<>(
-                            StandardCharsets.UTF_8.decode(payload.getData()).toString(),
-                            StandardCharsets.UTF_8.decode(payload.getMetadata()).toString());
+                        new Tuple<>(payload.getDataUtf8(), payload.getMetadataUtf8());
                     String marble = requestResponseMarbles.get(initialPayload);
                     consoleUtils.initialPayload(
                         "Received requestresponse "
@@ -213,9 +209,7 @@ public Flux<Payload> requestStream(Payload payload) {
               return Flux.from(
                   s -> {
                     Tuple<String, String> initialPayload =
-                        new Tuple<>(
-                            StandardCharsets.UTF_8.decode(payload.getData()).toString(),
-                            StandardCharsets.UTF_8.decode(payload.getMetadata()).toString());
+                        new Tuple<>(payload.getDataUtf8(), payload.getMetadataUtf8());
                     String marble = requestStreamMarbles.get(initialPayload);
                     consoleUtils.initialPayload(
                         "Received Stream " + initialPayload.getK() + " " + initialPayload.getV());
diff --git a/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java b/rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java
@@ -63,8 +63,7 @@ public void testRequestResponse1() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -81,8 +80,7 @@ public void testRequestResponse10() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -99,8 +97,7 @@ public void testRequestResponse100() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -117,8 +114,7 @@ public void testRequestResponse10_000() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
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
@@ -64,8 +64,7 @@ public void testRequestResponse1() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -82,8 +81,7 @@ public void testRequestResponse10() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -100,8 +98,7 @@ public void testRequestResponse100() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();
@@ -118,8 +115,7 @@ public void testRequestResponse10_000() {
                     setup
                         .getRSocket()
                         .requestResponse(new PayloadImpl("hello", "metadata"))
-                        .map(
-                            payload -> StandardCharsets.UTF_8.decode(payload.getData()).toString()))
+                        .map(Payload::getDataUtf8))
             .doOnError(Throwable::printStackTrace)
             .count()
             .block();

Original file line number	Diff line number	Diff line change
`@@ -266,6 +266,10 @@ public static Frame from(final ByteBuf content) {`
`266`	`266`	`return frame;`
`267`	`267`	`}`
`268`	`268`
	`269`	`+ public String getDataUtf8() {`
	`270`	`+ return StandardCharsets.UTF_8.decode(getData()).toString();`
	`271`	`+ }`
	`272`	`+`
`269`	`273`	`/* TODO:`
`270`	`274`	`*`
`271`	`275`	`* fromRequest(type, id, payload)`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@`
`16`	`16`	`import static org.hamcrest.MatcherAssert.*;`
`17`	`17`	`import static org.hamcrest.Matchers.*;`
`18`	`18`
`19`		`-import io.rsocket.TestUtil;`
`20`	`19`	`import org.junit.Test;`
`21`	`20`
`22`	`21`	`public class PayloadImplTest {`
`@@ -39,8 +38,8 @@ public void testReuseWithExternalMark() throws Exception {`
`39`	`38`	`}`
`40`	`39`
`41`	`40`	`public void assertDataAndMetadata(PayloadImpl p) {`
`42`		`- assertThat("Unexpected data.", TestUtil.byteToString(p.getData()), equalTo(DATA_VAL));`
	`41`	`+ assertThat("Unexpected data.", p.getDataUtf8(), equalTo(DATA_VAL));`
`43`	`42`	`assertThat(`
`44`		`- "Unexpected metadata.", TestUtil.byteToString(p.getMetadata()), equalTo(METADATA_VAL));`
	`43`	`+ "Unexpected metadata.", p.getMetadataUtf8(), equalTo(METADATA_VAL));`
`45`	`44`	`}`
`46`	`45`	`}`