Building a Robust Real-Time System with RxJS WebSocket

Building a robust real-time system with RxJS WebSocket requires a solid understanding of how to handle WebSocket connections and events.

To establish a connection, you can use the WebSocket API, which is supported by most modern browsers.

In the previous section, we demonstrated how to create a WebSocket connection using RxJS's WebSocketSubject. This connection is then used to send and receive messages in real-time.

By using RxJS WebSocket, you can create a robust real-time system that can handle a large number of connections and events.

You might enjoy: How Do Websockets Improve Real-time Communication in Web Applications

Setting up the WebSocket connection is a crucial step in creating a robust and reliable communication channel. The setUpWebSocketSubjectConfig method sets up the configuration for the WebSocketSubject, which includes the close and open observers.

The close observer is a callback function that runs whenever the socket is closed, and it attempts to reconnect the socket to maintain a continuous connection. This ensures that the socket stays connected even in the event of a disconnection.

Connecting to the socket is a straightforward process that involves creating a socket using the configured settings and returning a stream of received messages. This stream is then tapped and pushed into a messages Subject to decouple the received messages from the socket connection and disconnection.

You might like: Websocket Close Code

Connecting to the socket is pretty simple. The socket is created, using the created configuration, and it returns the stream of messages received.

These messages are tapped, and pushed into a messages Subject, which decouples the received messages from the socket connection and disconnection.

No matter how many times the socket connects or disconnects, it just keeps pushing messages into the same messages Subject, providing continuity of the stream.

The code that handles clients subscribing sends messages received from the messages Subject back to the client components, ensuring they receive the stream of messages without worrying about the socket's connection status.

You might like: Websocket vs Socket

When setting up the connection, it's essential to choose the right architecture to ensure seamless communication between your application and the server.

To isolate interactions with the WebSocketSubject, create a separate service that handles all WebSocket-related tasks.

This service should include the following methods:

The connect method should be called from your root component to establish the connection.

To receive the most recent values, subscribe to the messages observable in your Angular component, but don't subscribe directly to it. Instead, process the incoming messages from the server and do a transformation of the current observable map(rows => rows.data) first. This will result in the liveData$ observable.

To handle errors, use the RxJS catchError operator, and for logging purposes, use the tap operator.

For your interest: When to Use Websockets

Implementing reconnection strategies is crucial when working with WebSockets, as it ensures that your application remains functional even when the connection is lost.

To achieve this, you can leverage the retry function provided by RxJS, as demonstrated in Example 1. This allows you to monitor the connection status and attempt reconnection when necessary.

In a scenario with multiple UI instances connected to a WebSocket server, you'll need to assign a unique WebSocket session ID to each connection or UI instance to distinguish the initiator of changes. This is facilitated by an HttpInterceptor that passes the session ID in the header of every API call.

When the connection breaks, each component will need to re-subscribe to the new connection and send the subscribe message to the server. To circumvent this issue, you can implement a solution using a higher-order ReplaySubject, as shown in Example 2.

This approach effectively abstracts the WebSocket connection dependency from individual components, allowing each component to subscribe to its respective topics without worrying about the real-time status of the WebSocket connection.

To handle topic subscription in reconnection or not connected WebSocket scenarios, you can use a ReplaySubject with a dummy inner observable, as described in Example 2. This allows the WebSocket service to allow a subscription even when the connection is down, and then replace it with an actual observable once the connection is established.

The code handling re-connection is simple and just repeatedly tries to connect until it gives up, as seen in Example 3. However, this could be enhanced to notify the user if too many retries are attempted.

In a real-time world, the reconnection capability is a must, and you can attempt to reconnect every 2 seconds, as shown in Example 4. The trick is to intercept the closure of the socket and retry the connection.

Expand your knowledge: Asp.net Websocket Example

To intercept the closure of the connection, you can use the WebSocketSubjectConfig, as demonstrated in Example 5. This allows you to customize the socket lifecycle and display a custom message when the connection is closed.

To retry the connection, you can combine the retryWhen operator with the delayWhen operator, as shown in Example 6. This will resubscribe to a subject conditionally after it completes and set the delay between one connection and another.

The entire management of the WebSocket connection is done in an Angular service, ensuring a single connection to the server shared with all components that subscribe to server events.

This service is responsible for re-connecting when the socket disconnects, providing a seamless experience for users.

In case of a lost connection, the socket will be closed and the WebSocketSubject will no longer emit values, which is not the expected behavior in a real-time world. To handle this, the reconnection capability is a must, and the service can attempt to reconnect every 2 seconds.

The WebSocketSubjectConfig can be used to intercept the closure of the connection and display a custom message, making it easier to handle unexpected disconnections.

For your interest: Websocket Service

Managing a WebSocket connection is crucial for real-time applications, and it's essential to handle disconnections properly.

The entire management of the WebSocket connection is done in an Angular service, which ensures a single connection to the server shared among all components.

To prevent a lost connection from causing issues, you can intercept the closure of the socket and retry the connection.

The WebSocketSubjectConfig is responsible for customizing the socket lifecycle, including the closure of the connection.

Intercepting the closure event is possible by passing a WebSocketSubjectConfig object to the webSocket function, which allows you to display a custom message or retry the connection.

In the case of a disconnection, you can attempt to reconnect each 2 seconds, for example.

Reconnecting is a must in most cases, especially in real-time applications where a lost connection can cause unexpected behavior.

Managing the SocketService is crucial for a seamless WebSocket connection. It's done in an Angular service, ensuring a single connection to the server shared with all components that subscribe to server events.

This approach also handles reconnection when the socket disconnects. The SocketService takes care of reconnecting the socket, so your components don't have to worry about it.

The SocketService is responsible for creating the WebSocket connection using the created configuration. It then returns the stream of messages received from the server.

These messages are tapped and pushed into a messages Subject, which decouples the received messages from the socket connection and disconnection. This means that no matter how many times the socket connects or disconnects, it just keeps pushing messages into the same messages Subject.

The messages Subject ensures continuity of the stream, even if the socket disconnects and reconnects. This is achieved by pushing messages into the same Subject, regardless of the socket's connection status.

By using a messages Subject, you can easily handle clients subscribing to receive messages from the Subject. This way, the client components simply receive the stream of messages and don't care about the socket's connection status.

Curious to learn more? Check out: Websocket Stream

Error handling is crucial in managing connections, as it allows you to report custom errors and close the connection while still providing the server with the cause.

You can report a custom error by calling the error method on the subject, which will send specific messages when an event is fired.

This technique is useful when you have separate services with different WebSocket endpoints, running on separate machines with only a GUI combining them together.

A single Gateway can communicate with the services and manipulate streams separately from the client using multiplexing.

For your interest: Websocket Error

You can handle real-time updates in your Angular application without adding a third-party dependency, which can increase your bundle size and affect performance.

RxJS surprisingly ships with a WebSocketSubject, a wrapper around the w3c-webSocket object available in the browser, allowing you to both send and receive data via WS connection.

This is a great alternative to using a third-party package, considering compatibility, versions management, reliability, active support, and maintainability.

RxJS also provides a cool multiplexing feature, which is useful when you want to listen only to specific events coming from the server, keeping the same stream open and routing messages to the adequate observer.

You can use RxJS's WebSocketSubject to handle real-time updates in your Angular application, which is a wrapper around the native WebSocket object available in the browser.

This approach is preferable to adding a third-party dependency, as it reduces the bundle size and improves performance.

RxJS provides a multiplexing feature that allows you to listen to specific events coming from the server, while keeping the same WebSocket connection open.

The multiplex method produces an observable and accepts three parameters: subscription and unsubscription messages, and a function returning a subscription message.

The subscription message is sent on every subscription to the observable, and the unsubscription message is sent on every unsubscription to the observable, notifying the server to start or stop sending messages to the client.

Using RxJS's webSocket Subject is a great choice for implementing WebSockets in Angular, as it provides a stream of messages from the WebSocket.

Angular's RxJS library is a powerful tool for handling real-time updates, and it's a good idea to use it instead of the native browser WebSocket for a more efficient and scalable solution.

You might like: Can Audio Be Sent through Websockets

When sending messages to a server, they need to be in a format that can be easily understood by the server. This is where message serialization comes in.

The message will be serialized before sending it to the server. By default, the JSON.stringify method is used for this purpose.

You can customize the serialization function if you want to. To do this, you define your own function in the WebSocketSubjectConfig.

Deserialization is also an important process, where the server converts the received message back into a usable format. By default, the JSON.parse method is used for this.

Just like serialization, you can also define your own deserialization function in the WebSocketSubject if you need to customize it.

Intriguing read: Server Side Events vs Websockets

WebSocket configuration is crucial for establishing a robust and reliable connection with the server. The WebSocketSubjectConfig object is used to provide additional configuration to the WebSocketSubject created using the webSocket() operator.

The WebSocketSubjectConfig object has several configuration properties, including url, deserializer, serializer, openObserver, and closeObserver. The url property is the only required property, specifying the WebSocket connection URL.

For another approach, see: Websocket Create Connection

Here are the configuration properties of WebSocketSubjectConfig in a table:

The openObserver and closeObserver properties can be used to provide callback functions for when the socket is opened or closed. This is useful for reconnecting the socket when it is closed.

Security is a top priority when working with WebSockets, and rxJS provides a robust way to handle authentication.

To establish a secure connection, you can use a JWT token, which is validated by the backend service with the authentication service.

Including a JWT token in the WebSocket connection is a common practice, as seen in the example where a simple WebSocket service is developed to include a JWT token for authentication.

The AuthInterceptor plays a crucial role in this process, ensuring that the token is validated correctly.

Validating the JWT token is essential to authorize the connection, as shown in the example where the backend service validates the token with the authentication service.

This approach provides an additional layer of security to your WebSocket application.

Angular is a popular choice for building front-end applications, and in this case, it was used to create a demo project that connects to a back-end WebSocket.

The demo front-end project showcases connecting to the back-end web socket, and implements a simple chat messaging service, similar to many other applications.

Angular's component-based architecture makes it easy to break down the front-end into manageable parts, such as the Messenger component, which allows sending and receiving "chat" messages.

Two components subscribe to events in the demo app: the Messenger component and the connection monitoring component, which shows notifications from the server whenever another client connects or disconnects.

A component was also built to show the state of the WebSocket connection, and various statistics about how many subscriptions are active, the number of messages received, the number of times it's connected, and the number of retries when re-connecting.

Here are the different components in the demo app:

You have a few options when it comes to implementing WebSockets in Angular. My personal preference is to use RxJS, which provides a specialized Subject called webSocket.

This Subject is a wrapper around the native WebSocket, providing a stream of messages from the WebSocket. It's a great choice if you want to easily use RxJS in your Angular project.

One of the key benefits of using RxJS's webSocket Subject is that it allows you to easily manage WebSocket connections and messages in your Angular application.

Here are some of the benefits of using RxJS's webSocket Subject:

However, it's worth noting that you may also consider using the native browser WebSocket for your Angular project. This can be a good option if you're already familiar with it and don't need the additional features provided by RxJS's webSocket Subject.

Ultimately, the choice between RxJS's webSocket Subject and the native browser WebSocket will depend on your specific needs and preferences.

Angular was used for the front-end, and Nx was used to generate the project and build it out.

The demo front-end project showcases connecting to the back-end web socket and implements a simple chat messaging service.

A simple Angular application was built, broken down into a number of components for clarity.

Two components subscribe to events: the Messenger component, which allows sending and receiving "chat" messages, and the connection monitoring component, which shows notifications from the server whenever another client connects or disconnects.

A component shows the state of the WebSocket connection and various statistics about how many subscriptions are active, the number of messages received, the number of times it's connected, and the number of retries when re-connecting.

Here are the main aspects of the demo app:

In a demo project, a shared WebSocket is used to send subscription requests to the server and listen for notification events, which are piped back to the subscribing component.

This project also makes use of traditional REST, where some requests or user activity trigger notifications to be sent over the WebSocket to subscribing components.

To get started with a WebSocket example, we can look at a code snippet that imports the webSocket() function from the "rxjs/webSocket" module.

We create a new WebSocketSubject using the webSocket() function, specifying the generic type of string for the next notification values, and the connection URL.

The "first" message is emitted using the next() method before opening the connection by subscribing to the WebSocketSubject. All messages sent when the connection is closed are buffered until the connection is opened.

Here are the key steps to create a WebSocketSubject:

In this example, we use the "first" message to demonstrate the buffering of messages when the connection is closed.

To use the RxJS WebSocket package, you'll need to install it first. You can do this by calling the WebSocket factory function that produces the WebSocketSubject.

The WebSocketSubject can be used with both regular and secure WebSocket connections, denoted by ws and wss respectively. This allows you to choose the type of connection that best suits your needs.

To install, you need to call the WebSocket factory function that produces a special type of subject.

This function takes as a parameter the endpoint of your ws server, and you can use wss for a secure WebSocket connection.

You'll then have a ready-to-use subject that you should subscribe to, which will establish the connection with your endpoint and allow you to receive and send data.

As a WebSocketSubject is just a regular RxJS subject, you can send data to the WebSocket using the next method and register callbacks to process incoming messages.

To get started, you can subscribe to the liveUpdates$ observable in the component's template using the async pipe.

Check this out: Websocket Use Cases

The RxJS WebSocket Package is a great tool for developers. It's available for everyone, taken from the official RxJS documentation WebSocket page.

To get started, you can begin by surfing the RxJS website. This will give you a sense of what the package has to offer.

The RxJS WebSocket Package is designed for pushing messages to the server. It's a powerful tool that can help you build better websites and web applications.

You can start by thinking like an Observable, as suggested in the documentation. This will help you build a better understanding of how the package works.

The package is worth your time, as mentioned in the documentation. So, give it a try and see the benefits for yourself.