Cleanup, more clean contract on dataFormat

Author: artem-v

services-api/src/main/java/io/scalecube/services/ServiceCall.java

@@ -45,7 +45,7 @@ public class ServiceCall {
   private Router router;
   private ServiceClientErrorMapper errorMapper = DefaultErrorMapper.INSTANCE;
   private Map<String, String> credentials = Collections.emptyMap();
-  private String contentType;
+  private String contentType = ServiceMessage.DEFAULT_DATA_FORMAT;
 
   public ServiceCall() {}
 
@@ -429,24 +429,21 @@ private Address addressLookup(ServiceMessage request) {
   private ServiceMessage toServiceMessage(MethodInfo methodInfo, Object request) {
     final ServiceMessage.Builder builder;
     final Object data;
-    final String dataFormat;
 
     if (request instanceof ServiceMessage) {
       ServiceMessage message = (ServiceMessage) request;
       builder = ServiceMessage.from(message);
       data = message.data();
-      dataFormat = Optional.ofNullable(message.dataFormat()).orElse(contentType);
     } else {
       builder = ServiceMessage.builder();
       data = request;
-      dataFormat = contentType;
     }
 
     return builder
         .qualifier(methodInfo.serviceName(), methodInfo.methodName())
         .headers(credentials)
         .data(data)
-        .dataFormat(dataFormat)
+        .dataFormat(contentType)
         .build();
   }
 
@@ -486,15 +483,15 @@ private Optional<Object> toStringOrEqualsOrHashCode(
   }
 
   private Function<Flux<ServiceMessage>, Flux<Object>> asFlux(boolean isReturnTypeServiceMessage) {
-    return flux -> isReturnTypeServiceMessage ? flux.cast(Object.class) : flux.map(msgToResp());
+    return flux -> isReturnTypeServiceMessage ? flux.cast(Object.class) : flux.map(toResponse());
   }
 
   private Function<Mono<ServiceMessage>, Mono<Object>> asMono(boolean isReturnTypeServiceMessage) {
-    return mono -> isReturnTypeServiceMessage ? mono.cast(Object.class) : mono.map(msgToResp());
+    return mono -> isReturnTypeServiceMessage ? mono.cast(Object.class) : mono.map(toResponse());
   }
 
-  private Function<ServiceMessage, Object> msgToResp() {
-    return sm -> sm.hasData() ? sm.data() : UNEXPECTED_EMPTY_RESPONSE;
+  private Function<ServiceMessage, Object> toResponse() {
+    return message -> message.hasData() ? message.data() : UNEXPECTED_EMPTY_RESPONSE;
   }
 
   private ServiceMessage throwIfError(ServiceMessage message) {

services-api/src/main/java/io/scalecube/services/api/ServiceMessage.java

@@ -9,26 +9,20 @@
 
 public final class ServiceMessage {
 
-  /** Default message data content type. */
-  static final String DEFAULT_DATA_FORMAT = "application/json";
+  /** Default data format. */
+  public static final String DEFAULT_DATA_FORMAT = "application/json";
 
-  /**
-   * This header is supposed to be used by application in case if same data type can be reused for
-   * several messages so it will allow to qualify the specific message type.
-   */
-  static final String HEADER_QUALIFIER = "q";
+  /** Qualifier header. */
+  public static final String HEADER_QUALIFIER = "q";
 
   /**
    * This is a system header which used by transport for serialization and deserialization purpose.
    * It is not supposed to be used by application directly and it is subject to changes in future
    * releases.
    */
-  static final String HEADER_DATA_TYPE = "_type";
+  public static final String HEADER_DATA_TYPE = "_type";
 
-  /**
-   * Data format header. Json, Protostuff and etc. Note that default data format is defined at
-   * {@link #DEFAULT_DATA_FORMAT}.
-   */
+  /** Data format header. */
   public static final String HEADER_DATA_FORMAT = "_data_format";
 
   private Map<String, String> headers = new HashMap<>(1);

services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java

@@ -70,7 +70,7 @@ public ServiceMethodInvoker(
   public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
     return Mono.deferWithContext(context -> authenticate(message, context))
         .flatMap(authData -> deferWithContextOne(message, authData))
-        .map(response -> toResponse(response, message.qualifier(), message.dataFormatOrDefault()))
+        .map(response -> toResponse(response, message))
         .onErrorResume(throwable -> Mono.just(errorMapper.toMessage(throwable)));
   }
 
@@ -83,7 +83,7 @@ public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
   public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
     return Mono.deferWithContext(context -> authenticate(message, context))
         .flatMapMany(authData -> deferWithContextMany(message, authData))
-        .map(response -> toResponse(response, message.qualifier(), message.dataFormatOrDefault()))
+        .map(response -> toResponse(response, message))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
   }
 
@@ -99,12 +99,7 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
             (first, messages) ->
                 Mono.deferWithContext(context -> authenticate(first.get(), context))
                     .flatMapMany(authData -> deferWithContextBidirectional(messages, authData))
-                    .map(
-                        response ->
-                            toResponse(
-                                response,
-                                first.get().qualifier(),
-                                first.get().dataFormatOrDefault())))
+                    .map(response -> toResponse(response, first.get())))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
   }
 
@@ -192,7 +187,7 @@ private Object toRequest(ServiceMessage message) {
     return methodInfo.isRequestTypeServiceMessage() ? request : request.data();
   }
 
-  private ServiceMessage toResponse(Object response, String qualifier, String dataFormat) {
+  private ServiceMessage toResponse(Object response, ServiceMessage request) {
     final ServiceMessage.Builder builder;
     final Object data;
 
@@ -205,7 +200,11 @@ private ServiceMessage toResponse(Object response, String qualifier, String data
       data = response;
     }
 
-    return builder.qualifier(qualifier).data(data).dataFormat(dataFormat).build();
+    return builder
+        .qualifier(request.qualifier())
+        .data(data)
+        .dataFormat(request.dataFormatOrDefault())
+        .build();
   }
 
   private Context newPrincipalContext(Object authData, Context context) {

services/src/main/java/io/scalecube/services/Microservices.java

@@ -1,6 +1,7 @@
 package io.scalecube.services;
 
 import io.scalecube.net.Address;
+import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.auth.Authenticator;
 import io.scalecube.services.auth.PrincipalMapper;
 import io.scalecube.services.discovery.api.ServiceDiscovery;
@@ -28,6 +29,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.StringJoiner;
 import java.util.UUID;
@@ -320,7 +322,7 @@ public static final class Builder {
     private ServiceMessageDataDecoder dataDecoder =
         Optional.ofNullable(ServiceMessageDataDecoder.INSTANCE)
             .orElse((message, dataType) -> message);
-    private String contentType = "application/json";
+    private String contentType = ServiceMessage.DEFAULT_DATA_FORMAT;
     private PrincipalMapper<Object, Object> principalMapper = authData -> authData;
 
     public Mono<Microservices> start() {
@@ -377,11 +379,9 @@ public Builder methodRegistry(ServiceMethodRegistry methodRegistry) {
      * @param authenticator authenticator
      * @return this builder with applied parameter
      */
-    @SuppressWarnings("unchecked")
     @Deprecated
     public <T> Builder authenticator(Authenticator<? extends T> authenticator) {
-      this.authenticator = (Authenticator<Object>) authenticator;
-      return this;
+      return defaultAuthenticator(authenticator);
     }
 
     public Builder discovery(Function<ServiceEndpoint, ServiceDiscovery> factory) {
@@ -405,7 +405,7 @@ public Builder gateway(Function<GatewayOptions, Gateway> factory) {
     }
 
     /**
-     * Setter for {@code errorMapper}.
+     * Setter for default {@code errorMapper}.
      *
      * @param errorMapper error mapper
      * @return this builder with applied parameter
@@ -416,7 +416,7 @@ public Builder defaultErrorMapper(ServiceProviderErrorMapper errorMapper) {
     }
 
     /**
-     * Setter for {@code dataDecoder}.
+     * Setter for default {@code dataDecoder}.
      *
      * @param dataDecoder data decoder
      * @return this builder with applied parameter
@@ -429,23 +429,22 @@ public Builder defaultDataDecoder(ServiceMessageDataDecoder dataDecoder) {
     /**
      * Setter for default {@code contentType}. Deprecated. Use {@link #defaultContentType(String)}.
      *
-     * @param contentType contentType
+     * @param contentType contentType; not null
      * @return this builder with applied parameter
      */
     @Deprecated
     public Builder contentType(String contentType) {
-      this.contentType = contentType;
-      return this;
+      return defaultContentType(contentType);
     }
 
     /**
      * Setter for default {@code contentType}.
      *
-     * @param contentType contentType
+     * @param contentType contentType; not null
      * @return this builder with applied parameter
      */
     public Builder defaultContentType(String contentType) {
-      this.contentType = contentType;
+      this.contentType = Objects.requireNonNull(contentType, "default contentType");
       return this;
     }