websocket

A WebSocket connection wrapper that exposes events as RxJS observables, allowing for idiomatic management of connection lifecycle.

Usage (server)

import { makeServer, Connection } from '@service-broker/websocket'

makeServer({ port: 8080 }).pipe(
  rxjs.exhaustMap(server =>
    rxjs.merge(
      server.connection$.pipe(
        rxjs.mergeMap(handleConnection)
      ),
      server.error$.pipe(
        rxjs.tap(event => console.error(event.error))
      )
    ).pipe(
      rxjs.finalize(() => server.close())
    )
  ),
  rxjs.takeUntil(shutdown$)
).subscribe()

function handleConnection(con: Connection) {
  return rxjs.merge(
    con.message$.pipe(
      rxjs.tap(event => handleMessage(event.data))
    ),
    con.error$.pipe(
      rxjs.tap(event => console.error(event.error))
    ),
    con.keepAlive(interval, pingTimeout).pipe(
      rxjs.catchError(err => {
        console.error('Ping timed out')
        con.terminate()
        return rxjs.EMPTY
      })
    )
  ).pipe(
    rxjs.takeUntil(con.close$),
    rxjs.finalize(() => con.close())
  )
}

Usage (client)

import { connect } from '@service-broker/websocket'

connect(wsUrl).pipe(
  rxjs.retry(),
  rxjs.exhaustMap(handleConnection),
  rxjs.repeat(),
  rxjs.takeUntil(shutdown$)
).subscribe()

Lifecycle contract

This package wraps WebSocket setup and raw events as observables, but it does not prescribe a full connection lifecycle policy. Callers are expected to compose retry, repeat, shutdown, keepalive, error handling, and connection cleanup in the stream that consumes the connection.

makeServer() and connect() own the resource acquisition phase. If a subscription is cancelled before the server is listening or before the client connection opens, the pending resource is closed or aborted.

After a Connection is emitted, the caller owns the connection lifecycle. message$ and error$ are live event streams and do not buffer events for late subscribers.

close$ is the terminal lifecycle signal. It resolves once and remains observable for late subscribers, so cleanup paths can safely use it as a lifecycle fence without racing the close event.

Caller-provided abort signals are honored during connect(). An already-aborted signal aborts the connection attempt before the WebSocket is constructed, and a later abort is forwarded until the WebSocket reaches close.