Merge branch '1.0.x'

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

rsocket-core/src/main/java/io/rsocket/internal/ClientServerInputMultiplexer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2018 the original author or authors.
+ * Copyright 2015-2020 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -55,6 +55,8 @@ public class ClientServerInputMultiplexer implements Closeable {
   private final DuplexConnection source;
   private final DuplexConnection clientServerConnection;
 
+  private boolean setupReceived;
+
   public ClientServerInputMultiplexer(DuplexConnection source) {
     this(source, emptyInterceptorRegistry, false);
   }
@@ -87,6 +89,7 @@ public ClientServerInputMultiplexer(
                   case RESUME:
                   case RESUME_OK:
                     type = Type.SETUP;
+                    setupReceived = true;
                     break;
                   case LEASE:
                   case KEEPALIVE:
@@ -101,6 +104,11 @@ public ClientServerInputMultiplexer(
               } else {
                 type = Type.CLIENT;
               }
+              if (!isClient && type != Type.SETUP && !setupReceived) {
+                frame.release();
+                throw new IllegalStateException(
+                    "SETUP or LEASE frame must be received before any others.");
+              }
               return type;
             })
         .subscribe(
@@ -119,7 +127,11 @@ public ClientServerInputMultiplexer(
                   break;
               }
             },
-            t -> {});
+            ex -> {
+              setup.onError(ex);
+              server.onError(ex);
+              client.onError(ex);
+            });
   }
 
   public DuplexConnection asClientServerConnection() {

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

@@ -1,9 +1,19 @@
 package io.rsocket.core;
 
 import static io.rsocket.frame.FrameLengthCodec.FRAME_LENGTH_MASK;
+import static org.assertj.core.api.Assertions.assertThat;
 
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.rsocket.RSocket;
+import io.rsocket.frame.RequestResponseFrameCodec;
+import io.rsocket.test.util.TestDuplexConnection;
 import io.rsocket.test.util.TestServerTransport;
+import java.time.Duration;
+import java.util.Random;
 import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
 import reactor.test.StepVerifier;
 
 public class RSocketServerTest {
@@ -42,4 +52,34 @@ public void ensuresMaxFrameLengthCanNotBeGreaterThenMaxPossibleFrameLength() {
                 + FRAME_LENGTH_MASK)
         .verify();
   }
+
+  @Test
+  public void unexpectedFramesBeforeSetup() {
+    MonoProcessor<Void> connectedMono = MonoProcessor.create();
+
+    TestServerTransport transport = new TestServerTransport();
+    RSocketServer.create()
+        .acceptor(
+            (setup, sendingSocket) -> {
+              connectedMono.onComplete();
+              return Mono.just(new RSocket() {});
+            })
+        .bind(transport)
+        .block();
+
+    byte[] bytes = new byte[16_000_000];
+    new Random().nextBytes(bytes);
+
+    TestDuplexConnection connection = transport.connect();
+    connection.addToReceivedBuffer(
+        RequestResponseFrameCodec.encode(
+            ByteBufAllocator.DEFAULT,
+            1,
+            false,
+            Unpooled.EMPTY_BUFFER,
+            ByteBufAllocator.DEFAULT.buffer(bytes.length).writeBytes(bytes)));
+
+    StepVerifier.create(connection.onClose()).expectComplete().verify(Duration.ofSeconds(30));
+    assertThat(connectedMono.isTerminated()).as("Connection should not succeed").isFalse();
+  }
 }

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

@@ -16,17 +16,25 @@
 
 package io.rsocket.internal;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.rsocket.buffer.LeaksTrackingByteBufAllocator;
-import io.rsocket.frame.*;
+import io.rsocket.frame.ErrorFrameCodec;
+import io.rsocket.frame.KeepAliveFrameCodec;
+import io.rsocket.frame.LeaseFrameCodec;
+import io.rsocket.frame.MetadataPushFrameCodec;
+import io.rsocket.frame.ResumeFrameCodec;
+import io.rsocket.frame.ResumeOkFrameCodec;
+import io.rsocket.frame.SetupFrameCodec;
 import io.rsocket.plugins.InitializingInterceptorRegistry;
 import io.rsocket.test.util.TestDuplexConnection;
 import io.rsocket.util.DefaultPayload;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -68,44 +76,44 @@ public void clientSplits() {
         .doOnNext(f -> setupFrames.incrementAndGet())
         .subscribe();
 
+    source.addToReceivedBuffer(setupFrame().retain());
+    assertEquals(0, clientFrames.get());
+    assertEquals(0, serverFrames.get());
+    assertEquals(1, setupFrames.get());
+
     source.addToReceivedBuffer(errorFrame(1).retain());
     assertEquals(1, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(1).retain());
     assertEquals(2, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(leaseFrame().retain());
     assertEquals(3, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(keepAliveFrame().retain());
     assertEquals(4, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(2).retain());
     assertEquals(4, clientFrames.get());
     assertEquals(1, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(0).retain());
     assertEquals(5, clientFrames.get());
     assertEquals(1, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(metadataPushFrame().retain());
     assertEquals(5, clientFrames.get());
     assertEquals(2, serverFrames.get());
-    assertEquals(0, setupFrames.get());
-
-    source.addToReceivedBuffer(setupFrame().retain());
-    assertEquals(5, clientFrames.get());
-    assertEquals(2, serverFrames.get());
     assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(resumeFrame().retain());
@@ -141,44 +149,44 @@ public void serverSplits() {
         .doOnNext(f -> setupFrames.incrementAndGet())
         .subscribe();
 
+    source.addToReceivedBuffer(setupFrame().retain());
+    assertEquals(0, clientFrames.get());
+    assertEquals(0, serverFrames.get());
+    assertEquals(1, setupFrames.get());
+
     source.addToReceivedBuffer(errorFrame(1).retain());
     assertEquals(1, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(1).retain());
     assertEquals(2, clientFrames.get());
     assertEquals(0, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(leaseFrame().retain());
     assertEquals(2, clientFrames.get());
     assertEquals(1, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(keepAliveFrame().retain());
     assertEquals(2, clientFrames.get());
     assertEquals(2, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(2).retain());
     assertEquals(2, clientFrames.get());
     assertEquals(3, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(errorFrame(0).retain());
     assertEquals(2, clientFrames.get());
     assertEquals(4, serverFrames.get());
-    assertEquals(0, setupFrames.get());
+    assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(metadataPushFrame().retain());
     assertEquals(3, clientFrames.get());
     assertEquals(4, serverFrames.get());
-    assertEquals(0, setupFrames.get());
-
-    source.addToReceivedBuffer(setupFrame().retain());
-    assertEquals(3, clientFrames.get());
-    assertEquals(4, serverFrames.get());
     assertEquals(1, setupFrames.get());
 
     source.addToReceivedBuffer(resumeFrame().retain());
@@ -192,6 +200,43 @@ public void serverSplits() {
     assertEquals(3, setupFrames.get());
   }
 
+  @Test
+  public void unexpectedFramesBeforeSetupFrame() {
+    AtomicInteger clientFrames = new AtomicInteger();
+    AtomicInteger serverFrames = new AtomicInteger();
+    AtomicInteger setupFrames = new AtomicInteger();
+
+    AtomicReference<Throwable> clientError = new AtomicReference<>();
+    AtomicReference<Throwable> serverError = new AtomicReference<>();
+    AtomicReference<Throwable> setupError = new AtomicReference<>();
+
+    serverMultiplexer
+        .asClientConnection()
+        .receive()
+        .subscribe(bb -> clientFrames.incrementAndGet(), clientError::set);
+    serverMultiplexer
+        .asServerConnection()
+        .receive()
+        .subscribe(bb -> serverFrames.incrementAndGet(), serverError::set);
+    serverMultiplexer
+        .asSetupConnection()
+        .receive()
+        .subscribe(bb -> setupFrames.incrementAndGet(), setupError::set);
+
+    source.addToReceivedBuffer(keepAliveFrame().retain());
+
+    assertThat(clientError.get().getMessage())
+        .isEqualTo("SETUP or LEASE frame must be received before any others.");
+    assertThat(serverError.get().getMessage())
+        .isEqualTo("SETUP or LEASE frame must be received before any others.");
+    assertThat(setupError.get().getMessage())
+        .isEqualTo("SETUP or LEASE frame must be received before any others.");
+
+    assertEquals(0, clientFrames.get());
+    assertEquals(0, serverFrames.get());
+    assertEquals(0, setupFrames.get());
+  }
+
   private ByteBuf resumeFrame() {
     return ResumeFrameCodec.encode(allocator, Unpooled.EMPTY_BUFFER, 0, 0);
   }