Merge branch 'master' into improve-requset-context

Author: artem-v

services-gateway/src/test/java/io/scalecube/services/gateway/files/FileUploadTest.java

@@ -86,7 +86,7 @@ static void afterAll() {
   }
 
   @ParameterizedTest(name = "Upload file of size {0} bytes")
-  @ValueSource(longs = {64, 512, 1024, 1024 * 1024, 10 * 1024 * 1024})
+  @ValueSource(longs = {0, 64, 512, 1024, 1024 * 1024, 10 * 1024 * 1024})
   public void uploadSuccessfully(long fileSize) throws Exception {
     final var client = new OkHttpClient();
     final var file = generateFile(Files.createTempFile("export_report_", null), fileSize);

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

@@ -138,16 +138,11 @@ public ServiceMessage decode(ByteBuf dataBuffer, ByteBuf headersBuffer)
       throws MessageCodecException {
     ServiceMessage.Builder builder = ServiceMessage.builder();
 
-    if (dataBuffer.isReadable()) {
-      builder.data(dataBuffer);
-    }
-    if (headersBuffer.isReadable()) {
-      try (ByteBufInputStream stream = new ByteBufInputStream(headersBuffer, true)) {
-        builder.headers(headersCodec.decode(stream));
-      } catch (Throwable ex) {
-        safestRelease(dataBuffer); // release data buf as well
-        throw new MessageCodecException("Failed to decode service message headers", ex);
-      }
+    try (ByteBufInputStream stream = new ByteBufInputStream(headersBuffer, true)) {
+      builder.headers(headersCodec.decode(stream)).data(dataBuffer);
+    } catch (Throwable ex) {
+      safestRelease(dataBuffer); // release data buf as well
+      throw new MessageCodecException("Failed to decode service message headers", ex);
     }
 
     return builder.build();
@@ -173,15 +168,20 @@ public static ServiceMessage decodeData(
     }
 
     final ByteBuf dataBuffer = message.data();
-    if (dataBuffer.readableBytes() == 0) {
-      return message;
-    }
+
     if (dataType == byte[].class) {
       final var bytes = new byte[dataBuffer.readableBytes()];
       dataBuffer.getBytes(dataBuffer.readerIndex(), bytes);
+      if (!copyOnDecode) {
+        safestRelease(dataBuffer);
+      }
       return ServiceMessage.from(message).data(bytes).build();
     }
 
+    if (dataBuffer.readableBytes() == 0) {
+      return ServiceMessage.from(message).data(null).build();
+    }
+
     try (ByteBufInputStream inputStream =
         copyOnDecode
             ? new ByteBufInputStream(Unpooled.copiedBuffer(dataBuffer))