Removed loggin from bidirection invoker

Author: artem-v

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

@@ -156,39 +156,12 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
         .switchOnFirst(
             (first, messages) -> {
               final var message = first.get();
-              final var request = toRequest(message);
               final var qualifier = message.qualifier();
               final var dataFormat = message.dataFormat();
 
               return messages
                   .map(this::toRequest)
                   .transform(this::invokeRequest)
-                  .doOnSubscribe(
-                      s -> {
-                        if (logger != null && logger.isDebugEnabled()) {
-                          logger.debug(
-                              "[{}][subscribe] request: {}",
-                              message.qualifier(),
-                              toString(request));
-                        }
-                      })
-                  .doOnComplete(
-                      () -> {
-                        if (logger != null && logger.isDebugEnabled()) {
-                          logger.debug(
-                              "[{}][complete] request: {}", message.qualifier(), toString(request));
-                        }
-                      })
-                  .doOnError(
-                      ex -> {
-                        if (logger != null) {
-                          logger.error(
-                              "[{}][error] request: {}",
-                              message.qualifier(),
-                              toString(request),
-                              ex);
-                        }
-                      })
                   .map(response -> toResponse(response, qualifier, dataFormat))
                   .onErrorResume(ex -> Flux.just(errorMapper.toMessage(qualifier, ex)))
                   .subscribeOn(methodInfo.scheduler());

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/ServiceMessageCodec.java

@@ -1,5 +1,6 @@
 package io.scalecube.services.transport.rsocket;
 
+import static io.scalecube.services.transport.rsocket.ReferenceCountUtil.safestRelease;
 import static java.util.function.Function.identity;
 import static java.util.stream.Collectors.collectingAndThen;
 import static java.util.stream.Collectors.toMap;
@@ -102,7 +103,7 @@ public <T> T encodeAndTransform(
         DataCodec dataCodec = getDataCodec(message.dataFormatOrDefault());
         dataCodec.encode(new ByteBufOutputStream(dataBuffer), message.data());
       } catch (Throwable ex) {
-        ReferenceCountUtil.safestRelease(dataBuffer);
+        safestRelease(dataBuffer);
         LOGGER.error(
             "Failed to encode service message data on: {}, cause: {}", message, ex.toString());
         throw new MessageCodecException("Failed to encode service message data", ex);
@@ -114,8 +115,8 @@ public <T> T encodeAndTransform(
       try {
         headersCodec.encode(new ByteBufOutputStream(headersBuffer), message.headers());
       } catch (Throwable ex) {
-        ReferenceCountUtil.safestRelease(headersBuffer);
-        ReferenceCountUtil.safestRelease(dataBuffer); // release data buf as well
+        safestRelease(headersBuffer);
+        safestRelease(dataBuffer); // release data buf as well
         LOGGER.error(
             "Failed to encode service message headers on: {}, cause: {}", message, ex.toString());
         throw new MessageCodecException("Failed to encode service message headers", ex);
@@ -144,7 +145,7 @@ public ServiceMessage decode(ByteBuf dataBuffer, ByteBuf headersBuffer)
       try (ByteBufInputStream stream = new ByteBufInputStream(headersBuffer, true)) {
         builder.headers(headersCodec.decode(stream));
       } catch (Throwable ex) {
-        ReferenceCountUtil.safestRelease(dataBuffer); // release data buf as well
+        safestRelease(dataBuffer); // release data buf as well
         throw new MessageCodecException("Failed to decode service message headers", ex);
       }
     }
@@ -163,12 +164,15 @@ public ServiceMessage decode(ByteBuf dataBuffer, ByteBuf headersBuffer)
    */
   public static ServiceMessage decodeData(ServiceMessage message, Type dataType)
       throws MessageCodecException {
-    if (dataType == null || !message.hasData(ByteBuf.class)) {
+    if (dataType == null
+        || dataType == ByteBuf.class
+        || message.data() == null
+        || !(message.data() instanceof ByteBuf)) {
       return message;
     }
 
     final ByteBuf dataBuffer = message.data();
-    if (dataBuffer.readableBytes() == 0 || dataType == ByteBuf.class) {
+    if (dataBuffer.readableBytes() == 0) {
       return message;
     }
     if (dataType == byte[].class) {
@@ -177,17 +181,14 @@ public static ServiceMessage decodeData(ServiceMessage message, Type dataType)
       return ServiceMessage.from(message).data(bytes).build();
     }
 
-    Object data;
-    Type targetType = message.isError() ? ErrorData.class : dataType;
-
     try (ByteBufInputStream inputStream = new ByteBufInputStream(dataBuffer, true)) {
-      DataCodec dataCodec = DataCodec.getInstance(message.dataFormatOrDefault());
-      data = dataCodec.decode(inputStream, targetType);
+      final var targetType = message.isError() ? ErrorData.class : dataType;
+      final var dataCodec = DataCodec.getInstance(message.dataFormatOrDefault());
+      final var decodedData = dataCodec.decode(inputStream, targetType);
+      return ServiceMessage.from(message).data(decodedData).build();
     } catch (Throwable ex) {
       throw new MessageCodecException("Failed to decode service message data", ex);
     }
-
-    return ServiceMessage.from(message).data(data).build();
   }
 
   private DataCodec getDataCodec(String contentType) {