sendingQueue

send

start

class PersistentWsClient(private val wsClient: WsClient, private val scope: CoroutineScope) {
    private var session: WsSession? = null
    private val _errors = Channel<Error>()
    val errors = _errors.consumeAsFlow()

    private val sendingQueue = Channel<Frame>(Channel.UNLIMITED)
    private val _incoming = BroadcastChannel<StreamUpdate>(CONFLATED) // TODO is conflated right?
    val incoming: Flow<StreamUpdate> = _incoming.asFlow()

    fun open(host: String, path: String, port: Int = 8080) {
        scope.launch {
            when (val open = wsClient.open(host, path, port)) {
                is Result.Success -> {
                    session = open.success
                    start()
                }
                is Result.Error -> {
                    this@PersistentWsClient.w("error opening connection: ${open.error.message}", open.error)
                    _errors.offer(Error.OpeningConnection(open.error))

                    // TODO back-off and try again
                }
            }
        }
    }

    fun send(payload: Frame) {
        // ignoring result bc Channel.UNLIMITED always returns true to `offer`
        sendingQueue.offer(payload)
        this.e("offerring $payload")
    }

    private fun start() {
        _incoming.offer(StreamUpdate.ConnectionChange.Connected)
        scope.launch {
            session?.let { session ->
                this.e("starting persistent client")
                launch {
                    for (frame in sendingQueue) {
                        this@PersistentWsClient.e("sending queue for $frame")
                        when (val send = session.send(frame)) {
                            UnitResult.Success -> {
                                this@PersistentWsClient.d("send frame")
                            }
                            is UnitResult.Error -> {
                                this@PersistentWsClient.e("error sending message: ${send.error.message}")
                                _errors.offer(Error.SendingFrame(frame))

                                // TODO check if disconnected!
                            }
                        }
                    }
                }

                launch {
                    session.incoming
                        .onEach { _incoming.offer(StreamUpdate.FrameWrapper(it)) }
                        .collect()
                }
            } ?: e("start called when session id is null!")
        }
    }

    suspend fun close() {
        scope.cancel()
        session?.close()
        session = null
    }

    sealed class Error {
        class OpeningConnection(val e: Exception): Error()
        class SendingFrame(val frame: Frame): Error()
    }

    sealed class StreamUpdate {
        class FrameWrapper(frame: Frame): StreamUpdate()

        sealed class ConnectionChange: StreamUpdate() {
            object ConnectionLost : ConnectionChange()
            object Connected : ConnectionChange()
        }
    }
}

offering ...
starting persistent client