RxJS and WebSocket

What is RxJS?

RxJS (Reactive Extensions for JavaScript) is a library for reactive programming. It lets you handle asynchronous operations and events as streams, written in a declarative style.

The three core concepts are Observable, Observer, and Subject.

Observable and Observer

An Observable is a stream that emits data. An Observer subscribes to that stream and executes a callback each time data arrives.

observable$.subscribe(value => {
  console.log(value); // called every time data is emitted
});

What is WebSocketSubject?

webSocket() in RxJS returns a WebSocketSubject. A Subject extends Observable and has both the ability to receive (Observable) and send (Observer).

• socket$.subscribe(...) → acts as an Observable, receiving messages from the server
• socket$.next(...) → acts as an Observer, sending messages to the server

This single object handles two-way communication — that's what makes Subject powerful.

Code

const { webSocket } = rxjs.webSocket;

// Create a WebSocketSubject
const socket$ = webSocket("wss://ws.postman-echo.com/raw");

// Subscribe to receive messages from the server
socket$.subscribe(msg => {
  document.getElementById("log").innerHTML += `<p>Received: ${msg}</p>`;
});

// Send the user's input via next()
const txt = document.getElementById("textForWebsocket").value;
socket$.next(txt);

Demo

Type a message and press Send. It will be sent to an echo server and the response will appear below.