Fix keepalive from hanging the connection when input completes.

If input completes prior to close frame, emit error.

Author: mbonneau

src/MessageSubject.php

@@ -14,6 +14,7 @@
 use Rx\Disposable\CompositeDisposable;
 use Rx\DisposableInterface;
 use Rx\Exception\TimeoutException;
+use Rx\Notification\OnNextNotification;
 use Rx\Observable;
 use Rx\ObserverInterface;
 use Rx\Subject\Subject;
@@ -100,15 +101,21 @@ function (FrameInterface $frame) {
                         });
                 })
                 ->switch()
-                ->flatMapTo(Observable::never());
+                ->flatMapTo(Observable::never())
+                // This detects close or error notifications from the raw data input and stops the keepalive
+                ->takeUntil($this->rawDataIn
+                                ->materialize()
+                                ->filter(function($notification) { return ! $notification instanceof OnNextNotification; })
+                                ->take(1));
         }
 
         $this->rawDataDisp = $this->rawDataIn
             ->merge($keepAliveObs)
             ->subscribe(
                 [$messageBuffer, 'onData'],
                 function (\Throwable $e) { parent::onError($e); },
-                function () { parent::onCompleted(); }
+                // onCompleted needs to send an error. If a close frame comes in, this should be disposed already
+                function () { parent::onError(new WebsocketErrorException(Frame::CLOSE_ABNORMAL)); }
             );
 
         $this->subProtocol = $subProtocol;

test/MessageSubjectTest.php

@@ -183,4 +183,37 @@ public function testDisposeOnMessageSubjectClosesConnection()
             onCompleted(300)
         ], $dataOut->getMessages());
     }
-}
+
+    public function testMessageSubjectErrorsIfDataInStreamEndsClosesOrErrors() {
+        $dataIn = $this->createHotObservable([
+            onNext(201, (new Frame('', true, Frame::OP_TEXT))->getContents()),
+            onCompleted(205)
+        ]);
+
+        $dataOut = new MockObserver($this->scheduler);
+
+        $ms = new MessageSubject(
+            $dataIn,
+            $dataOut,
+            true,
+            false,
+            '',
+            new Request('GET', '/ws'),
+            new Response(),
+            300
+        );
+
+        $result = $this->scheduler->startWithDispose(function () use ($ms) {
+            return $ms;
+        }, 500);
+
+        $this->assertMessages([
+            onNext(201, ''),
+            onError(205, new WebsocketErrorException(Frame::CLOSE_ABNORMAL))
+        ], $result->getMessages());
+
+        $this->assertSubscriptions([
+            subscribe(0,205)
+        ], $dataIn->getSubscriptions());
+    }
+}