Skip to content

utf8 helpers on payload #355

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yschimke merged 1 commit into from
Jul 17, 2017
Merged

utf8 helpers on payload #355

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions rsocket-core/src/main/java/io/rsocket/Frame.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
9 changes: 9 additions & 0 deletions rsocket-core/src/main/java/io/rsocket/Payload.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
}
2 changes: 1 addition & 1 deletion rsocket-core/src/main/java/io/rsocket/RSocketClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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: "
Expand Down
2 changes: 1 addition & 1 deletion rsocket-core/src/main/java/io/rsocket/exceptions/Exceptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
52 changes: 0 additions & 52 deletions rsocket-core/src/test/java/io/rsocket/TestUtil.java
View file
Open in desktop

This file was deleted.

5 changes: 2 additions & 3 deletions rsocket-core/src/test/java/io/rsocket/util/PayloadImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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));
}
}
4 changes: 2 additions & 2 deletions ...t-examples/src/main/java/io/rsocket/examples/transport/tcp/channel/ChannelEchoClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -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);
}
Expand Down
2 changes: 1 addition & 1 deletion rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/duplex/DuplexClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();

Expand Down
6 changes: 3 additions & 3 deletions ...les/src/main/java/io/rsocket/examples/transport/tcp/requestresponse/HelloWorldClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/StreamingClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
11 changes: 5 additions & 6 deletions rsocket-examples/src/test/java/io/rsocket/integration/TcpIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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());
}
};

Expand Down
19 changes: 4 additions & 15 deletions rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/common/MySubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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);
Expand All @@ -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));

Expand Down Expand Up @@ -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;
}

Expand Down
12 changes: 3 additions & 9 deletions rsocket-tck-drivers/src/main/java/io/rsocket/tckdrivers/server/JavaServerDriver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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 "
Expand All @@ -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());
Expand Down
12 changes: 4 additions & 8 deletions rsocket-test/src/main/java/io/rsocket/test/BaseClientServerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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();
Expand All @@ -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();
Expand All @@ -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();
Expand Down
12 changes: 4 additions & 8 deletions rsocket-transport-aeron/src/test/java/io/rsocket/aeron/ClientServerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand All @@ -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();
Expand All @@ -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();
Expand All @@ -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();
Expand Down