Building a Resilient WebSocket Observable

Hello! Welcome back to our series on "Real-World Architectural Patterns."

In our last lesson, we mastered the switchMap operator to build a robust type-ahead search feature. We saw how RxJS elegantly handles user-initiated, asynchronous "pull" operations, solving tricky issues like race conditions.

Today, we shift our focus from user-initiated events to managing persistent, server-pushed data streams. Our learning outcome is to wrap the WebSocket API in a custom Observable that manages its connection lifecycle and message stream.

WebSockets are a cornerstone of modern real-time applications, from chat apps and live dashboards to collaborative editing tools. By wrapping the native WebSocket API in an Observable, we can integrate these real-time data streams seamlessly into our reactive RxJS-powered applications.

We will explore two primary methods to achieve this:

1. The "From Scratch" Method: We'll build a wrapper using the new Observable() constructor. This will give you a fundamental understanding of how to bridge any non-RxJS asynchronous API into the reactive world.
2. The RxJS Built-in Method: We'll use the powerful webSocket factory function and WebSocketSubject, which provide a production-ready, feature-rich solution.

Understanding both approaches will give you deep insight and the flexibility to choose the right tool for any situation.

1. The "From Scratch" Approach: Building a Custom WebSocket Observable

Before diving into the RxJS-specific helpers, it's invaluable to understand how to wrap an external, event-based API yourself. This skill is transferable to many other scenarios, like wrapping third-party libraries or browser APIs that don't natively support Observables.

As the article states, a custom Observable acts as a translator. We need to map the "language" of the WebSocket API to the "language" of an RxJS Observable. This mapping is quite direct:

• Connection: The new WebSocket(url) call is made inside the Observable's subscriber function, making it lazy—the connection is only established when someone subscribes.
• Receiving Messages: The socket.onmessage event handler will call observer.next().
• Handling Errors: The socket.onerror event handler will call observer.error().
• Closing the Connection: The socket.onclose event handler will call observer.complete().
• Teardown Logic: When a consumer unsubscribes, we must close the socket to prevent memory and resource leaks. The function returned from the subscriber function is the perfect place for this cleanup, calling socket.close().

Now, let's see this pattern in a complete code example.

This pattern is fundamental. To reinforce it, here is a very short video clip showing the exact same technique.

This manual approach gives you complete control and a clear understanding of the Observable contract. However, it has one significant drawback: it creates a cold Observable. Every time you subscribe, a new WebSocket connection is created. In an application where multiple components need to listen to the same data stream, this is highly inefficient.

This is the problem that the built-in RxJS solution solves.

2. The RxJS Way: webSocket and WebSocketSubject

RxJS provides a specialized tool for this exact job: the webSocket factory function, which returns a WebSocketSubject. This is a powerful, production-ready implementation that addresses the shortcomings of our manual approach.

A WebSocketSubject is a Subject, meaning it has a dual nature:

• It's an Observable you can subscribe to for receiving messages from the server.
• It's an Observer you can push values to (using .next()) to send messages to the server.

Most importantly, it multicasts its source, meaning it maintains a single underlying WebSocket connection that is shared among all its subscribers.

For a more guided and visual explanation of these features, the following video provides an excellent deep dive.

As you saw, using WebSocketSubject is straightforward.

import { webSocket } from 'rxjs/webSocket';

// 1. Create the subject
const wsSubject = webSocket('wss://your-api.com/socket');

// 2. Subscribe to receive messages from the server
wsSubject.subscribe({
  next: msg => console.log('Message received:', msg),
  error: err => console.error('Error:', err),
  complete: () => console.log('Connection closed')
});

// 3. Use .next() to send messages to the server
// RxJS automatically runs JSON.stringify on the object
wsSubject.next({ action: 'subscribe', channel: 'updates' });

This is far more powerful than our custom Observable. It's bi-directional and intelligently manages the connection for us. For advanced scenarios, you can pass a configuration object to the webSocket function to customize serialization, deserialization, and hook into the open/close events, which is a topic covered in the video you just watched (from 16:41 onwards).

3. Comparison and When to Use Which

Let's summarize the differences in a table.

Feature	Custom new Observable()	webSocket() / WebSocketSubject
Connection Type	Cold. A new connection is created for every subscription.	Hot (Shared). One connection is shared among all subscribers of the same instance.
Direction	Unidirectional (read-only). It's an Observable, not an Observer.	Bi-directional. It's a Subject, so you can call .next() to send data.
Message Buffering	None. You cannot send messages.	Yes. Outgoing messages sent via .next() are buffered until the connection is open.
Reconnection	Must be implemented manually (e.g., using retry or retryWhen).	Must also be implemented manually, but the architecture is well-suited for it.
Use Case	Good for simple, read-only streams, educational purposes, or wrapping other non-standard event APIs.	The standard choice for complex, bi-directional WebSocket communication in most applications.

For virtually all real-world WebSocket needs in an RxJS application, WebSocketSubject is the correct and more powerful choice.

---

Conclusion

Today, we've thoroughly explored how to bring WebSocket streams into the RxJS ecosystem.

Key Takeaways:

• You can wrap any event-based API, like the native WebSocket, by creating a custom Observable with the new Observable() constructor. This involves mapping API events to observer calls and providing essential teardown logic.
• The manual approach creates a cold Observable, which is inefficient for shared data streams.
• RxJS provides the webSocket factory and WebSocketSubject as a robust, hot, and bi-directional solution for managing WebSocket communication.
• WebSocketSubject automatically handles connection sharing, buffers outgoing messages, and provides a simple API for both sending and receiving data, making it the preferred method for production applications.

Next Lesson Preview:

A WebSocket connection is not always stable; it can drop due to network issues. A robust application must be able to recover from this. In our next lesson, we will build directly on today's topic and learn how to implement a robust automatic reconnection strategy with exponential backoff for a failing Observable stream, making our WebSocket integration truly resilient.