Feature/ipc refactoring (#49)

* adding IPC style interactions

* fix typo

* refactoring for composite metadata

Signed-off-by: Oleh Dokuka <shadowgun@i.ua>

* drafts refactored internals of RSocket IPC

* refactors class structure

* provides encoders and minor fixes

* provides refactoring of rpc code-generation module

* provides provides refactoring of client and blocking interfaces/stubs generator

* provides fixes for ipc client

* provides fixes for ipc client. Fixes a couple of ByteByf leaks

* rebases to master

Author: OlegDokuka

gradle.properties

@@ -1,2 +1,3 @@
+
 group=io.rsocket.rpc
 version=0.2.20-SNAPSHOT

rsocket-ipc-core/build.gradle

@@ -1,11 +1,10 @@
 plugins {
-    id 'io.spring.dependency-management' version '1.0.7.RELEASE'
+    id 'io.spring.dependency-management' version '1.0.8.RELEASE'
 }
 
 description = 'RSocket IPC Library'
 
 dependencies {
-    compile project (':rsocket-rpc-core')
     implementation 'org.slf4j:slf4j-api'
 
     api 'io.opentracing:opentracing-api'

rsocket-ipc-core/src/main/java/io/rsocket/ipc/Client.java

@@ -17,15 +17,17 @@
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
 import io.opentracing.Tracer;
 import io.rsocket.Payload;
 import io.rsocket.RSocket;
-import io.rsocket.rpc.frames.Metadata;
-import io.rsocket.rpc.metrics.Metrics;
-import io.rsocket.rpc.tracing.Tag;
-import io.rsocket.rpc.tracing.Tracing;
+import io.rsocket.ipc.encoders.CompositeMetadataEncoder;
+import io.rsocket.ipc.encoders.PlainMetadataEncoder;
+import io.rsocket.ipc.metrics.Metrics;
+import io.rsocket.ipc.tracing.SimpleSpanContext;
+import io.rsocket.ipc.tracing.Tag;
+import io.rsocket.ipc.tracing.Tracing;
 import io.rsocket.util.ByteBufPayload;
+import java.nio.charset.Charset;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
@@ -38,6 +40,7 @@
 public final class Client<I, O> {
 
   private final String service;
+  private final MetadataEncoder metadataEncoder;
   private final Marshaller<I> marshaller;
   private final Unmarshaller<O> unmarshaller;
   private final RSocket rSocket;
@@ -46,12 +49,14 @@ public final class Client<I, O> {
 
   private Client(
       final String service,
+      final MetadataEncoder metadataEncoder,
       final Marshaller marshaller,
       final Unmarshaller unmarshaller,
       final RSocket rSocket,
       final MeterRegistry meterRegistry,
       final Tracer tracer) {
     this.service = service;
+    this.metadataEncoder = metadataEncoder;
     this.marshaller = marshaller;
     this.unmarshaller = unmarshaller;
     this.rSocket = rSocket;
@@ -60,7 +65,15 @@ private Client(
   }
 
   public interface R {
-    M rsocket(RSocket rSocket);
+    E rsocket(RSocket rSocket);
+  }
+
+  public interface E {
+    M compositeMetadataEncoder();
+
+    M plainMetadataEncoder();
+
+    M customMetadataEncoder(MetadataEncoder encoder);
   }
 
   public interface M {
@@ -217,9 +230,10 @@ <X> Functions.FireAndForget<X> genericFireAndForget(String route, Marshaller<X>
         doFireAndForget(service, route, rSocket, marshaller, o, byteBuf, metrics, tracing);
   }
 
-  private static class Builder implements P, U, R, M, T {
+  private static class Builder implements P, U, E, R, M, T {
     private final String service;
     private Marshaller marshaller;
+    private MetadataEncoder encoder;
     private MeterRegistry meterRegistry;
     private Tracer tracer;
     private RSocket rSocket;
@@ -228,6 +242,24 @@ private Builder(String service) {
       this.service = service;
     }
 
+    @Override
+    public M compositeMetadataEncoder() {
+      this.encoder = new CompositeMetadataEncoder();
+      return this;
+    }
+
+    @Override
+    public M plainMetadataEncoder() {
+      this.encoder = new PlainMetadataEncoder(".", Charset.defaultCharset());
+      return this;
+    }
+
+    @Override
+    public M customMetadataEncoder(MetadataEncoder encoder) {
+      this.encoder = encoder;
+      return this;
+    }
+
     @Override
     public <I> U<I> marshall(Marshaller<I> marshaller) {
       this.marshaller = Objects.requireNonNull(marshaller);
@@ -237,11 +269,11 @@ public <I> U<I> marshall(Marshaller<I> marshaller) {
     @Override
     public Client unmarshall(Unmarshaller unmarshaller) {
       Objects.requireNonNull(unmarshaller);
-      return new Client(service, marshaller, unmarshaller, rSocket, meterRegistry, tracer);
+      return new Client(service, encoder, marshaller, unmarshaller, rSocket, meterRegistry, tracer);
     }
 
     @Override
-    public M rsocket(RSocket rSocket) {
+    public E rsocket(RSocket rSocket) {
       this.rSocket = Objects.requireNonNull(rSocket);
       return this;
     }
@@ -279,17 +311,23 @@ private <X> Mono<Void> doFireAndForget(
       Function<? super Publisher<Void>, ? extends Publisher<Void>> metrics,
       Function<Map<String, String>, Function<? super Publisher<Void>, ? extends Publisher<Void>>>
           tracing) {
-    try {
-      HashMap<String, String> map = new HashMap<>();
-      ByteBuf d = marshaller.apply(o);
-      ByteBuf t = Tracing.mapToByteBuf(ByteBufAllocator.DEFAULT, map);
-      ByteBuf m = Metadata.encode(ByteBufAllocator.DEFAULT, service, route, t, metadata);
-
-      Payload payload = ByteBufPayload.create(d, m);
-      return r.fireAndForget(payload).transform(metrics).transform(tracing.apply(map));
-    } catch (Throwable t) {
-      return Mono.error(t);
-    }
+    final HashMap<String, String> map = new HashMap<>();
+    return Mono.defer(
+            () -> {
+              try {
+                ByteBuf d = marshaller.apply(o);
+                ByteBuf m =
+                    metadataEncoder.encode(metadata, new SimpleSpanContext(map), service, route);
+                metadata.release();
+                Payload payload = ByteBufPayload.create(d, m);
+                return r.fireAndForget(payload);
+              } catch (Throwable t) {
+                metadata.release();
+                return Mono.error(t);
+              }
+            })
+        .transform(metrics)
+        .transform(tracing.apply(map));
   }
 
   private <X, Y> Mono<Y> doRequestResponse(
@@ -303,27 +341,31 @@ private <X, Y> Mono<Y> doRequestResponse(
       Function<? super Publisher<Y>, ? extends Publisher<Y>> metrics,
       Function<Map<String, String>, Function<? super Publisher<Y>, ? extends Publisher<Y>>>
           tracing) {
-    try {
-      HashMap<String, String> map = new HashMap<>();
-      ByteBuf d = marshaller.apply(o);
-      ByteBuf t = Tracing.mapToByteBuf(ByteBufAllocator.DEFAULT, map);
-      ByteBuf m = Metadata.encode(ByteBufAllocator.DEFAULT, service, route, t, metadata);
-
-      Payload payload = ByteBufPayload.create(d, m);
-      return r.requestResponse(payload)
-          .map(
-              p -> {
-                try {
-                  return unmarshaller.apply(p.sliceData());
-                } finally {
-                  p.release();
-                }
-              })
-          .transform(metrics)
-          .transform(tracing.apply(map));
-    } catch (Throwable t) {
-      return Mono.error(t);
-    }
+    final HashMap<String, String> map = new HashMap<>();
+    return Mono.defer(
+            () -> {
+              try {
+                ByteBuf d = marshaller.apply(o);
+                ByteBuf m =
+                    metadataEncoder.encode(metadata, new SimpleSpanContext(map), service, route);
+                metadata.release();
+                Payload payload = ByteBufPayload.create(d, m);
+                return r.requestResponse(payload);
+              } catch (Throwable t) {
+                metadata.release();
+                return Mono.error(t);
+              }
+            })
+        .map(
+            p -> {
+              try {
+                return unmarshaller.apply(p.sliceData());
+              } finally {
+                p.release();
+              }
+            })
+        .transform(metrics)
+        .transform(tracing.apply(map));
   }
 
   private <X, Y> Flux<Y> doRequestStream(
@@ -337,27 +379,31 @@ private <X, Y> Flux<Y> doRequestStream(
       Function<? super Publisher<Y>, ? extends Publisher<Y>> metrics,
       Function<Map<String, String>, Function<? super Publisher<Y>, ? extends Publisher<Y>>>
           tracing) {
-    try {
-      HashMap<String, String> map = new HashMap<>();
-      ByteBuf d = marshaller.apply(o);
-      ByteBuf t = Tracing.mapToByteBuf(ByteBufAllocator.DEFAULT, map);
-      ByteBuf m = Metadata.encode(ByteBufAllocator.DEFAULT, service, route, t, metadata);
-
-      Payload payload = ByteBufPayload.create(d, m);
-      return r.requestStream(payload)
-          .map(
-              p -> {
-                try {
-                  return unmarshaller.apply(p.sliceData());
-                } finally {
-                  p.release();
-                }
-              })
-          .transform(metrics)
-          .transform(tracing.apply(map));
-    } catch (Throwable t) {
-      return Flux.error(t);
-    }
+    final HashMap<String, String> map = new HashMap<>();
+    return Flux.defer(
+            () -> {
+              try {
+                ByteBuf d = marshaller.apply(o);
+                ByteBuf m =
+                    metadataEncoder.encode(metadata, new SimpleSpanContext(map), service, route);
+                metadata.release();
+                Payload payload = ByteBufPayload.create(d, m);
+                return r.requestStream(payload);
+              } catch (Throwable t) {
+                metadata.release();
+                return Flux.error(t);
+              }
+            })
+        .map(
+            p -> {
+              try {
+                return unmarshaller.apply(p.sliceData());
+              } finally {
+                p.release();
+              }
+            })
+        .transform(metrics)
+        .transform(tracing.apply(map));
   }
 
   private <X, Y> Flux<Y> doRequestChannel(
@@ -373,18 +419,28 @@ private <X, Y> Flux<Y> doRequestChannel(
           tracing) {
     try {
 
-      HashMap<String, String> map = new HashMap<>();
+      final HashMap<String, String> map = new HashMap<>();
 
       Flux<Payload> input =
           Flux.from(pub)
               .map(
-                  o -> {
-                    ByteBuf d = marshaller.apply(o);
-                    ByteBuf t = Tracing.mapToByteBuf(ByteBufAllocator.DEFAULT, map);
-                    ByteBuf m =
-                        Metadata.encode(ByteBufAllocator.DEFAULT, service, route, t, metadata);
-
-                    return ByteBufPayload.create(d, m);
+                  new Function<X, Payload>() {
+                    boolean first = true;
+
+                    @Override
+                    public Payload apply(X o) {
+                      ByteBuf d = marshaller.apply(o);
+                      if (first) {
+                        first = false;
+                        ByteBuf m =
+                            metadataEncoder.encode(
+                                metadata, new SimpleSpanContext(map), service, route);
+                        metadata.release();
+                        return ByteBufPayload.create(d, m);
+                      }
+
+                      return ByteBufPayload.create(d);
+                    }
                   });
 
       return r.requestChannel(input)

rsocket-ipc-core/src/main/java/io/rsocket/ipc/IPCRSocket.java

@@ -15,6 +15,17 @@
  */
 package io.rsocket.ipc;
 
-import io.rsocket.rpc.RSocketRpcService;
+import io.rsocket.Payload;
+import io.rsocket.ResponderRSocket;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 
-public interface IPCRSocket extends RSocketRpcService {}
+public interface IPCRSocket extends ResponderRSocket, SelfRegistrable {
+  String getService();
+
+  Flux<Payload> requestChannel(Payload payload, Flux<Payload> publisher);
+
+  default Flux<Payload> requestChannel(Payload payload, Publisher<Payload> payloads) {
+    return requestChannel(payload, Flux.from(payloads));
+  }
+}