Materialize errors in event flows

Resolves:
#516

Author: joffrey-bion

src/commonMain/kotlin/org/hildan/chrome/devtools/protocol/ChromeDPConnection.kt

@@ -34,6 +34,7 @@ internal class ChromeDPConnection(
     private val frames = webSocket.incoming.receiveAsFlow()
         .filterIsInstance<Frame.Text>()
         .map { frame -> chromeDpJson.decodeFromString(InboundFrameSerializer, frame.readText()) }
+        .materializeErrors()
         .shareIn(
             scope = coroutineScope,
             started = SharingStarted.Eagerly,
@@ -47,6 +48,7 @@ internal class ChromeDPConnection(
      */
     suspend fun request(request: RequestFrame): ResponseFrame {
         val resultFrame = frames.onSubscription { sendOrFailUniformly(request) }
+            .dematerializeErrors()
             .filterIsInstance<ResultFrame>()
             .filter { it.matchesRequest(request) }
             .first() // a shared flow never completes, so this will never throw NoSuchElementException (but can hang forever)
@@ -76,7 +78,7 @@ internal class ChromeDPConnection(
     /**
      * A flow of incoming events.
      */
-    fun events() = frames.filterIsInstance<EventFrame>()
+    fun events() = frames.dematerializeErrors().filterIsInstance<EventFrame>()
 
     /**
      * Stops listening to incoming events and closes the underlying web socket connection.
@@ -87,6 +89,17 @@ internal class ChromeDPConnection(
     }
 }
 
+private fun Flow<InboundFrame>.materializeErrors(): Flow<InboundFrameOrError> =
+    catch<InboundFrameOrError> { emit(InboundFramesConnectionError(cause = it)) }
+
+private fun Flow<InboundFrameOrError>.dematerializeErrors(): Flow<InboundFrame> =
+    map {
+        when (it) {
+            is InboundFramesConnectionError -> throw it.cause
+            is InboundFrame -> it
+        }
+    }
+
 /**
  * An exception thrown when an error occurred during the processing of a request on Chrome side.
  */

src/commonMain/kotlin/org/hildan/chrome/devtools/protocol/ChromeDPFrames.kt

@@ -33,10 +33,20 @@ data class RequestFrame(
     val sessionId: String? = null,
 )
 
+/**
+ * General interface represent both real incoming .
+ */
+internal sealed interface InboundFrameOrError
+
+/**
+ * Represents errors in the incoming frames flow.
+ */
+internal class InboundFramesConnectionError(val cause: Throwable) : InboundFrameOrError
+
 /**
  * A generic inbound frame received from the server. It can represent responses to requests, or server-initiated events.
  */
-internal sealed class InboundFrame {
+internal sealed class InboundFrame : InboundFrameOrError {
     /** The session ID of the target concerned by this event. */
     abstract val sessionId: SessionID?
 }