npm package discovery and stats viewer.

Discover Tips

  • General search

    [free text search, go nuts!]

  • Package details

    pkg:[package-name]

  • User packages

    @[username]

Sponsor

Optimize Toolset

I’ve always been into building performant and accessible sites, but lately I’ve been taking it extremely seriously. So much so that I’ve been building a tool to help me optimize and monitor the sites that I build to make sure that I’m making an attempt to offer the best experience to those who visit them. If you’re into performant, accessible and SEO friendly sites, you might like it too! You can check it out at Optimize Toolset.

About

Hi, 👋, I’m Ryan Hefner  and I built this site for me, and you! The goal of this site was to provide an easy way for me to check the stats on my npm packages, both for prioritizing issues and updates, and to give me a little kick in the pants to keep up on stuff.

As I was building it, I realized that I was actually using the tool to build the tool, and figured I might as well put this out there and hopefully others will find it to be a fast and useful way to search and browse npm packages as I have.

If you’re interested in other things I’m working on, follow me on Twitter or check out the open source projects I’ve been publishing on GitHub.

I am also working on a Twitter bot for this site to tweet the most popular, newest, random packages from npm. Please follow that account now and it will start sending out packages soon–ish.

Open Software & Tools

This site wouldn’t be possible without the immense generosity and tireless efforts from the people who make contributions to the world and share their work via open source initiatives. Thank you 🙏

© 2024 – Pkg Stats / Ryan Hefner

@fusionx7/rxjs-websockets

v8.0.2

Published

rxjs 6 websockets library

Downloads

11

Readme

rxjs-websockets

build status Known Vulnerabilities Renovate

An rxjs websocket library with a simple and flexible implementation. Supports the browser and node.js.

Comparisons to other rxjs websocket libraries:

  • observable-socket
    • observable-socket provides an input subject for the user, rxjs-websockets allows the user to supply the input stream as a parameter to allow the user to select an observable with semantics appropriate for their own use case (queueing-subject can be used to achieve the same semantics as observable-socket).
    • With observable-socket the WebSocket object must be used and managed by the user, rxjs-websocket manages the WebSocket(s) for the user lazily according to subscriptions to the messages observable.
    • With observable-socket the WebSocket object must be observed using plain old events to detect the connection status, rxjs-websockets presents the connection status through observables.
  • rxjs built-in websocket subject
    • Implemented as a Subject so lacks the flexibility that rxjs-websockets and observable-socket provide.
    • Does not provide any ability to monitor the web socket connection state.

Installation

npm install -S rxjs-websockets
# or
yarn add rxjs-websockets

Changelog

Changelog here

Simple usage

import { QueueingSubject } from 'queueing-subject'
import { Subscription } from 'rxjs'
import { share, switchMap } from 'rxjs/operators'
import makeWebSocketObservable, {
  GetWebSocketResponses,
  // WebSocketPayload = string | ArrayBuffer | Blob
  WebSocketPayload,
  normalClosureMessage,
} from 'rxjs-websockets'

// this subject queues as necessary to ensure every message is delivered
const input$ = new QueueingSubject<string>()

// queue up a request to be sent when the websocket connects
input$.next('some data')

// create the websocket observable, does *not* open the websocket connection
const socket$ = makeWebSocketObservable('ws://localhost/websocket-path')

const messages$: Observable<WebSocketPayload> = socket$.pipe(
  // the observable produces a value once the websocket has been opened
  switchMap((getResponses: GetWebSocketResponses) => {
    console.log('websocket opened')
    return getResponses(input$)
  }),
  share(),
)

const messagesSubscription: Subscription = messages.subscribe(
  (message: string) => {
    console.log('received message:', message)
    // respond to server
    input$.next('i got your message')
  },
  (error: Error) => {
    const { message } = error
    if (message === normalClosureMessage) {
      console.log('server closed the websocket connection normally')
    } else {
      console.log('socket was disconnected due to error:', message)
    }
  },
  () => {
    // The clean termination only happens in response to the last
    // subscription to the observable being unsubscribed, any
    // other closure is considered an error.
    console.log('the connection was closed in response to the user')
  },
)

function closeWebsocket() {
  // this also caused the websocket connection to be closed
  messagesSubscription.unsubscribe()
}

setTimeout(closeWebsocket, 2000)

The observable returned by makeWebSocketObservable is cold, this means the websocket connection is attempted lazily as subscriptions are made to it. Advanced users of this library will find it important to understand the distinction between hot and cold observables, for most it will be sufficient to use the share operator as shown in the example above. The share operator ensures at most one websocket connection is attempted regardless of the number of subscriptions to the observable while ensuring the socket is closed when the last subscription is unsubscribed. When only one subscription is made the operator has no effect.

By default the websocket supports binary messages so the payload type is string | ArrayBuffer | Blob, when you only need string messages the generic parameter to makeWebSocketObservable can be used:

const socket$ = makeWebSocketObservable<string>('ws://localhost/websocket-path')
const input$ = new QueueingSubject<string>()

const messages$: Observable<string> = socket$.pipe(
  switchMap((getResponses: GetWebSocketResponses<string>) => getResponses(input$)),
  share(),
)

Reconnecting on unexpected connection closures

This can be done with the built-in rxjs operator retryWhen:

import { Subject } from 'rxjs'
import { switchMap, retryWhen } from 'rxjs/operators'
import makeWebSocketObservable from 'rxjs-websockets'

const input$ = new Subject<string>()

const socket$ = makeWebSocketObservable('ws://localhost/websocket-path')

const messages$ = socket$.pipe(
  switchMap((getResponses) => getResponses(input$)),
  retryWhen((errors) => errors.pipe(delay(1000))),
)

Alternate WebSocket implementations

A custom websocket factory function can be supplied that takes a URL and returns an object that is compatible with WebSocket:

import makeWebSocketObservable, { WebSocketOptions } from 'rxjs-websockets'

const options: WebSocketOptions = {
  // this is used to create the websocket compatible object,
  // the default is shown here
  makeWebSocket: (url: string, protocols?: string | string[]) => new WebSocket(url, protocols),

  // optional argument, passed to `makeWebSocket`
  // protocols: '...',
}

const socket$ = makeWebSocketObservable('ws://127.0.0.1:4201/ws', options)

JSON messages and responses

This example shows how to use the map operator to handle JSON encoding of outgoing messages and parsing of responses:

import { Observable } from 'rxjs'
import makeWebSocketObservable, { WebSocketOptions } from 'rxjs-websockets'

function makeJsonWebSocketObservable(
  url: string,
  options?: WebSocketOptions,
): Observable<unknown> {
  const socket$ = makeWebSocketObservable<string>(url, options)
  return socket$.pipe(
    map((getResponses: GetWebSocketReponses<string>) => (input$: Observable<object>) =>
      getResponses(input$.pipe(map((request) => JSON.stringify(request)))).pipe(
        map((response) => JSON.parse(response)),
      ),
    ),
  )
}

The function above can be used identically to makeWebSocketObservable only the requests/responses will be transparently encoded/decoded.