Setup payload: distinguish missing and empty metadata

Signed-off-by: Maksym Ostroverkhov <[email protected]>

Author: mostroverkhov

rsocket-core/src/main/java/io/rsocket/RSocketFactory.java

@@ -355,8 +355,7 @@ public Mono<RSocket> start() {
                           resumeToken,
                           metadataMimeType,
                           dataMimeType,
-                          setupPayload.sliceMetadata(),
-                          setupPayload.sliceData());
+                          setupPayload);
 
                   RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
 

rsocket-core/src/main/java/io/rsocket/frame/SetupFrameFlyweight.java

@@ -4,6 +4,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
+import io.rsocket.Payload;
 import java.nio.charset.StandardCharsets;
 
 public class SetupFrameFlyweight {
@@ -27,13 +28,12 @@ public class SetupFrameFlyweight {
 
   public static ByteBuf encode(
       final ByteBufAllocator allocator,
-      boolean lease,
+      final boolean lease,
       final int keepaliveInterval,
       final int maxLifetime,
       final String metadataMimeType,
       final String dataMimeType,
-      final ByteBuf metadata,
-      final ByteBuf data) {
+      final Payload setupPayload) {
     return encode(
         allocator,
         lease,
@@ -42,20 +42,21 @@ public static ByteBuf encode(
         Unpooled.EMPTY_BUFFER,
         metadataMimeType,
         dataMimeType,
-        metadata,
-        data);
+        setupPayload);
   }
 
   public static ByteBuf encode(
       final ByteBufAllocator allocator,
-      boolean lease,
+      final boolean lease,
       final int keepaliveInterval,
       final int maxLifetime,
       final ByteBuf resumeToken,
       final String metadataMimeType,
       final String dataMimeType,
-      final ByteBuf metadata,
-      final ByteBuf data) {
+      final Payload setupPayload) {
+
+    ByteBuf metadata = setupPayload.hasMetadata() ? setupPayload.sliceMetadata() : null;
+    ByteBuf data = setupPayload.sliceData();
 
     int flags = 0;
 

rsocket-core/src/test/java/io/rsocket/ConnectionSetupPayloadTest.java

@@ -0,0 +1,88 @@
+package io.rsocket;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.rsocket.frame.SetupFrameFlyweight;
+import io.rsocket.util.DefaultPayload;
+import org.junit.jupiter.api.Test;
+
+class ConnectionSetupPayloadTest {
+  private static final int KEEP_ALIVE_INTERVAL = 5;
+  private static final int KEEP_ALIVE_MAX_LIFETIME = 500;
+  private static final String METADATA_TYPE = "metadata_type";
+  private static final String DATA_TYPE = "data_type";
+
+  @Test
+  void testSetupPayloadWithDataMetadata() {
+    ByteBuf data = Unpooled.wrappedBuffer(new byte[] {5, 4, 3});
+    ByteBuf metadata = Unpooled.wrappedBuffer(new byte[] {2, 1, 0});
+    Payload payload = DefaultPayload.create(data, metadata);
+    boolean leaseEnabled = true;
+
+    ByteBuf frame = encodeSetupFrame(leaseEnabled, payload);
+    ConnectionSetupPayload setupPayload = ConnectionSetupPayload.create(frame);
+
+    assertTrue(setupPayload.willClientHonorLease());
+    assertEquals(KEEP_ALIVE_INTERVAL, setupPayload.keepAliveInterval());
+    assertEquals(KEEP_ALIVE_MAX_LIFETIME, setupPayload.keepAliveMaxLifetime());
+    assertEquals(METADATA_TYPE, SetupFrameFlyweight.metadataMimeType(frame));
+    assertEquals(DATA_TYPE, SetupFrameFlyweight.dataMimeType(frame));
+    assertTrue(setupPayload.hasMetadata());
+    assertNotNull(setupPayload.metadata());
+    assertEquals(payload.metadata(), setupPayload.metadata());
+    assertEquals(payload.data(), setupPayload.data());
+    frame.release();
+  }
+
+  @Test
+  void testSetupPayloadWithNoMetadata() {
+    ByteBuf data = Unpooled.wrappedBuffer(new byte[] {5, 4, 3});
+    ByteBuf metadata = null;
+    Payload payload = DefaultPayload.create(data, metadata);
+    boolean leaseEnabled = false;
+
+    ByteBuf frame = encodeSetupFrame(leaseEnabled, payload);
+    ConnectionSetupPayload setupPayload = ConnectionSetupPayload.create(frame);
+
+    assertFalse(setupPayload.willClientHonorLease());
+    assertFalse(setupPayload.hasMetadata());
+    assertNotNull(setupPayload.metadata());
+    assertEquals(0, setupPayload.metadata().readableBytes());
+    assertEquals(payload.data(), setupPayload.data());
+    frame.release();
+  }
+
+  @Test
+  void testSetupPayloadWithEmptyMetadata() {
+    ByteBuf data = Unpooled.wrappedBuffer(new byte[] {5, 4, 3});
+    ByteBuf metadata = Unpooled.EMPTY_BUFFER;
+    Payload payload = DefaultPayload.create(data, metadata);
+    boolean leaseEnabled = false;
+
+    ByteBuf frame = encodeSetupFrame(leaseEnabled, payload);
+    ConnectionSetupPayload setupPayload = ConnectionSetupPayload.create(frame);
+
+    assertFalse(setupPayload.willClientHonorLease());
+    assertTrue(setupPayload.hasMetadata());
+    assertNotNull(setupPayload.metadata());
+    assertEquals(0, setupPayload.metadata().readableBytes());
+    assertEquals(payload.data(), setupPayload.data());
+    frame.release();
+  }
+
+  private static ByteBuf encodeSetupFrame(boolean leaseEnabled, Payload setupPayload) {
+    return SetupFrameFlyweight.encode(
+        ByteBufAllocator.DEFAULT,
+        leaseEnabled,
+        KEEP_ALIVE_INTERVAL,
+        KEEP_ALIVE_MAX_LIFETIME,
+        null,
+        METADATA_TYPE,
+        DATA_TYPE,
+        setupPayload);
+  }
+}

rsocket-core/src/test/java/io/rsocket/RSocketLeaseTest.java

@@ -124,8 +124,7 @@ public void serverRSocketFactoryRejectsUnsupportedLease() {
             30_000,
             "application/octet-stream",
             "application/octet-stream",
-            payload.sliceMetadata(),
-            payload.sliceData());
+            payload);
 
     TestServerTransport transport = new TestServerTransport();
     Closeable server =

rsocket-core/src/test/java/io/rsocket/SetupRejectionTest.java

@@ -149,14 +149,7 @@ public void connect() {
       Payload payload = DefaultPayload.create(DefaultPayload.EMPTY_BUFFER);
       ByteBuf setup =
           SetupFrameFlyweight.encode(
-              ByteBufAllocator.DEFAULT,
-              false,
-              0,
-              42,
-              "mdMime",
-              "dMime",
-              payload.sliceMetadata(),
-              payload.sliceData());
+              ByteBufAllocator.DEFAULT, false, 0, 42, "mdMime", "dMime", payload);
 
       conn.addToReceivedBuffer(setup);
     }

rsocket-core/src/test/java/io/rsocket/frame/SetupFrameFlyweightTest.java

@@ -5,6 +5,8 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
+import io.rsocket.Payload;
+import io.rsocket.util.DefaultPayload;
 import java.util.Arrays;
 import org.junit.jupiter.api.Test;
 
@@ -13,9 +15,10 @@ class SetupFrameFlyweightTest {
   void testEncodingNoResume() {
     ByteBuf metadata = Unpooled.wrappedBuffer(new byte[] {1, 2, 3, 4});
     ByteBuf data = Unpooled.wrappedBuffer(new byte[] {5, 4, 3});
+    Payload payload = DefaultPayload.create(data, metadata);
     ByteBuf frame =
         SetupFrameFlyweight.encode(
-            ByteBufAllocator.DEFAULT, false, 5, 500, "metadata_type", "data_type", metadata, data);
+            ByteBufAllocator.DEFAULT, false, 5, 500, "metadata_type", "data_type", payload);
 
     assertEquals(FrameType.SETUP, FrameHeaderFlyweight.frameType(frame));
     assertFalse(SetupFrameFlyweight.resumeEnabled(frame));
@@ -34,18 +37,11 @@ void testEncodingResume() {
     Arrays.fill(tokenBytes, (byte) 1);
     ByteBuf metadata = Unpooled.wrappedBuffer(new byte[] {1, 2, 3, 4});
     ByteBuf data = Unpooled.wrappedBuffer(new byte[] {5, 4, 3});
+    Payload payload = DefaultPayload.create(data, metadata);
     ByteBuf token = Unpooled.wrappedBuffer(tokenBytes);
     ByteBuf frame =
         SetupFrameFlyweight.encode(
-            ByteBufAllocator.DEFAULT,
-            true,
-            5,
-            500,
-            token,
-            "metadata_type",
-            "data_type",
-            metadata,
-            data);
+            ByteBufAllocator.DEFAULT, true, 5, 500, token, "metadata_type", "data_type", payload);
 
     assertEquals(FrameType.SETUP, FrameHeaderFlyweight.frameType(frame));
     assertTrue(SetupFrameFlyweight.honorLease(frame));

rsocket-core/src/test/java/io/rsocket/internal/ClientServerInputMultiplexerTest.java

@@ -24,6 +24,7 @@
 import io.rsocket.frame.*;
 import io.rsocket.plugins.PluginRegistry;
 import io.rsocket.test.util.TestDuplexConnection;
+import io.rsocket.util.DefaultPayload;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Before;
 import org.junit.Test;
@@ -199,8 +200,7 @@ private ByteBuf setupFrame() {
         42,
         "application/octet-stream",
         "application/octet-stream",
-        Unpooled.EMPTY_BUFFER,
-        Unpooled.EMPTY_BUFFER);
+        DefaultPayload.create(Unpooled.EMPTY_BUFFER, Unpooled.EMPTY_BUFFER));
   }
 
   private ByteBuf leaseFrame() {

rsocket-test/src/main/java/io/rsocket/test/TestFrames.java

@@ -22,6 +22,7 @@
 import io.rsocket.Payload;
 import io.rsocket.frame.*;
 import io.rsocket.util.DefaultPayload;
+import io.rsocket.util.EmptyPayload;
 
 /** Test instances of all frame types. */
 public final class TestFrames {
@@ -104,7 +105,6 @@ public static ByteBuf createTestSetupFrame() {
         Unpooled.EMPTY_BUFFER,
         "metadataType",
         "dataType",
-        null,
-        Unpooled.EMPTY_BUFFER);
+        EmptyPayload.INSTANCE);
   }
 }