Merge pull request #161 from scalecube/error-mapper-tests

Add ability use custom error mapper

Author: artem-v

services-gateway-client-transport/src/main/java/io/scalecube/services/gateway/transport/websocket/WebsocketGatewayClientSession.java

@@ -4,7 +4,6 @@
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import io.scalecube.services.api.ErrorData;
 import io.scalecube.services.api.ServiceMessage;
-import io.scalecube.services.exceptions.DefaultErrorMapper;
 import io.scalecube.services.gateway.transport.GatewayClientCodec;
 import io.scalecube.services.transport.api.ReferenceCountUtil;
 import java.nio.channels.ClosedChannelException;
@@ -168,10 +167,7 @@ private void handleResponse(
         if (signal == Signal.ERROR) {
           // decode error data to retrieve real error cause
           ServiceMessage errorMessage = codec.decodeData(response, ErrorData.class);
-          Throwable error = DefaultErrorMapper.INSTANCE.toError(errorMessage);
-          String sid = response.header(STREAM_ID);
-          LOGGER.error("Received error response: sid={}, error={}", sid, error);
-          onError.accept(error);
+          onNext.accept(errorMessage);
         }
       } else {
         // handle normal response

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

@@ -5,6 +5,8 @@
 import io.netty.handler.codec.http.cors.CorsConfigBuilder;
 import io.netty.handler.codec.http.cors.CorsHandler;
 import io.scalecube.net.Address;
+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.gateway.GatewayTemplate;
@@ -20,6 +22,8 @@
 
 public class HttpGateway extends GatewayTemplate {
 
+  private final ServiceProviderErrorMapper errorMapper;
+
   private DisposableServer server;
   private LoopResources loopResources;
 
@@ -31,7 +35,12 @@ public class HttpGateway extends GatewayTemplate {
           .allowedRequestMethods(HttpMethod.POST);
 
   public HttpGateway(GatewayOptions options) {
+    this(options, DefaultErrorMapper.INSTANCE);
+  }
+
+  public HttpGateway(GatewayOptions options, ServiceProviderErrorMapper errorMapper) {
     super(options);
+    this.errorMapper = errorMapper;
   }
 
   private HttpGateway(HttpGateway other) {
@@ -40,6 +49,7 @@ private HttpGateway(HttpGateway other) {
     this.loopResources = other.loopResources;
     this.corsEnabled = other.corsEnabled;
     this.corsConfigBuilder = copy(other.corsConfigBuilder);
+    this.errorMapper = other.errorMapper;
   }
 
   /**
@@ -108,7 +118,7 @@ private CorsConfigBuilder copy(CorsConfigBuilder other) {
   public Mono<Gateway> start() {
     return Mono.defer(
         () -> {
-          HttpGatewayAcceptor acceptor = new HttpGatewayAcceptor(options.call());
+          HttpGatewayAcceptor acceptor = new HttpGatewayAcceptor(options.call(), errorMapper);
 
           loopResources = LoopResources.create("http-gateway");
 

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

@@ -15,6 +15,7 @@
 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;
 import java.util.function.BiFunction;
@@ -34,9 +35,15 @@ public class HttpGatewayAcceptor
   private static final String ERROR_NAMESPACE = "io.scalecube.services.error";
 
   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;
   }
 
   @Override
@@ -58,8 +65,7 @@ public Publisher<Void> apply(HttpServerRequest httpRequest, HttpServerResponse h
         .switchIfEmpty(Mono.defer(() -> ByteBufMono.just(Unpooled.EMPTY_BUFFER)))
         .map(ByteBuf::retain)
         .flatMap(content -> handleRequest(content, httpRequest, httpResponse))
-        .onErrorResume(
-            t -> error(httpResponse, DefaultErrorMapper.INSTANCE.toMessage(ERROR_NAMESPACE, t)));
+        .onErrorResume(t -> error(httpResponse, errorMapper.toMessage(ERROR_NAMESPACE, t)));
   }
 
   private Mono<Void> handleRequest(

services-gateway-netty/src/main/java/io/scalecube/services/gateway/rsocket/RSocketGateway.java

@@ -5,6 +5,8 @@
 import io.rsocket.transport.netty.server.CloseableChannel;
 import io.rsocket.transport.netty.server.WebsocketServerTransport;
 import io.scalecube.net.Address;
+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.gateway.GatewaySessionHandler;
@@ -22,25 +24,45 @@ public class RSocketGateway extends GatewayTemplate {
   private static final Logger LOGGER = LoggerFactory.getLogger(RSocketGateway.class);
 
   private final GatewaySessionHandler sessionHandler;
+  private final ServiceProviderErrorMapper errorMapper;
+
   private CloseableChannel server;
   private LoopResources loopResources;
 
   public RSocketGateway(GatewayOptions options) {
-    super(options);
-    this.sessionHandler = GatewaySessionHandler.DEFAULT_INSTANCE;
+    this(options, GatewaySessionHandler.DEFAULT_INSTANCE, DefaultErrorMapper.INSTANCE);
   }
 
   public RSocketGateway(GatewayOptions options, GatewaySessionHandler sessionHandler) {
+    this(options, sessionHandler, DefaultErrorMapper.INSTANCE);
+  }
+
+  public RSocketGateway(GatewayOptions options, ServiceProviderErrorMapper errorMapper) {
+    this(options, GatewaySessionHandler.DEFAULT_INSTANCE, errorMapper);
+  }
+
+  /**
+   * Constructor.
+   *
+   * @param options gateway options
+   * @param sessionHandler session handler
+   * @param errorMapper error mapper
+   */
+  public RSocketGateway(
+      GatewayOptions options,
+      GatewaySessionHandler sessionHandler,
+      ServiceProviderErrorMapper errorMapper) {
     super(options);
     this.sessionHandler = sessionHandler;
+    this.errorMapper = errorMapper;
   }
 
   @Override
   public Mono<Gateway> start() {
     return Mono.defer(
         () -> {
           RSocketGatewayAcceptor acceptor =
-              new RSocketGatewayAcceptor(options.call(), sessionHandler);
+              new RSocketGatewayAcceptor(options.call(), sessionHandler, errorMapper);
 
           loopResources = LoopResources.create("rsocket-gateway");
 

services-gateway-netty/src/main/java/io/scalecube/services/gateway/rsocket/RSocketGatewayAcceptor.java

@@ -4,6 +4,7 @@
 import io.rsocket.RSocket;
 import io.rsocket.SocketAcceptor;
 import io.scalecube.services.ServiceCall;
+import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.gateway.GatewaySessionHandler;
 import io.scalecube.services.gateway.ServiceMessageCodec;
 import io.scalecube.services.transport.api.HeadersCodec;
@@ -19,16 +20,22 @@ public class RSocketGatewayAcceptor implements SocketAcceptor {
 
   private final ServiceCall serviceCall;
   private final GatewaySessionHandler sessionHandler;
+  private final ServiceProviderErrorMapper errorMapper;
 
   /**
    * Creates new acceptor for RS gateway.
    *
    * @param serviceCall to call remote service
    * @param sessionHandler handler for session events
+   * @param errorMapper error mapper
    */
-  public RSocketGatewayAcceptor(ServiceCall serviceCall, GatewaySessionHandler sessionHandler) {
+  public RSocketGatewayAcceptor(
+      ServiceCall serviceCall,
+      GatewaySessionHandler sessionHandler,
+      ServiceProviderErrorMapper errorMapper) {
     this.serviceCall = serviceCall;
     this.sessionHandler = sessionHandler;
+    this.errorMapper = errorMapper;
   }
 
   @Override
@@ -43,7 +50,8 @@ public Mono<RSocket> accept(ConnectionSetupPayload setup, RSocket rsocket) {
             serviceCall,
             messageCodec,
             headers(messageCodec, setup),
-            (session, req) -> sessionHandler.mapMessage(session, req, Context.empty()));
+            (session, req) -> sessionHandler.mapMessage(session, req, Context.empty()),
+            errorMapper);
     sessionHandler.onSessionOpen(gatewaySession);
     rsocket
         .onClose()

services-gateway-netty/src/main/java/io/scalecube/services/gateway/rsocket/RSocketGatewaySession.java

@@ -5,7 +5,7 @@
 import io.rsocket.util.ByteBufPayload;
 import io.scalecube.services.ServiceCall;
 import io.scalecube.services.api.ServiceMessage;
-import io.scalecube.services.exceptions.DefaultErrorMapper;
+import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.gateway.GatewaySession;
 import io.scalecube.services.gateway.ReferenceCountUtil;
 import io.scalecube.services.gateway.ServiceMessageCodec;
@@ -31,23 +31,27 @@ public final class RSocketGatewaySession extends AbstractRSocket implements Gate
   private final long sessionId;
   private final BiFunction<GatewaySession, ServiceMessage, ServiceMessage> messageMapper;
   private final Map<String, String> headers;
+  private final ServiceProviderErrorMapper errorMapper;
 
   /**
    * Constructor for gateway rsocket.
    *
    * @param serviceCall service call coming from microservices.
    * @param messageCodec message messageCodec.
+   * @param errorMapper error mapper
    */
   public RSocketGatewaySession(
       ServiceCall serviceCall,
       ServiceMessageCodec messageCodec,
       Map<String, String> headers,
-      BiFunction<GatewaySession, ServiceMessage, ServiceMessage> messageMapper) {
+      BiFunction<GatewaySession, ServiceMessage, ServiceMessage> messageMapper,
+      ServiceProviderErrorMapper errorMapper) {
     this.serviceCall = serviceCall;
     this.messageCodec = messageCodec;
     this.messageMapper = messageMapper;
     this.sessionId = SESSION_ID_GENERATOR.incrementAndGet();
     this.headers = Collections.unmodifiableMap(new HashMap<>(headers));
+    this.errorMapper = errorMapper;
   }
 
   @Override
@@ -73,8 +77,7 @@ public Mono<Payload> requestResponse(Payload payload) {
           return serviceCall
               .requestOne(request)
               .doOnError(th -> releaseRequestOnError(request))
-              .onErrorResume(
-                  th -> Mono.just(DefaultErrorMapper.INSTANCE.toMessage(request.qualifier(), th)))
+              .onErrorResume(th -> Mono.just(errorMapper.toMessage(request.qualifier(), th)))
               .map(this::toPayload);
         });
   }
@@ -87,8 +90,7 @@ public Flux<Payload> requestStream(Payload payload) {
           return serviceCall
               .requestMany(request)
               .doOnError(th -> releaseRequestOnError(request))
-              .onErrorResume(
-                  th -> Mono.just(DefaultErrorMapper.INSTANCE.toMessage(request.qualifier(), th)))
+              .onErrorResume(th -> Mono.just(errorMapper.toMessage(request.qualifier(), th)))
               .map(this::toPayload);
         });
   }

services-gateway-netty/src/main/java/io/scalecube/services/gateway/ws/GatewayMessages.java

@@ -1,7 +1,7 @@
 package io.scalecube.services.gateway.ws;
 
 import io.scalecube.services.api.ServiceMessage;
-import io.scalecube.services.exceptions.DefaultErrorMapper;
+import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 
 public final class GatewayMessages {
 
@@ -34,19 +34,22 @@ public static ServiceMessage newCancelMessage(long sid, String qualifier) {
   /**
    * Returns error message by given arguments.
    *
-   * @param message request
+   * @param errorMapper error mapper
+   * @param request request
    * @param th cause
    * @return {@link ServiceMessage} instance as the error signal
    */
-  public static ServiceMessage newErrorMessage(ServiceMessage message, Throwable th) {
-    ServiceMessage.Builder builder =
-        ServiceMessage.from(DefaultErrorMapper.INSTANCE.toMessage(message.qualifier(), th))
-            .header(SIGNAL_FIELD, Signal.ERROR.code());
-    String sid = message.header(STREAM_ID_FIELD);
-    if (sid != null) {
-      builder.header(STREAM_ID_FIELD, sid);
+  public static ServiceMessage toErrorResponse(
+      ServiceProviderErrorMapper errorMapper, ServiceMessage request, Throwable th) {
+    ServiceMessage errorMessage = errorMapper.toMessage(request.qualifier(), th);
+    String sid = request.header(STREAM_ID_FIELD);
+    if (sid == null) {
+      return ServiceMessage.from(errorMessage).header(SIGNAL_FIELD, Signal.ERROR.code()).build();
     }
-    return builder.build();
+    return ServiceMessage.from(errorMessage)
+        .header(SIGNAL_FIELD, Signal.ERROR.code())
+        .header(STREAM_ID_FIELD, sid)
+        .build();
   }
 
   /**

services-gateway-netty/src/main/java/io/scalecube/services/gateway/ws/WebsocketContextException.java

@@ -21,11 +21,6 @@ public static WebsocketContextException badRequest(String errorMessage, ServiceM
         new io.scalecube.services.exceptions.BadRequestException(errorMessage), request, null);
   }
 
-  public static WebsocketContextException wrap(
-      Throwable th, ServiceMessage request, ServiceMessage response) {
-    return new WebsocketContextException(th, request, response);
-  }
-
   public ServiceMessage request() {
     return request;
   }