Merge pull request #267 from http4s/pr/fix-ws-connect-error

Fix WebSocket connect error handling

Author: armanbilge

dom/src/main/scala/org/http4s/dom/WebSocketClient.scala

@@ -39,6 +39,7 @@ import org.scalajs.dom.WebSocket
 import org.typelevel.ci._
 import scodec.bits.ByteVector
 
+import java.io.IOException
 import scala.scalajs.js
 import scala.scalajs.js.JSConverters._
 
@@ -50,7 +51,6 @@ object WebSocketClient {
         dispatcher <- Dispatcher.sequential[F]
         messages <- Queue.unbounded[F, Option[MessageEvent]].toResource
         semaphore <- Semaphore[F](1).toResource
-        error <- F.deferred[Throwable].toResource
         close <- F.deferred[CloseEvent].toResource
         ws <- Resource.makeCase {
           F.async_[WebSocket] { cb =>
@@ -67,15 +67,23 @@ object WebSocketClient {
             ws.binaryType = "arraybuffer" // the default is blob
 
             ws.onopen = { _ =>
-              ws.onerror = // replace the error handler
-                e => dispatcher.unsafeRunAndForget(error.complete(js.JavaScriptException(e)))
+              ws.onmessage = // setup message handler
+                e => dispatcher.unsafeRunAndForget(messages.offer(Some(e)))
+
+              ws.onclose = // replace the close handler
+                e => dispatcher.unsafeRunAndForget(messages.offer(None) *> close.complete(e))
+
+              // no explicit error handler. according to spec:
+              //   1. an error event is *always* followed by a close event and
+              //   2. an error event doesn't carry any useful information *by design*
+
               cb(Right(ws))
             }
 
-            ws.onerror = e => cb(Left(js.JavaScriptException(e)))
-            ws.onmessage = e => dispatcher.unsafeRunAndForget(messages.offer(Some(e)))
-            ws.onclose =
-              e => dispatcher.unsafeRunAndForget(messages.offer(None) *> close.complete(e))
+            // a close at this stage can only be an error
+            // following spec we cannot get any detail about the error
+            // https://websockets.spec.whatwg.org/#eventdef-websocket-error
+            ws.onclose = _ => cb(Left(new IOException("Connection failed")))
           }
         } {
           case (ws, exitCase) =>
@@ -117,33 +125,27 @@ object WebSocketClient {
         def closeFrame: DeferredSource[F, WSFrame.Close] =
           (close: DeferredSource[F, CloseEvent]).map(e => WSFrame.Close(e.code, e.reason))
 
-        def receive: F[Option[WSDataFrame]] = semaphore
-          .permit
-          .surround(OptionT(messages.take).map(decodeMessage).value)
-          .race(error.get.flatMap(F.raiseError[Option[WSDataFrame]]))
-          .map(_.merge)
+        def receive: F[Option[WSDataFrame]] =
+          semaphore.permit.surround(OptionT(messages.take).map(decodeMessage).value)
 
         override def receiveStream: Stream[F, WSDataFrame] =
-          Stream
-            .resource(semaphore.permit)
-            .flatMap(_ => Stream.fromQueueNoneTerminated(messages))
-            .map(decodeMessage)
-            .concurrently(Stream.exec(error.get.flatMap(F.raiseError)))
+          Stream.resource(semaphore.permit) >>
+            Stream.fromQueueNoneTerminated(messages).map(decodeMessage)
 
         private def decodeMessage(e: MessageEvent): WSDataFrame =
           e.data match {
             case s: String => WSFrame.Text(s)
             case b: js.typedarray.ArrayBuffer =>
               WSFrame.Binary(ByteVector.fromJSArrayBuffer(b))
             case _ => // this should never happen
-              throw new RuntimeException
+              throw new AssertionError
           }
 
         override def sendText(text: String): F[Unit] =
-          errorOr(F.delay(ws.send(text)))
+          F.delay(ws.send(text))
 
         override def sendBinary(bytes: ByteVector): F[Unit] =
-          errorOr(F.delay(ws.send(bytes.toJSArrayBuffer)))
+          F.delay(ws.send(bytes.toJSArrayBuffer))
 
         def send(wsf: WSDataFrame): F[Unit] =
           wsf match {
@@ -153,11 +155,6 @@ object WebSocketClient {
               F.raiseError(new IllegalArgumentException("DataFrames cannot be fragmented"))
           }
 
-        private def errorOr(fu: F[Unit]): F[Unit] = error.tryGet.flatMap {
-          case Some(error) => F.raiseError(error)
-          case None => fu
-        }
-
         def sendMany[G[_]: Foldable, A <: WSDataFrame](wsfs: G[A]): F[Unit] =
           wsfs.foldMapM(send(_))
 

testsBrowser/src/test/scala/org/http4s/dom/WebSocketSuite.scala

@@ -24,6 +24,8 @@ import org.http4s.client.websocket.WSRequest
 import org.http4s.dom.BuildInfo.fileServicePort
 import scodec.bits.ByteVector
 
+import java.io.IOException
+
 class WebSocketSuite extends CatsEffectSuite {
 
   test("send and receive frames") {
@@ -46,4 +48,12 @@ class WebSocketSuite extends CatsEffectSuite {
       )
   }
 
+  test("Raises IOException on connect error") {
+    WebSocketClient[IO]
+      .connectHighLevel(
+        WSRequest(Uri.fromString(s"ws://localhost:${fileServicePort}/not-ws").toOption.get))
+      .use_
+      .intercept[IOException]
+  }
+
 }