Make Rfc6455FrameCompiler::assertState() pass with interleaved contro…

…l frames
amphp · Oct 28, 2024 · b352336 · b352336
1 parent 1dc9d2c
commit b352336
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 4 deletions.
diff --git a/src/Parser/Rfc6455FrameCompiler.php b/src/Parser/Rfc6455FrameCompiler.php
@@ -29,10 +29,13 @@ public function compileFrame(WebsocketFrameType $frameType, string $data, bool $
             $this->compressPayload = !$isFinal || \strlen($data) > $this->compressionContext->getCompressionThreshold();
         }
 
-        $this->currentFrameType = match ($frameType) {
-            WebsocketFrameType::Text, WebsocketFrameType::Binary => $frameType,
-            default => $this->currentFrameType,
+        $isDataFrame = match ($frameType) {
+            WebsocketFrameType::Text, WebsocketFrameType::Binary => true,
+            default => false,
         };
+        if ($isDataFrame) {
+            $this->currentFrameType = $frameType;
+        }
 
         $rsv = 0;
 
@@ -51,7 +54,7 @@ public function compileFrame(WebsocketFrameType $frameType, string $data, bool $
             $isFinal = true; // Reset state in finally.
             throw $exception;
         } finally {
-            if ($isFinal) {
+            if (($isDataFrame || $frameType === WebsocketFrameType::Continuation) && $isFinal) {
                 $this->currentFrameType = null;
                 $this->compressPayload = false;
             }

diff --git a/test/WebsocketClientTest.php b/test/WebsocketClientTest.php
@@ -266,6 +266,45 @@ public function testStreamMultipleChunks(): void
         $client->streamText($stream);
     }
 
+    public function testStreamWithInterleavedControlFrames(): void
+    {
+        $packets = \array_map(fn(string $packet) => [$packet], [
+            compile(WebsocketFrameType::Text, false, false, 'chunk1'),
+            compile(WebsocketFrameType::Continuation, false, false, 'chunk2'),
+            compile(WebsocketFrameType::Ping, false, true, "1"),
+            compile(WebsocketFrameType::Continuation, false, false, 'chunk3'),
+            compile(WebsocketFrameType::Ping, false, true, "2"),
+            compile(WebsocketFrameType::Continuation, false, true, 'chunk4'),
+        ]);
+
+        $socket = $this->createSocket();
+        $future = new DeferredFuture();
+        $socket->expects($this->any())->method("read")->willReturnCallback(function() use ($future) {
+            $future->getFuture()->await();
+        });
+        $socket->expects($this->atLeastOnce())
+            ->method('write')
+            ->withConsecutive(...$packets);
+
+        $client = $this->createClient($socket, frameSplitThreshold: 6);
+
+        $stream = new ReadableIterableStream((function () use ($client) {
+            yield 'chunk1';
+            yield 'chunk2';
+            yield '';
+            $client->ping();
+            yield 'chunk3';
+            yield '';
+            $client->ping();
+            yield 'chunk4';
+        })());
+
+        $client->streamText($stream);
+
+        $future->complete();
+        $client->close();
+    }
+
     public function testSendWithFailedSocket(): void
     {
         $socket = $this->createSocket();