I'm developing a simple message server app, using WebSockets and NestJs gateways. Server receives an event to indentify a listener with name, and then sends websocket updates to that listener.
Right now I'm using subject/observable approach, server after listen-as event returns a new observable, which filters all upcoming messages:
export class ListenAsDto {
  name: string;
}

export interface PersonalisedMessage {
  from: string;
  title: string;
  content: string;
}

export class MessagesService {
  // ...
  listenAs(listener: ListenAsDto): Observable<WsResponse<PersonalisedMessage>> {
    return this.messageObservable.pipe(
      filter((item) => item.to === listener.name),
      map((item) => {
        const copy = { ...item };
        delete copy.to;
        return { event: INBOX_MESSAGE_NAME, data: copy };
      }),
    );
  }
}

But this only works for the first time, after another request client now receives both updates from 2 users at the time.
Is there a way to stop previous observable for current user an start new?
EDIT: I'm using ws platform

Is there a way to stop old observable and set new after event in Nest Gateways?
Found a solution by using a native WebSocket instance. It passes the same object if request was made by same client. So, using a WeakMap, we can manually track all subscriptions objects, and unsubscribe them, if we get the same WebSocket client
export class MessagesService {
  private subs = new WeakMap<WebSocket, Subscription>();

  listenAs(ip: WebSocket, dto: ListenAsDto) {
    if (this.subs.has(ip)) this.subs.get(ip).unsubscribe();
    // 'createObservable' is basically question`s 'listenAs' function
    const obs = this.createObservable(dto);
    this.subs.set(
      ip,
      obs.subscribe({
        next(value) {
          ip.send(JSON.stringify(value));
        },
      }),
    );

    return dto;
  }
}

