diff --git a/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/DefaultHeadersCodec.java b/services-api/src/main/java/io/scalecube/services/transport/api/JdkCodec.java similarity index 70% rename from services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/DefaultHeadersCodec.java rename to services-api/src/main/java/io/scalecube/services/transport/api/JdkCodec.java index c7782d338..9d574a646 100644 --- a/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/DefaultHeadersCodec.java +++ b/services-api/src/main/java/io/scalecube/services/transport/api/JdkCodec.java @@ -4,14 +4,17 @@ import java.io.IOException; import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.io.OutputStream; +import java.lang.reflect.Type; import java.util.Collections; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; -/** Simple binary codec for headers service message. */ -public final class DefaultHeadersCodec implements HeadersCodec { +/** Simple headers and data codec based on JDK only. */ +public class JdkCodec implements DataCodec, HeadersCodec { /** * {@inheritDoc} @@ -21,6 +24,19 @@ public String contentType() { return "application/octet-stream"; } + /** + * Uses Jdk Object Serialization. + * + *

{@inheritDoc} + */ + @Override + public void encode(OutputStream stream, Object value) throws IOException { + try (ObjectOutputStream oos = new ObjectOutputStream(stream)) { + oos.writeObject(value); + oos.flush(); + } + } + /** * {@inheritDoc} */ @@ -40,6 +56,20 @@ public void encode(OutputStream stream, Map headers) throws IOEx } } + /** + * Uses Jdk Object Serialization. + * + *

{@inheritDoc} + */ + @Override + public Object decode(InputStream stream, Type type) throws IOException { + try (ObjectInputStream is = new ObjectInputStream(stream)) { + return is.readObject(); + } catch (ClassNotFoundException e) { + throw new IOException(e.getMessage(), e); + } + } + /** * {@inheritDoc} */ diff --git a/services-api/src/main/resources/META-INF/services/io.scalecube.services.transport.api.DataCodec b/services-api/src/main/resources/META-INF/services/io.scalecube.services.transport.api.DataCodec new file mode 100644 index 000000000..a8885001a --- /dev/null +++ b/services-api/src/main/resources/META-INF/services/io.scalecube.services.transport.api.DataCodec @@ -0,0 +1 @@ +io.scalecube.services.transport.api.JdkCodec \ No newline at end of file diff --git a/services-api/src/test/java/io/scalecube/services/transport/api/JdkCodecTest.java b/services-api/src/test/java/io/scalecube/services/transport/api/JdkCodecTest.java new file mode 100644 index 000000000..cf4307ace --- /dev/null +++ b/services-api/src/test/java/io/scalecube/services/transport/api/JdkCodecTest.java @@ -0,0 +1,70 @@ +package io.scalecube.services.transport.api; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.Serializable; +import java.util.Arrays; +import java.util.Objects; +import java.util.stream.Stream; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; + +class JdkCodecTest { + + private final DataCodec codec = new JdkCodec(); + + @ParameterizedTest + @MethodSource("provider") + void test(Object body) throws IOException { + Object decoded = writeAndRead(body); + assertEquals(body, decoded); + } + + static Stream provider() { + return Stream.of("hello", Arrays.asList(1, 2, 3), new Greeting("joe")); + } + + private Object writeAndRead(Object body) throws IOException { + try (ByteArrayOutputStream os = new ByteArrayOutputStream()) { + codec.encode(os, body); + byte[] bytes = os.toByteArray(); + try (ByteArrayInputStream is = new ByteArrayInputStream(bytes)) { + return codec.decode(is, body.getClass()); + } + } + } + + static class Greeting implements Serializable { + + private final String name; + + Greeting(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Greeting greeting = (Greeting) o; + return Objects.equals(name, greeting.name); + } + + @Override + public int hashCode() { + return Objects.hash(name); + } + } + +} diff --git a/services-bytebuf-codec/src/test/java/io/scalecube/services/transport/api/DefaultHeadersCodecTest.java b/services-api/src/test/java/io/scalecube/services/transport/api/JdkHeadersCodecTest.java similarity index 94% rename from services-bytebuf-codec/src/test/java/io/scalecube/services/transport/api/DefaultHeadersCodecTest.java rename to services-api/src/test/java/io/scalecube/services/transport/api/JdkHeadersCodecTest.java index 00e07bd24..f0ab1f2ab 100644 --- a/services-bytebuf-codec/src/test/java/io/scalecube/services/transport/api/DefaultHeadersCodecTest.java +++ b/services-api/src/test/java/io/scalecube/services/transport/api/JdkHeadersCodecTest.java @@ -12,9 +12,9 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.MethodSource; -class DefaultHeadersCodecTest { +class JdkHeadersCodecTest { - private HeadersCodec codec = new DefaultHeadersCodec(); + private final HeadersCodec codec = new JdkCodec(); @ParameterizedTest @MethodSource("provider") diff --git a/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/ServiceMessageCodec.java b/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/ServiceMessageCodec.java index be100d9a5..ad0824d38 100644 --- a/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/ServiceMessageCodec.java +++ b/services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/ServiceMessageCodec.java @@ -26,9 +26,12 @@ public final class ServiceMessageCodec { private static final Logger LOGGER = LoggerFactory.getLogger(ServiceMessageCodec.class); + private static final HeadersCodec DEFAULT_HEADERS_CODEC = new JdkCodec(); + private final HeadersCodec headersCodec; private final Map dataCodecs; + /** Message codec with default Headers/Data Codecs. */ public ServiceMessageCodec() { this(null, null); @@ -44,12 +47,12 @@ public ServiceMessageCodec() { *

Default HeadersCodec is DefaultHeadersCodec. This is lightweight binary codec written on * vanilla java. * - * @param headersCodec codec for message headers. Default, {@link DefaultHeadersCodec} + * @param headersCodec codec for message headers. Default, {@link JdkCodec} * @param dataCodecs codecs for message body. Codec will select by Message Content Type. */ public ServiceMessageCodec( @Nullable HeadersCodec headersCodec, @Nullable Collection dataCodecs) { - this.headersCodec = headersCodec == null ? new DefaultHeadersCodec() : headersCodec; + this.headersCodec = headersCodec == null ? DEFAULT_HEADERS_CODEC : headersCodec; Map defaultCodecs = DataCodec.INSTANCES; if (dataCodecs == null) { this.dataCodecs = defaultCodecs; diff --git a/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/codecs/Example2.java b/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/codecs/Example2.java new file mode 100644 index 000000000..f7349af55 --- /dev/null +++ b/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/codecs/Example2.java @@ -0,0 +1,53 @@ +package io.scalecube.services.examples.codecs; + +import io.scalecube.net.Address; +import io.scalecube.services.Microservices; +import io.scalecube.services.discovery.ScalecubeServiceDiscovery; +import io.scalecube.services.examples.helloworld.service.GreetingServiceImpl; +import io.scalecube.services.examples.helloworld.service.api.GreetingsService; +import io.scalecube.services.transport.rsocket.RSocketServiceTransport; + +public class Example2 { + + public static final String CONTENT_TYPE = "application/octet-stream"; + + /** + * Start the example. + * + * @param args ignored + */ + public static void main(String[] args) { + // ScaleCube Node node with no members + Microservices seed = + Microservices.builder() + .discovery(ScalecubeServiceDiscovery::new) + .transport(RSocketServiceTransport::new) + .defaultContentType(CONTENT_TYPE) // need to send with non-default data format + .startAwait(); + + final Address seedAddress = seed.discovery().address(); + + // Construct a ScaleCube node which joins the cluster hosting the Greeting Service + Microservices ms = + Microservices.builder() + .discovery( + endpoint -> + new ScalecubeServiceDiscovery(endpoint) + .membership(cfg -> cfg.seedMembers(seedAddress))) + .transport(RSocketServiceTransport::new) + .services(new GreetingServiceImpl()) + .startAwait(); + + // Create service proxy + GreetingsService service = seed.call().api(GreetingsService.class); + + // Execute the services and subscribe to service events + service + .sayHello("joe") + .subscribe(consumer -> System.out.println(consumer.message())); + + seed.onShutdown().block(); + ms.onShutdown().block(); + } + +} diff --git a/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/service/api/Greeting.java b/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/service/api/Greeting.java index 6ba8db8d5..f0fab4a58 100644 --- a/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/service/api/Greeting.java +++ b/services-examples-parent/services-examples/src/main/java/io/scalecube/services/examples/helloworld/service/api/Greeting.java @@ -1,6 +1,8 @@ package io.scalecube.services.examples.helloworld.service.api; -public class Greeting { +import java.io.Serializable; + +public class Greeting implements Serializable { String message;