Support of heartbeat for websocket gateway (#864)

services-api/src/main/java/io/scalecube/services/gateway/Gateway.java

@@ -1,6 +1,8 @@
 package io.scalecube.services.gateway;
 
 import io.scalecube.services.Address;
+import io.scalecube.services.ServiceCall;
+import io.scalecube.services.registry.api.ServiceRegistry;
 
 public interface Gateway {
 
@@ -21,9 +23,11 @@ public interface Gateway {
  /**
  * Starts gateway.
  *
+ * @param call {@link ServiceCall} instance
+ * @param serviceRegistry {@link ServiceRegistry} instance
  * @return gateway instance
  */
- Gateway start();
+ Gateway start(ServiceCall call, ServiceRegistry serviceRegistry);
 
  /** Stops gateway. */
  void stop();

services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java

@@ -18,7 +18,7 @@ public interface ServiceTransport {
  /**
  * Provider for {@link ServerTransport}.
  *
- * @param serviceRegistry serviceRegistry
+ * @param serviceRegistry {@link ServiceRegistry} instance
  * @return {@code ServerTransport} instance
  */
  ServerTransport serverTransport(ServiceRegistry serviceRegistry);

services-gateway/src/main/java/io/scalecube/services/gateway/http/HttpGateway.java

@@ -4,20 +4,23 @@
 import io.netty.handler.codec.http.cors.CorsConfigBuilder;
 import io.netty.handler.codec.http.cors.CorsHandler;
 import io.scalecube.services.Address;
+import io.scalecube.services.ServiceCall;
 import io.scalecube.services.exceptions.DefaultErrorMapper;
 import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.gateway.Gateway;
-import io.scalecube.services.gateway.GatewayOptions;
+import io.scalecube.services.registry.api.ServiceRegistry;
 import java.net.InetSocketAddress;
-import java.util.StringJoiner;
 import java.util.function.Consumer;
+import java.util.function.Function;
 import reactor.netty.DisposableServer;
 import reactor.netty.http.server.HttpServer;
 import reactor.netty.resources.LoopResources;
 
 public class HttpGateway implements Gateway {
 
- private final GatewayOptions options;
+ private final String id;
+ private final int port;
+ private final Function<ServiceCall, ServiceCall> callFactory;
  private final ServiceProviderErrorMapper errorMapper;
  private final boolean corsEnabled;
  private final CorsConfigBuilder corsConfigBuilder;
@@ -26,28 +29,35 @@ public class HttpGateway implements Gateway {
  private LoopResources loopResources;
 
  private HttpGateway(Builder builder) {
- this.options = builder.options;
+ this.id = builder.id;
+ this.port = builder.port;
+ this.callFactory = builder.callFactory;
  this.errorMapper = builder.errorMapper;
  this.corsEnabled = builder.corsEnabled;
  this.corsConfigBuilder = builder.corsConfigBuilder;
  }
 
  @Override
  public String id() {
- return options.id();
+ return id;
  }
 
  @Override
- public Gateway start() {
- HttpGatewayAcceptor gatewayAcceptor = new HttpGatewayAcceptor(options.call(), errorMapper);
-
+ public Gateway start(ServiceCall call, ServiceRegistry serviceRegistry) {
  loopResources =
- LoopResources.create(
- options.id() + ":" + options.port(), LoopResources.DEFAULT_IO_WORKER_COUNT, true);
+ LoopResources.create(id + ":" + port, LoopResources.DEFAULT_IO_WORKER_COUNT, true);
 
  try {
- prepareHttpServer(loopResources, options.port())
- .handle(gatewayAcceptor)
+ HttpServer.create()
+ .runOn(loopResources)
+ .bindAddress(() -> new InetSocketAddress(port))
+ .doOnConnection(
+ connection -> {
+ if (corsEnabled) {
+ connection.addHandlerLast(new CorsHandler(corsConfigBuilder.build()));
+ }
+ })
+ .handle(new HttpGatewayAcceptor(callFactory.apply(call), errorMapper))
  .bind()
  .doOnSuccess(server -> this.server = server)
  .toFuture()
@@ -59,23 +69,6 @@ public Gateway start() {
  return this;
  }
 
- private HttpServer prepareHttpServer(LoopResources loopResources, int port) {
- HttpServer httpServer = HttpServer.create();
-
- if (loopResources != null) {
- httpServer = httpServer.runOn(loopResources);
- }
-
- return httpServer
- .bindAddress(() -> new InetSocketAddress(port))
- .doOnConnection(
- connection -> {
- if (corsEnabled) {
- connection.addHandlerLast(new CorsHandler(corsConfigBuilder.build()));
- }
- });
- }
-
  @Override
  public Address address() {
  InetSocketAddress address = (InetSocketAddress) server.address();
@@ -100,21 +93,11 @@ private void shutdownLoopResources(LoopResources loopResources) {
  }
  }
 
- @Override
- public String toString() {
- return new StringJoiner(", ", HttpGateway.class.getSimpleName() + "[", "]")
- .add("options=" + options)
- .add("errorMapper=" + errorMapper)
- .add("corsEnabled=" + corsEnabled)
- .add("corsConfigBuilder=" + corsConfigBuilder)
- .add("server=" + server)
- .add("loopResources=" + loopResources)
- .toString();
- }
-
  public static class Builder {
 
- private GatewayOptions options;
+ private String id = "http@" + Integer.toHexString(hashCode());
+ private int port;
+ private Function<ServiceCall, ServiceCall> callFactory = call -> call;
  private ServiceProviderErrorMapper errorMapper = DefaultErrorMapper.INSTANCE;
  private boolean corsEnabled = false;
  private CorsConfigBuilder corsConfigBuilder =
@@ -125,12 +108,26 @@ public static class Builder {
 
  public Builder() {}
 
- public GatewayOptions options() {
- return options;
+ public String id() {
+ return id;
+ }
+
+ public Builder id(String id) {
+ this.id = id;
+ return this;
+ }
+
+ public int port() {
+ return port;
+ }
+
+ public Builder port(int port) {
+ this.port = port;
+ return this;
  }
 
- public Builder options(GatewayOptions options) {
- this.options = options;
+ public Builder serviceCall(Function<ServiceCall, ServiceCall> operator) {
+ callFactory = callFactory.andThen(operator);
  return this;
  }
 

services-gateway/src/main/java/io/scalecube/services/gateway/http/HttpGatewayAcceptor.java

@@ -14,7 +14,6 @@
 import io.scalecube.services.ServiceCall;
 import io.scalecube.services.api.ErrorData;
 import io.scalecube.services.api.ServiceMessage;
-import io.scalecube.services.exceptions.DefaultErrorMapper;
 import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.gateway.ReferenceCountUtil;
 import io.scalecube.services.transport.api.DataCodec;
@@ -37,10 +36,6 @@ public class HttpGatewayAcceptor
  private final ServiceCall serviceCall;
  private final ServiceProviderErrorMapper errorMapper;
 
- HttpGatewayAcceptor(ServiceCall serviceCall) {
- this(serviceCall, DefaultErrorMapper.INSTANCE);
- }
-
  HttpGatewayAcceptor(ServiceCall serviceCall, ServiceProviderErrorMapper errorMapper) {
  this.serviceCall = serviceCall;
  this.errorMapper = errorMapper;

services-gateway/src/main/java/io/scalecube/services/gateway/websocket/HeartbeatService.java

@@ -0,0 +1,14 @@
+package io.scalecube.services.gateway.websocket;
+
+import io.scalecube.services.annotations.Service;
+import io.scalecube.services.annotations.ServiceMethod;
+import reactor.core.publisher.Mono;
+
+@Service(HeartbeatService.NAMESPACE)
+public interface HeartbeatService {
+
+ String NAMESPACE = "v1/scalecube.websocket.heartbeat";
+
+ @ServiceMethod
+ Mono<Long> ping(long value);
+}

services-gateway/src/main/java/io/scalecube/services/gateway/websocket/HeartbeatServiceImpl.java

@@ -0,0 +1,11 @@
+package io.scalecube.services.gateway.websocket;
+
+import reactor.core.publisher.Mono;
+
+public class HeartbeatServiceImpl implements HeartbeatService {
+
+ @Override
+ public Mono<Long> ping(long value) {
+ return Mono.just(value);
+ }
+}