Feature/routing (#55)

Author: OlegDokuka

gradle.properties

@@ -2,7 +2,7 @@ group=io.rsocket.rpc
 version=0.3.0
 
 reactorBomVersion=Dysprosium-RELEASE
-rsocketVersion=1.0.0-RC6-SNAPSHOT
+rsocketVersion=1.0.0-RC6-bugfix-prioritization-SNAPSHOT
 graphqlVersion=11.0
 protobufVersion=3.7.1
 log4jVersion=2.12.1

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

@@ -17,61 +17,31 @@
 
 import io.opentracing.Tracer;
 import io.rsocket.Payload;
-import io.rsocket.ipc.util.IPCChannelFunction;
-import io.rsocket.ipc.util.IPCFunction;
-import java.util.Map;
+import io.rsocket.ipc.routing.SimpleRouter;
 import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 @SuppressWarnings("unchecked")
 class IPCServerRSocket extends RoutingServerRSocket implements IPCRSocket, SelfRegistrable {
 
   private final String service;
+  private final SimpleRouter simpleRouter;
 
-  public IPCServerRSocket(
-      String service,
-      Map<String, IPCFunction<Mono<Void>>> fireAndForgetRegistry,
-      Map<String, IPCFunction<Mono<Payload>>> requestResponseRegistry,
-      Map<String, IPCFunction<Flux<Payload>>> requestStreamRegistry,
-      Map<String, IPCChannelFunction> requestChannelRegistry) {
-    super(
-        fireAndForgetRegistry,
-        requestResponseRegistry,
-        requestStreamRegistry,
-        requestChannelRegistry);
+  public IPCServerRSocket(String service, SimpleRouter simpleRouter) {
+    super(simpleRouter);
     this.service = service;
+    this.simpleRouter = simpleRouter;
   }
 
-  public IPCServerRSocket(
-      String service,
-      Tracer tracer,
-      Map<String, IPCFunction<Mono<Void>>> fireAndForgetRegistry,
-      Map<String, IPCFunction<Mono<Payload>>> requestResponseRegistry,
-      Map<String, IPCFunction<Flux<Payload>>> requestStreamRegistry,
-      Map<String, IPCChannelFunction> requestChannelRegistry) {
-    super(
-        tracer,
-        fireAndForgetRegistry,
-        requestResponseRegistry,
-        requestStreamRegistry,
-        requestChannelRegistry);
+  public IPCServerRSocket(String service, Tracer tracer, SimpleRouter simpleRouter) {
+    super(tracer, simpleRouter);
     this.service = service;
+    this.simpleRouter = simpleRouter;
   }
 
-  public IPCServerRSocket(
-      String service,
-      MetadataDecoder decoder,
-      Map<String, IPCFunction<Mono<Void>>> fireAndForgetRegistry,
-      Map<String, IPCFunction<Mono<Payload>>> requestResponseRegistry,
-      Map<String, IPCFunction<Flux<Payload>>> requestStreamRegistry,
-      Map<String, IPCChannelFunction> requestChannelRegistry) {
-    super(
-        decoder,
-        fireAndForgetRegistry,
-        requestResponseRegistry,
-        requestStreamRegistry,
-        requestChannelRegistry);
+  public IPCServerRSocket(String service, MetadataDecoder decoder, SimpleRouter simpleRouter) {
+    super(decoder, simpleRouter);
     this.service = service;
+    this.simpleRouter = simpleRouter;
   }
 
   @Override
@@ -80,15 +50,11 @@ public String getService() {
   }
 
   @Override
-  public void selfRegister(
-      Map<String, IPCFunction<Mono<Void>>> fireAndForgetRegistry,
-      Map<String, IPCFunction<Mono<Payload>>> requestResponseRegistry,
-      Map<String, IPCFunction<Flux<Payload>>> requestStreamRegistry,
-      Map<String, IPCChannelFunction> requestChannelRegistry) {
-    fireAndForgetRegistry.putAll(this.fireAndForgetRegistry);
-    requestResponseRegistry.putAll(this.requestResponseRegistry);
-    requestStreamRegistry.putAll(this.requestStreamRegistry);
-    requestChannelRegistry.putAll(this.requestChannelRegistry);
+  public void selfRegister(MutableRouter mutableRouter) {
+    simpleRouter.getFireAndForgetRegistry().forEach(mutableRouter::withFireAndForgetRoute);
+    simpleRouter.getRequestResponseRegistry().forEach(mutableRouter::withRequestResponseRoute);
+    simpleRouter.getRequestStreamRegistry().forEach(mutableRouter::withRequestStreamRoute);
+    simpleRouter.getRequestChannelRegistry().forEach(mutableRouter::withRequestChannelRoute);
   }
 
   @Override

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

@@ -1,16 +1,35 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.rsocket.ipc;
 
 import io.netty.buffer.ByteBuf;
 import io.opentracing.SpanContext;
-import io.rsocket.Payload;
 
 @FunctionalInterface
 public interface MetadataDecoder {
 
-  <RESULT> RESULT decode(Payload payload, Handler<RESULT> transformer) throws Exception;
+  Metadata decode(ByteBuf metadataByteBuf) throws Exception;
 
-  interface Handler<RESULT> {
-    RESULT handleAndReply(ByteBuf data, ByteBuf metadata, String route, SpanContext spanContext)
-        throws Exception;
+  interface Metadata {
+    ByteBuf metadata();
+
+    String route();
+
+    SpanContext spanContext();
+
+    boolean isComposite();
   }
 }

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

@@ -1,3 +1,18 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.rsocket.ipc;
 
 import io.netty.buffer.ByteBuf;

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

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.rsocket.ipc;
+
+import io.rsocket.Payload;
+import io.rsocket.ipc.util.IPCChannelFunction;
+import io.rsocket.ipc.util.IPCFunction;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface MutableRouter<SELF extends MutableRouter<SELF>> extends Router {
+  SELF withFireAndForgetRoute(String route, IPCFunction<Mono<Void>> function);
+
+  SELF withRequestResponseRoute(String route, IPCFunction<Mono<Payload>> function);
+
+  SELF withRequestStreamRoute(String route, IPCFunction<Flux<Payload>> function);
+
+  SELF withRequestChannelRoute(String route, IPCChannelFunction function);
+}

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

@@ -1,43 +1,40 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package io.rsocket.ipc;
 
 import io.opentracing.Tracer;
 import io.rsocket.ResponderRSocket;
-import java.util.concurrent.ConcurrentHashMap;
+import io.rsocket.ipc.routing.SimpleRouter;
 
 public class RequestHandlingRSocket extends RoutingServerRSocket implements ResponderRSocket {
 
   public RequestHandlingRSocket() {
-    super(
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>());
+    super(new SimpleRouter());
   }
 
   public RequestHandlingRSocket(Tracer tracer) {
-    super(
-        tracer,
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>());
+    super(tracer, new SimpleRouter());
   }
 
   public RequestHandlingRSocket(MetadataDecoder decoder) {
-    super(
-        decoder,
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>(),
-        new ConcurrentHashMap<>());
+    super(decoder, new SimpleRouter());
   }
 
   public RequestHandlingRSocket withEndpoint(SelfRegistrable selfRegistrable) {
-    selfRegistrable.selfRegister(
-        fireAndForgetRegistry,
-        requestResponseRegistry,
-        requestStreamRegistry,
-        requestChannelRegistry);
+    selfRegistrable.selfRegister((MutableRouter) router);
     return this;
   }
 }

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

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.rsocket.ipc;
+
+import io.rsocket.Payload;
+import io.rsocket.ipc.util.IPCChannelFunction;
+import io.rsocket.ipc.util.IPCFunction;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public interface Router {
+
+  IPCFunction<Mono<Void>> routeFireAndForget(String route);
+
+  IPCFunction<Mono<Payload>> routeRequestResponse(String route);
+
+  IPCFunction<Flux<Payload>> routeRequestStream(String route);
+
+  IPCChannelFunction routeRequestChannel(String route);
+}