Fixed tests

Author: artem-v

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

@@ -435,11 +435,11 @@ private ServiceMessage toServiceMessage(MethodInfo methodInfo, Object request) {
       ServiceMessage message = (ServiceMessage) request;
       builder = ServiceMessage.from(message);
       data = message.data();
-      dataFormat = message.dataFormat(); // data format from message
+      dataFormat = Optional.ofNullable(message.dataFormat()).orElse(contentType);
     } else {
       builder = ServiceMessage.builder();
       data = request;
-      dataFormat = contentType; // data format from service call
+      dataFormat = contentType;
     }
 
     return builder

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

@@ -93,7 +93,7 @@ public Map<String, String> headers() {
    * @return the message header by given header name
    */
   public String header(String name) {
-    Objects.requireNonNull(name);
+    Objects.requireNonNull(name, "header name");
     return headers.get(name);
   }
 
@@ -201,64 +201,113 @@ public static class Builder {
 
     private Builder() {}
 
+    /**
+     * Setter for {@code data}.
+     *
+     * @param data data; optional
+     * @return this builder
+     */
     public Builder data(Object data) {
       this.data = data;
       return this;
     }
 
-    public Builder dataType(Class<?> data) {
-      headers.put(HEADER_DATA_TYPE, data.getName());
+    /**
+     * Setter for {@code dataType}.
+     *
+     * @deprecated in future releases will be dropped without replacement
+     * @param dataType data type; no null
+     * @return this builder
+     */
+    @Deprecated
+    public Builder dataType(Class<?> dataType) {
+      Objects.requireNonNull(dataType, "dataType");
+      headers.put(HEADER_DATA_TYPE, dataType.getName());
       return this;
     }
 
+    /**
+     * Setter for {@code dataFormat}.
+     *
+     * @param dataFormat data format; not null
+     * @return this builder
+     */
     public Builder dataFormat(String dataFormat) {
+      Objects.requireNonNull(dataFormat, "dataFormat");
       headers.put(HEADER_DATA_FORMAT, dataFormat);
       return this;
     }
 
+    /**
+     * Returns {@code headers}.
+     *
+     * @return headers
+     */
     private Map<String, String> headers() {
       return this.headers;
     }
 
+    /**
+     * Setter for {@code headers}.
+     *
+     * @param headers headers
+     * @return this builder
+     */
     public Builder headers(Map<String, String> headers) {
       headers.forEach(this::header);
       return this;
     }
 
     /**
-     * Sets a header key value pair.
+     * Setter for header key-value pair.
      *
-     * @param key key; not null
-     * @param value value; not null
-     * @return self
+     * @param key header name; not null
+     * @param value header value; not null
+     * @return this builder
      */
     public Builder header(String key, String value) {
-      Objects.requireNonNull(key);
-      Objects.requireNonNull(value);
+      Objects.requireNonNull(key, "header name");
+      Objects.requireNonNull(value, "header value");
       headers.put(key, value);
       return this;
     }
 
     /**
-     * Sets a header key value pair.
+     * Setter for header key-value pair.
      *
-     * @param key key; not null
-     * @param value value; not null
-     * @return self
+     * @param key header name; not null
+     * @param value header value; not null
+     * @return this builder
      */
     public Builder header(String key, Object value) {
-      Objects.requireNonNull(key);
-      Objects.requireNonNull(value);
+      Objects.requireNonNull(key, "header name");
+      Objects.requireNonNull(value, "header value");
       headers.put(key, value.toString());
       return this;
     }
 
+    /**
+     * Setter for {@code qualifier}.
+     *
+     * @param qualifier qualifier; not null
+     * @return this builder
+     */
     public Builder qualifier(String qualifier) {
-      return header(HEADER_QUALIFIER, qualifier);
+      return header(HEADER_QUALIFIER, Objects.requireNonNull(qualifier, "qualifier"));
     }
 
-    public Builder qualifier(String serviceName, String methodName) {
-      return qualifier(Qualifier.asString(serviceName, methodName));
+    /**
+     * Setter for {@code qualifier}.
+     *
+     * @param namespace namespace; not null
+     * @param action action; not null
+     * @return this builder
+     */
+    public Builder qualifier(String namespace, String action) {
+      return qualifier(
+          Qualifier.asString(
+              Objects.requireNonNull(namespace, "namespace"),
+              Objects.requireNonNull(action, "action")));
     }
 
     public ServiceMessage build() {

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.dataFormat()))
+        .map(response -> toResponse(response, message.qualifier(), message.dataFormatOrDefault()))
         .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.dataFormat()))
+        .map(response -> toResponse(response, message.qualifier(), message.dataFormatOrDefault()))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
   }
 
@@ -102,7 +102,9 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
                     .map(
                         response ->
                             toResponse(
-                                response, first.get().qualifier(), first.get().dataFormat())))
+                                response,
+                                first.get().qualifier(),
+                                first.get().dataFormatOrDefault())))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
   }
 
@@ -195,9 +197,8 @@ private ServiceMessage toResponse(Object response, String qualifier, String data
     final Object data;
 
     if (response instanceof ServiceMessage) {
-      ServiceMessage message = (ServiceMessage) response;
-      builder = ServiceMessage.from(message);
-      data = message.data();
+      builder = ServiceMessage.from((ServiceMessage) response);
+      data = ((ServiceMessage) response).data();
     } else {
       builder = ServiceMessage.builder();
       data = response;

services-bytebuf-codec/src/main/java/io/scalecube/services/transport/api/ServiceMessageCodec.java

@@ -31,7 +31,6 @@ public final class ServiceMessageCodec {
   private final HeadersCodec headersCodec;
   private final Map<String, DataCodec> dataCodecs;
 
-
   /** Message codec with default Headers/Data Codecs. */
   public ServiceMessageCodec() {
     this(null, null);
@@ -108,8 +107,7 @@ public <T> T encodeAndTransform(
         ReferenceCountUtil.safestRelease(headersBuffer);
         ReferenceCountUtil.safestRelease(dataBuffer); // release data buf as well
         LOGGER.error("Failed to encode headers on: {}, cause: {}", message, ex.toString());
-        throw new MessageCodecException(
-            "Failed to encode headers on message q=" + message.qualifier(), ex);
+        throw new MessageCodecException("Failed to encode headers message q=[" + message + "]", ex);
       }
     }