drafts initial version of standard router API

Signed-off-by: Oleh Dokuka <[email protected]>

rsocket-router/build.gradle

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2015-2018 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.
+ */
+
+plugins {
+    id 'java-library'
+    id 'maven-publish'
+    id 'com.jfrog.artifactory'
+    id 'com.jfrog.bintray'
+}
+
+dependencies {
+    api project(':rsocket-core')
+
+    implementation 'org.slf4j:slf4j-api'
+
+    testImplementation project(':rsocket-test')
+    testImplementation 'org.junit.jupiter:junit-jupiter-api'
+    testImplementation 'org.junit.jupiter:junit-jupiter-params'
+    testImplementation 'org.mockito:mockito-core'
+    testImplementation 'org.assertj:assertj-core'
+    testImplementation 'io.projectreactor:reactor-test'
+
+    // TODO: Remove after JUnit5 migration
+    testCompileOnly 'junit:junit'
+    testImplementation 'org.hamcrest:hamcrest-library'
+    testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
+    testRuntimeOnly 'ch.qos.logback:logback-classic'
+}
+
+description = 'Transparent Load Balancer for RSocket'

rsocket-router/src/main/java/io/rsocket/router/CompositeMetadataRouteCodec.java

@@ -0,0 +1,61 @@
+/*
+ * 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.router;
+
+import io.netty.buffer.ByteBuf;
+import io.rsocket.frame.FrameType;
+import io.rsocket.metadata.CompositeMetadata;
+import io.rsocket.metadata.CompositeMetadata.Entry;
+import io.rsocket.metadata.CompositeMetadata.WellKnownMimeTypeEntry;
+import io.rsocket.metadata.RoutingMetadata;
+import io.rsocket.metadata.WellKnownMimeType;
+import reactor.util.annotation.Nullable;
+
+public class CompositeMetadataRouteCodec implements RouteCodec {
+
+  @Override
+  @Nullable
+  public Route decode(ByteBuf metadataByteBuf, FrameType requestType) {
+    final CompositeMetadata compositeMetadata = new CompositeMetadata(metadataByteBuf, false);
+
+    String route = null;
+    String mimeType = null;
+
+    for (Entry compositeMetadatum : compositeMetadata) {
+      if (compositeMetadatum instanceof WellKnownMimeTypeEntry) {
+        final WellKnownMimeTypeEntry wellKnownMimeTypeEntry = (WellKnownMimeTypeEntry) compositeMetadatum;
+        final WellKnownMimeType type = wellKnownMimeTypeEntry.getType();
+
+        if (type == WellKnownMimeType.MESSAGE_RSOCKET_ROUTING) {
+          final RoutingMetadata routingMetadata = new RoutingMetadata(compositeMetadatum.getContent());
+          for(String routeEntry : routingMetadata) {
+            route = routeEntry;
+            break;
+          }
+        } else if (type == WellKnownMimeType.MESSAGE_RSOCKET_MIMETYPE) {
+          // FIXME: once codecs are available
+          // mimeType = compositeMetadatum
+        }
+      }
+    }
+
+    if (route != null) {
+      return new Route(requestType, route, mimeType);
+    }
+
+    return null;
+  }
+}

rsocket-router/src/main/java/io/rsocket/router/HandlerFunction.java

@@ -0,0 +1,19 @@
+package io.rsocket.router;
+
+import io.rsocket.Payload;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.util.annotation.Nullable;
+
+public interface HandlerFunction {
+
+  Route route();
+
+  @SuppressWarnings("rawtypes")
+  default Publisher handle(Payload payload) {
+    return handle(payload, null);
+  }
+
+  @SuppressWarnings("rawtypes")
+  Publisher handle(Payload firstPayload, @Nullable Flux<Payload> payloads);
+}

rsocket-router/src/main/java/io/rsocket/router/ImmutableRoutingRSocket.java

@@ -0,0 +1,41 @@
+package io.rsocket.router;
+
+import java.util.HashMap;
+import java.util.Map;
+
+class ImmutableRoutingRSocket extends RoutingRSocket {
+
+  private final Map<Route, HandlerFunction> mapping;
+
+  ImmutableRoutingRSocket(Map<Route, HandlerFunction> mapping, RouteCodec routeCodec) {
+    super(routeCodec);
+    this.mapping = mapping;
+  }
+
+  @Override
+  protected HandlerFunction handlerFor(Route route) {
+    return mapping.get(route);
+  }
+
+  static final class ImmutableRouterBuilder implements RoutingRSocket.Builder<ImmutableRouterBuilder> {
+
+    final HashMap<Route, HandlerFunction> mapping = new HashMap<>();
+    final RouteCodec routeCodec;
+
+    ImmutableRouterBuilder(RouteCodec routeCodec) {
+      this.routeCodec = routeCodec;
+    }
+
+    @Override
+    public ImmutableRouterBuilder addHandler(HandlerFunction handler) {
+      this.mapping.put(handler.route(), handler);
+
+      return this;
+    }
+
+    @Override
+    public RoutingRSocket build() {
+      return new ImmutableRoutingRSocket(this.mapping, routeCodec);
+    }
+  }
+}

rsocket-router/src/main/java/io/rsocket/router/Route.java

@@ -0,0 +1,62 @@
+package io.rsocket.router;
+
+import io.rsocket.frame.FrameType;
+import reactor.util.annotation.Nullable;
+
+public final class Route {
+
+  final String route;
+  final String mimeType;
+  final FrameType requestType;
+
+  public Route(FrameType requestType, String route) {
+    this(requestType, route, null);
+  }
+
+  public Route(FrameType requestType, String route, @Nullable String mimeType) {
+    this.route = route;
+    this.mimeType = mimeType;
+    this.requestType = requestType;
+  }
+
+  public String route() {
+    return this.route;
+  }
+
+  @Nullable
+  public String mimeType() {
+    return this.mimeType;
+  }
+
+  public FrameType requestType() {
+    return requestType;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+
+    Route route1 = (Route) o;
+
+    if (!route.equals(route1.route)) {
+      return false;
+    }
+    if (mimeType != null ? !mimeType.equals(route1.mimeType) : route1.mimeType != null) {
+      return false;
+    }
+    return requestType == route1.requestType;
+  }
+
+  @Override
+  public int hashCode() {
+    int result = route.hashCode();
+    result = 31 * result + (mimeType != null ? mimeType.hashCode() : 0);
+    result = 31 * result + requestType.hashCode();
+    return result;
+  }
+}

rsocket-router/src/main/java/io/rsocket/router/RouteCodec.java

@@ -0,0 +1,26 @@
+/*
+ * 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.router;
+
+import io.netty.buffer.ByteBuf;
+import io.rsocket.frame.FrameType;
+import reactor.util.annotation.Nullable;
+
+public interface RouteCodec {
+
+  @Nullable
+  Route decode(ByteBuf metadataByteBuf, FrameType requestType);
+}

rsocket-router/src/main/java/io/rsocket/router/Router.java

@@ -0,0 +1,124 @@
+package io.rsocket.router;
+
+import io.rsocket.Payload;
+import io.rsocket.frame.FrameType;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.util.annotation.Nullable;
+
+public final class Router {
+
+  public static RequestSpec route(String route) {
+    return route(route, null);
+  }
+
+  public static RequestSpec route(String route, @Nullable String mimeType) {
+    return new RequestSpec(route, mimeType);
+  }
+
+  @SuppressWarnings("rawtypes")
+  static final class RequestSpec {
+    final String route;
+    final String mimeType;
+
+    RequestSpec(String route, @Nullable String mimeType) {
+      this.route = route;
+      this.mimeType = mimeType;
+    }
+
+    public HandlerFunction fireAndForget(Function<Payload, Mono<Void>> handler) {
+      final Route route = new Route(FrameType.REQUEST_FNF, this.route, mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          return handler.apply(firstPayload);
+        }
+      };
+    }
+
+    public HandlerFunction fireAndForget(Consumer<Payload> handler) {
+      final Route route = new Route(FrameType.REQUEST_FNF, this.route, this.mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          handler.accept(firstPayload);
+          return Mono.empty();
+        }
+      };
+    }
+
+    public HandlerFunction requestResponse(Function<Payload, Mono<Payload>> handler) {
+      final Route route = new Route(FrameType.REQUEST_RESPONSE, this.route, this.mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          return handler.apply(firstPayload);
+        }
+      };
+    }
+
+    public HandlerFunction requestStream(Function<Payload, Flux<Payload>> handler) {
+      final Route route = new Route(FrameType.REQUEST_STREAM, this.route, this.mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          return handler.apply(firstPayload);
+        }
+      };
+    }
+
+    public HandlerFunction requestChannel(Function<Flux<Payload>, Flux<Payload>> handler) {
+      final Route route = new Route(FrameType.REQUEST_CHANNEL, this.route, this.mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          return handler.apply(payloads);
+        }
+      };
+    }
+
+    public HandlerFunction requestChannel(BiFunction<Payload, Flux<Payload>, Flux<Payload>> handler) {
+      final Route route = new Route(FrameType.REQUEST_CHANNEL, this.route, this.mimeType);
+      return new HandlerFunction() {
+        @Override
+        public Route route() {
+          return route;
+        }
+
+        @Override
+        public Publisher handle(Payload firstPayload, Flux<Payload> payloads) {
+          return handler.apply(firstPayload, payloads);
+        }
+      };
+    }
+  }
+}