Angular two way data… messaging

author’s authentic art

I really hate the term “state management”. It became a hot topic after a certain library (cough framework) hilariously overcomplicated a very simple thing. These days, Svelte partially brought back sanity with svelte stores and signals are now becoming increasingly popular. Despite all of this, component communication is one thing. Client/server communication is another. This is where RxJS + Angular services shine. It is still the unbeatable combination for me. And no, you don’t need NgRx or similar stuff. Even when things get more complex.

Actually, let’s make them more complex. Forget http calls. Let’s use web sockets. Constant two-way data flow, with multiple clients.

Project overview

Keeping things as simple as possible, server app will generate an “item” every few seconds and add it to internal list of items generated thus far:

// item object
{
  id: number,
  ownerId: string;
}

Whenever the client app connects to the server, it must receive that list of items. Whenever a new item is generated, all clients will receive it. The client app can select an item from the list and request an ownership over it.

If item doesn’t have an owner yet, server will update ownerId (based on socketId) and publish a refreshed list to all connected clients, along with a message that item X is now owned by client Y.

If item already has an owner, transaction will be rejected, and server will publish that message to all clients.

Project setup

So we need two projects. A (web socket) server and (Angular) client app. I’ve chosen socket io for this purpose. I don’t normally write backend stuff in JS, so I‘m not really familiar with it. Take my implementation with a grain of salt.

Actually, take it with a mountain of salt. This is a dirty proof of concept. I’m skipping some “best practices” for brevity and clarity. You may find a simpler way to do the same thing.

Server

Create a new NodeJS project with npm init. Proceed with adding dependencies:

npm install express@4
npm install socket.io

After that, create app.js and file feel free to copy my code below. I’ll go through every line afterwards:

import express from 'express';
import { createServer } from 'node:http';
import { Server } from 'socket.io';

const app = express();
const server = createServer(app);
const io = new Server(server, { cors: {origin: '*'}});

let items = [];

setInterval(() => {
    const item = { id: items.length, ownerId: '' };
    items.push(item);
    io.emit('new-item', item);
}, 3000);

io.on('connection', (socket) => {
  console.log(`user ${socket.id} connected`);
  
  socket.emit('all-items', items);

  socket.on('own-item', (id) => {
    const requestedItem = items.find(x => x.id === id);
    if(requestedItem && requestedItem.ownerId === '') {
      requestedItem.ownerId = socket.id;
      io.emit('message', `item ${id} now owned by ${socket.id}`);
      io.emit('all-items', items);
    }
    else {
      io.emit('message', `item ${id} already owned by ${requestedItem.ownerId}`);
    }
  });

  socket.on('disconnect', () => {
    console.log(`user ${socket.id} disconnected`);
  });
});

server.listen(3000, () => {
  console.log('server running at http://localhost:3000');
});

Ignoring the first 6 declarations which you can find in the documentation, let’s break down the logic here:

// store generated items here
let items = [];

// generate a new item every 3 seconds and emit it
setInterval(() => {
    const item = { id: items.length, ownerId: '' };
    items.push(item);
    io.emit('new-item', item);
}, 3000);

Note the io.emit(‘new-item’, item); line. This will emit an item to all connected sockets that are listening the “new-item” channel. The official term is event, but I’ll call it channel here, since it is a two-way street. Moving on:

// this gets invoked every time a new socket is connected,
// so we apply same config to all sockets
io.on('connection', (socket) => {
  ...
});

Going line by line, the inner next step is:

socket.emit('all-items', items);

This will emit all items to the newly connected socket only. Other ones already have this data, so there’s no point to send it again. Remember:

io.emit('channel', data);      // emit message to all sockets
socket.emit('channel', data);  // emit message to a specific socket
socket.on('channel', data);    // await message from a specific socket

Moving on:

// when client sends "own item request" this will get invoked
socket.on('own-item', (id) => {
    // find an item and checked if it's already owned
    const requestedItem = items.find(x => x.id === id);
    if(requestedItem && requestedItem.ownerId === '') {
      requestedItem.ownerId = socket.id;

      // if ownership is assigned, inform everyone
      // by sending an updated list of items, and a message
      io.emit('message', `item ${id} now owned by ${socket.id}`);
      io.emit('all-items', items);
    }
    else {
      // inform everyone that transaction is denied
      io.emit('message', `item ${id} already owned by ${requestedItem.ownerId}`);
    }
  });

and the last few lines should be self explanatory.

io.on('connection', (socket) => {
  ...
  socket.on('disconnect', () => {
    console.log(`user ${socket.id} disconnected`);
  });
});

// run server
server.listen(3000, () => {
  console.log('server running at http://localhost:3000');
});

Client

Create an Angular app and add a package:

npm install socket.io-client

Time to enter TS land and define our models:

// the thing that comes from the server app
export interface IItem {
    id: number;
    ownerId: string;
}

// utility for displaying if item is selected
export interface IItemView {
  item: IItem;
  selected: boolean;
}

Note that the server app runs on http://localhost:3000. Create a new service with ng g s name-it-whatever-you-like.

Same as above, I’ll paste whole content of the service. Explanation in the comments.

import { Injectable } from "@angular/core";
import { BehaviorSubject } from 'rxjs';
import { io } from "socket.io-client";

@Injectable()
export class WebsocketService {
    private items$: BehaviorSubject<IItem[]> = new BehaviorSubject([]);
    private messages$: BehaviorSubject<string[]> = new BehaviorSubject([]);
    private socket;
    
    constructor() {
        // connect to server
        this.socket = io('http://localhost:3000');
        
        // listen for channels
        // receive one message, and add it to the list
        this.socket.on('message', (message) => {
            this.messages$.next(this.messages$.getValue().concat(message));
        });

        // when server sends updated list (ownership assigned)
        // overwrite the old list
        this.socket.on('all-items', (items) => {
            this.items$.next(items);
        });

        // when server sends newly created item
        // just add new item
        this.socket.on('new-item', (item) => {
            this.items$.next(this.items$.getValue().concat(item));
        });
    }

    // send ownership request... duh
    public sendOwnershipRequest(itemId: number) {
        this.socket.emit('own-item', itemId);
    }

    // expose observables to components
    messages() {
        return this.messages$.asObservable();
    };

    items() {
        return this.items$.asObservable();
    }
}

Moving on to app.component.ts (no point in creating new component):

import { Component, OnInit } from '@angular/core';
import { IItem, WebsocketService } from './ws.service';
import { Observable, map } from 'rxjs';

@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent implements OnInit {
  // inject service
  constructor(private ws: WebsocketService) {}

  
  private selectedItem: IItemView;
  items$ = new Observable<IItemView[]>();
  messages$ = new Observable<string[]>();
  
  // utility function to map view data
  private mapToView = (xs: IItem[]) =>
    xs.map(x => ({item: x, selected: false} as IItemView))
  
    ngOnInit() {
    this.messages$ = this.ws.messages();
    this.items$ = this.ws.items().pipe(map(x => this.mapToView(x)));
  }

  // when item is selected, deselect other items
  selectItem(iv: IItemView) {
    this.items$ = this.items$.pipe(map(ivs => {
      return ivs.map(i => {
        const selected = i.item.id === iv.item.id;
        if(selected) {
          this.selectedItem = iv;
        }
        return {...i, selected: selected }
      })
    }))
  }

  requestOwnership() {
    this.ws.sendOwnershipRequest(this.selectedItem.item.id);
  }
}

Component html, loop items and messages. Make every item clickable:

<button (click)="requestOwnership()">Request Ownership</button>

<div class="flex">
    <section>
        <h4>Items</h4>
        <div
            *ngFor="let i of items$ | async"
            (click)="selectItem(i)"
            [ngClass]="{'selected': i.selected, 'owned': i.item.ownerId !== ''}">
            Item: {{ i.item.id }} | Owner: {{ i.item.ownerId }}
        </div>
    </section>
    <section>
        <h4>Messages</h4>
        <div *ngFor="let m of messages$ | async">
            {{ m }}
        </div>
    </section>
</div>

And the bare minimum CSS to visually track item states:

.flex {
    display: flex;
}

.selected {
    border: 1px solid black;
}

.owned {
    background-color: gray;
}

That’s pretty much it. Run both projects and see the effects. Open up Angular app in 2 or more different tabs (or browsers) and you’ll see that changes are applied to all instances, almost instantly. Remember, we’re dealing with the real-time, multi user environment here.

The conclusion?

I really hate the term “state management”. This component has zero idea if it is dealing with http or ws protocol. There’s that item selection part, but even that logic can be moved to the service without hassle. RxJS provides a really good abstraction layer. You can even mix http calls with ws. Have fun experimenting.

Thanks for reading :)