Streaming Resources for a Chat with Web Sockets: Messages in a Glitch-Free World

The minimal example in the privous article was already comprehensive enough to discuss the use of Streaming Resources and their semantic details. With this article, I would like to go a step further and show a more practical use case. This is a simple chat client, allowing further details to be discussed.

Technically speaking, this chat example differs from the Timer in one crucial aspect: all messages received are relevant, not just the last one. This presents a small challenge in the signal-based world but can be solved.

A fork of a sample project from the Mozilla Developer Network is used as the chat server. This Node-based server is included in the sample project and its readme.md shows how to start it.

📂 Source Code

Big thanks to Alex Rickabaugh from the Angular team for reviewing the examples and for the insightful discussions about the proper usage of Streaming Resources for messaging.

This article is written using version 19.2.0-next.0. I will update the text to respect changes when necessary.

Consumer's Perspective

In our example, the Angular client initializes the chat with the chatConnection factory. It provides a structure representing the individual chat messages as a streaming resource and offers additional Signals and a send method:

@Component([…])
export class ChatResourceComponent {
  ResourceStatus = ResourceStatus;

  userName = signal('');
  chat = chatConnection('ws://localhost:6502', this.userName);
  messages = computed(() => this.chat.resource.value() ?? []);

  userNameInField = linkedSignal(() => this.chat.acceptedUserName());
  currentMessage = signal<string>('');

  send() {
    this.chat.send(this.currentMessage());
    this.currentMessage.set('');
  }

  join() {
    this.userName.set(this.userNameInField());
  }
}

The userName Signal passed to chatConnection triggers the Resource and establishes a connection with the chat server. The userNameInField and currentMessages Signals are bound to input fields. To establish a connection, the join function writes the value of userNameInField to the userName Signal.

The server can correct the requested username to ensure uniqueness. Therefore, userNameInField is implemented with linkedSignal. This Linked Signal overwrites the local value with the corrected value from the server, which is contained in the accpetedUserName Signal.

When the user wants to send a message, the application passes the currentMessage to the send method. It then resets the currentMessage so that the user can enter the next message.

Glitch-Free Property as Challenge at Streaming Resources

The chat example differs from the timer discussed in the previous article in one essential aspect: while the timer was only interested in the latest message from the stream, the chat is interested in all messages received so far, or at least all messages from a defined time window. In other words, the timer represents an object that changes over time. In contrast, the chat represents several events with messages the application has to take into account in their entirety.

The way we look at the timer here fits well with signals, especially since signals always reflect the most recent value. The glitch-free property, which ignores unnecessary intermediate values, also fits here: If the Resource were to change the timer from 0 to 1, from 1 to 2, and from 2 to 3 in one single task, the resource consumer would only see the value 3.

However, this behavior would be fatal in an event-based use, such as chat. Here, individual messages would be lost. This behavior can easily be tested by having the chat server send the same message several times in a row to the same client. Most of these repetitions would be lost in an implementation that only considers the last message received.

This consideration shows that, unlike Observables in RxJS, Signals are not well suited to representing events or messages! This presents us with a challenge when using streaming resources, which fortunately can be overcome by simply adjusting our perspective: If we let the Resource represent not only the current value but all messages received so far, or at least all messages in a time window that is of interest to us, we can again view this totality of messages as a value that changes over time.

In other words, we need to move the collection and management of individual messages to the Resource. This view also fits well with the observation in the previous article that in the Signals world, use case-specific and, thus, coarse-grained building blocks are used.

More on this: Angular Architecture Workshop (Remote, Interactive, Advanced)

Become an expert for enterprise-scale and maintainable Angular applications with our Angular Architecture workshop!

English Version | German Version

Messages and Protocol

The example defines several types for the individual messages that the client exchanges with the chat server:

export type ChatRequest =
  | {
      type: 'username';
      id: number;
      name: string;
    }
  | {
      type: 'message';
      id: number;
      text: string;
    };

export type ChatResponse =
  | {
      type: 'id';
      id: number;
    }
  | {
      type: 'username';
      id: number;
      name: string;
    }
  | {
      type: 'message';
      id: number;
      name: string;
      text: string;
    };

The protocol between the client and server is as follows:

1. The client establishes a web socket connection.
2. The server responds with a ChatResponse of type id, via which the client receives a unique session id.
3. The client sends a ChatRequest of type username to inform the user's name.
4. The server confirms this username with a ChatResponse of type username. If the desired username is already taken, the client receives a corrected username via this message, which the server creates by appending a number to the preferred one.
5. The client sends ChatRequests with its messages (type message). The server distributes this message to all clients as a ChatResponse of type message.

Representation of the Chat

The connection to the chat is represented by the example with the type ChatConnection:

export type SendFn = (message: string) => void;

export type ChatConnection = {
  resource: ResourceRef<ChatResponse[] | undefined>;
  connected: () => boolean;
  acceptedUserName: () => string;
  send: SendFn;
};

The heart of the ChatConnection is a Resource that represents all messages received so far in the form of an array. It also provides further status information as signals. This includes the connection status (connected) and the username (acceptedUserName) corrected by the server if necessary.

The send method allows the chat to send messages to the server. The implementation maps the passed string to a ChatRequest of type message.

Factory for the Chat

The factory for the chat is structured similarly to the one for the timer discussed previously. However, it initially prepares a few additional variables:

export function chatConnection(
  websocketUrl: string,
  userName: () => string
): ChatConnection {

  let connection: WebSocket;

  const connected = signal(false);
  const id = signal(0);
  const acceptedUserName = signal('');

  const request = computed(() => ({
    userName: userName(),
  }));

  const chatResource = resource({
    request,
    stream: async (params) => {
      // init web socket connection
      // handle and collect messages
      […]
    },
  });

  const send: SendFn = (message: string) => {
    const request: ChatRequest = {
      type: 'message',
      id: id(),
      text: message,
    };
    connection.send(JSON.stringify(request));
  };

  return {
    connected,
    resource: chatResource,
    acceptedUserName,
    send,
  };
}

These variables include the connection, which reflects the web socket connection established by the Streaming Loader, the previously mentioned connected and accpetedUserName Signals, and an id Signal representing the current user session. In the end, the factory returns a few of these Signals, the created Streaming Resource, and the send method as a ChatConnection. It does not publish the id Signal, as this is an implementation detail.

The streaming loader of the resource first creates an Array messages in which it collects the received chat messages:

const chatResource = resource({
  request,
  stream: async (params) => {
    const userName = params.request?.userName;

    let messages: ChatResponse[] = [];

    // 1. Create Signal representing the Stream
    const resultSignal = signal<StreamItem<ChatResponse>>({
      value: messages,
    });

    if (!userName) {
      return resultSignal;
    }

    // 2. Set up async logic updating the Signal
    connection = new WebSocket(websocketUrl, 'json');

    connection.addEventListener('open', (event) => {
      console.log('[open]');
      connected.set(true);
    });

    connection.addEventListener('message', (event) => {
      const value = JSON.parse(event.data) as ChatResponse;
      console.log('[message]', value);

      if (value.type === 'id') {
        id.set(value.id);
        sendUserName(value.id, userName, connection);
      }

      if (value.type === 'username' && value.id == id()) {
        acceptedUserName.set(value.name);
      }

      if (value.type === 'message' || value.type === 'username') {
        messages = [...messages, value];
        resultSignal.set({ value: messages });
      }
    });

    connection.addEventListener('error', (event) => {
      const error = event;
      console.log('[error]', error);
      resultSignal.set({ error });
    });

    // 3. Set up clean-up handler triggered by AbortSignal
    params.abortSignal.addEventListener('abort', () => {
      console.log('clean up!');
      connection.close();
      connected.set(false);
      id.set(0);
      acceptedUserName.set('');
    });

    // 4. Return Signal
    return resultSignal;
  },
});

function sendUserName(id: number, userName: string, connection: WebSocket) {
  const message: ChatRequest = {
    type: 'username',
    id: id,
    name: userName,
  };
  connection.send(JSON.stringify(message));
}

The rest of the implementation follows the four points discussed in the previous article. The prepared signal reflects the stream and contains the managed Array with the received messages.

The client receives information from the server via the configured event handlers. The message event takes care of the protocol described above. It adds all received messages to the messages Array. This must be done immutably, i.e., using a shallow copy, so the Signal perceives the change.

The open event sets connected to true, and the error event writes the current error to the stream. The AbortSignal’s abort event closes the web socket connection and resets the managed properties. At the end, the Streaming Loader returns the signal reflecting the stream as usual.

Conclusion

Signals in Angular always only represent the current value. Intermediate values for changes that follow one another in the same task are ignored (Glitch Free property). This is well suited for changing states, such as a counter, but can lead to data loss in event flows - such as a chat that should display all received messages.

The solution is to collect the messages received (in the time window of interest) in the Resource and publish the resulting array via the resource. This array, therefore, corresponds to a value that changes over time.

While this is an interesting pattern, of course, no one prevents us from using different approaches such as RxJS that directly represent the flow of events.