Many answers, and the HTTP API

So far one message got one answer. Some answers do not fit in one. A list the client wants to show as it arrives, item by item. An import of fifty cats, where the client wants to know at once that the request was accepted, and then hear about each cat as it is stored. And some news does not start with a socket at all: a cat registered at the front desk through the REST API should reach every screen just the same. This lesson covers all three.

An Observable of events

A handler can return an RxJS Observable of { event, data } objects. Nest subscribes to it and sends each value as its own event, to the sender, until the stream completes:

@SubscribeMessage('kennels:all')
all(): Observable<WsResponse<Kennel | number>> {
  const kennels = this.kennels.findAll();
  return concat(
    from(kennels).pipe(map((kennel) => ({ event: 'kennels:item', data: kennel }))),
    of({ event: 'kennels:end', data: kennels.length }),
  );
}

from() turns an array into a stream of its items, map() wraps each in a WsResponse, and concat() sends the closing event only after the items. A closing event matters: an event stream has no natural end on the client, which otherwise cannot tell "that was all" from "more is coming". Returning a plain array of { event, data } objects does not do this; an array is one value, sent as one acknowledgement.

Waiting for a promise

A handler can be async, or return a promise, and the acknowledgement is sent when it resolves. The socket stays free in the meantime: other messages from the same client are handled while one waits.

@SubscribeMessage('kennels:inspect')
async inspect(@MessageBody('kennel') kennel: string) {
  return (await this.inspections.latest(kennel)) ?? { error: `No inspection of ${kennel}` };
}

Acknowledging by hand

@Ack() gives the handler the client's acknowledgement callback itself. Call it when you choose, typically before slow work, so the client stops waiting, and then keep going. Once a handler asks for @Ack(), Nest no longer sends the return value as the acknowledgement: the callback is the only one, and only its first call counts.

@SubscribeMessage('kennels:clean')
clean(@MessageBody() kennels: string[], @Ack() ack: (receipt: { queued: number }) => void, @ConnectedSocket() client: Socket) {
  ack({ queued: kennels.length });
  for (const kennel of kennels) client.emit('kennels:cleaned', this.kennels.clean(kennel));
}

A gateway the rest of the app uses

A gateway is a provider, so a controller or a service can inject it. Give it a method that says what happened, and keep socket.io out of the callers:

// in the gateway
announce(kennel: Kennel): void {
  this.server.emit('kennels:opened', kennel);
}

// in a controller
@Post()
open(@Body() kennel: NewKennel): Kennel {
  const opened = this.kennels.open(kennel);
  this.kennelsGateway.announce(opened);
  return opened;
}

The HTTP response and the socket event are independent. The request is answered whether or not any socket is connected, and the event reaches whoever is connected at that moment. Events are not stored, so a client that connects a second later never sees it. That is the right behaviour for a live screen, and a reason to also give clients a way to fetch the current state when they connect.

Your task

The gateway and a REST controller share CatsService.

1. cats:stream answers with one cats:item event per cat, in order, then cats:end with how many there were.
2. cats:checkup with { id } acknowledges the vet's record once CatsService.checkup() resolves, or { error: 'Cat 9 not found' }.
3. cats:import with a list of { name, age } acknowledges { accepted } straight away, then stores each cat and sends the importer cats:imported with it.
4. A cat created with POST /cats is announced as cats:added to every connected socket, through a gateway method the controller calls.

Connect in the Socket mode, then send POST /cats from the Request mode and switch back to read the frame log.

When it fails

• ada received nothing, expected [{"event":"cats:item", ...}] while the client got an acknowledgement: the handler returned the array, or an array of { event, data }, which is one value.
• no acknowledgement for "cats:checkup" for cat 9 only: this.catsService.checkup(id) ?? { error } tests the promise, which is never null, so the handler returns the promise and it resolves to undefined. Await it, then fall back.
• no acknowledgement for "cats:import": the handler takes @Ack() and never calls it, so returning the receipt does nothing.
• ada received nothing after POST /cats: the controller stored the cat without calling the gateway. An HTTP request has no socket of its own, so the announcement goes through the gateway's server.

Remember

• An Observable of { event, data } sends one event per value; end a stream with an event of its own.
• An async handler is acknowledged when its promise resolves.
• @Ack() hands you the acknowledgement, and the return value stops being it.
• Inject the gateway where changes happen, and give it methods that name the news.