An rxjs websocket library with a simple and flexible implementation. Supports the browser and node.js.
- 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 provides the connection status as an observable.
- 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.
Install the dependency:
npm install -S rxjs-websockets
# the following dependency is recommended for most users
npm install -S queueing-subjectimport { QueueingSubject } from 'queueing-subject'
import websocketConnect from 'rxjs-websockets'
// this subject queues as necessary to ensure every message is delivered
const input = new QueueingSubject<string>()
// this method returns an object which contains two observables
const { messages, connectionStatus } = websocketConnect('ws://localhost/websocket-path', input)
// send data to the server
input.next('some data')
// the connectionStatus stream will provides the current number of websocket
// connections immediately to each new observer and updates as it changes
const connectionStatusSubscription = connectionStatus.subscribe(numberConnected => {
console.log('number of connected websockets:', numberConnected)
})
// the websocket connection is created lazily when the messages observable is
// subscribed to
const messagesSubscription = messages.subscribe((message: string) => {
console.log('received message:', message)
})
// this will close the websocket
messagesSubscription.unsubscribe()
// closing the websocket does not close the connection status observable, it
// can be used to monitor future connection status changes
connectionStatusSubscription.unsubscribe()messages is a cold observable, this means the websocket connection is attempted lazily when a subscription is made to the messages observable. 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 Angular example below.
This can be done with built-in rxjs operators:
const input = new QueueingSubject<string>()
const { messages, connectionStatus } = websocketConnect(`ws://server`, input)
// try to reconnect every second
messages.retryWhen(errors => errors.delay(1000)).subscribe(message => {
console.log(message)
})A custom websocket factory function can be supplied that takes a URL and returns an object that is compatible with WebSocket:
const { messages } = websocketConnect(
'ws://127.0.0.1:4201/ws',
this.inputStream = new QueueingSubject<string>(),
undefined,
(url, protocols) => new WebSocket(url, protocols)
)The API typings follow which show how to use all features including protocols:
export interface Connection {
connectionStatus: Observable<number>
messages: Observable<string>
}
export interface IWebSocket {
close(): any
send(data: string | ArrayBuffer | Blob): any
onopen?: (OpenEvent) => any
onclose?: (CloseEvent) => any
onmessage?: (MessageEvent) => any
onerror?: (ErrorEvent) => any
}
export declare type WebSocketFactory = (url: string, protocols?: string | string[]) => IWebSocket
export default function connect(
url: string,
input: Observable<string>,
protocols?: string | string[],
websocketFactory?: WebSocketFactory
): ConnectionThis example shows how to use the map operator to handle JSON encoding of outgoing messages and parsing of responses:
function jsonWebsocketConnect(url: string, input: Observable<object>, protocols?: string | string[]) {
const jsonInput = input.map(message => JSON.stringify(message))
const { connectionStatus, messages } = websocketConnect(url, jsonInput, protocols)
const jsonMessages = messages.map(message => JSON.parse(message))
return { connectionStatus, messages: jsonMessages }
}The following is a very simple example Angular 4 service that uses rxjs-websockets to expose the messages from the server as an observable and take input messages using a procedural API. In most cases it would be preferable to wire the input stream up directly from one or more source observables.
// file: server-socket.service.ts
import { Injectable } from '@angular/core'
import { QueueingSubject } from 'queueing-subject'
import { Observable } from 'rxjs/Observable'
import websocketConnect from 'rxjs-websockets'
import 'rxjs/add/operator/share'
@Injectable()
export class ServerSocket {
private inputStream: QueueingSubject<string>
public messages: Observable<string>
public connect() {
if (this.messages)
return
// Using share() causes a single websocket to be created when the first
// observer subscribes. This socket is shared with subsequent observers
// and closed when the observer count falls to zero.
this.messages = websocketConnect(
'ws://127.0.0.1:4201/ws',
this.inputStream = new QueueingSubject<string>()
).messages.share()
}
public send(message: string):void {
// If the websocket is not connected then the QueueingSubject will ensure
// that messages are queued and delivered when the websocket reconnects.
// A regular Subject can be used to discard messages sent when the websocket
// is disconnected.
this.inputStream.next(message)
}
}This service could be used like this:
import { Component } from '@angular/core'
import { Subscription } from 'rxjs/Subscription'
import { ServerSocket } from './server-socket.service'
@Component({
selector: 'socket-user',
templateUrl: './socket-user.component.html',
styleUrls: ['./socket-user.component.scss']
})
export class SocketUserComponent {
private socketSubscription: Subscription
constructor(private socket: ServerSocket) {}
ngOnInit() {
this.socket.connect()
this.socketSubscription = this.socket.messages.subscribe((message: string) => {
console.log('received message from server: ', message)
})
// send message to server, if the socket is not connected it will be sent
// as soon as the connection becomes available thanks to QueueingSubject
this.socket.send('hello')
}
ngOnDestroy() {
this.socketSubscription.unsubscribe()
}
}
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent