How to isolate an event for a specific user?

Question

Node.js

I use websockets and ws.
In the usual case, ws is a single specific connection, i.e. this code sends a response directly to the sender

wss.on('connection', ws => {
  ws.on('message', message => {
    const { event, payload } = message;
    ws.send(payload);
  });
})

I'm trying to make a subscription system using EventEmitter, and thus messages go out to all connections in general

const EventEmitter = require('events');
const pubsub = new EventEmitter();

wss.on('connection', ws => {
  pubsub.on('test', payload => {
    ws.send(payload);
  });
  ws.on('message', message => {
    const { event, payload } = message;
    pubsub.emit(event, payload);
  });
})

Accordingly, as soon as the event is completed, it is sent to all ws.
And how to limit in this case?

Answer 1

spoiler

const EventEmitter = require('events');
const pubsub = new EventEmitter();

wss.on('connection', ws => {
  pubsub.on('test', (payload, _ws) => {
    if (_ws !== ws) {
      return;
    }
    ws.send(payload);
  });
  ws.on('message', message => {
    const { event, payload } = message;
    pubsub.emit(event, payload, ws);
  });
})

const EventEmitter = require('events');
const pubsub = new EventEmitter();

wss.on('connection', ws => {
  ws.on('message', message => {
    const { event, payload } = message;
    pubsub.emit(event, payload, ws);
  });
});

 pubsub.on('test', (payload, _ws) => {
   _ws.send(payload);
});

const EventEmitter = require('events');
const pubsub = new EventEmitter();

wss.on('connection', ws => {
  ws.on('message', message => {
    const { event, payload } = message;
    pubsub.emit(event, { payload, ws });
  });
});

 pubsub.on('test', ({ payload, ws }) => {
   ws.send(payload);
});

const EventEmitter = require('events');
const pubsub = new EventEmitter();

wss.on('connection', ws => {
  ws.on('message', message => {
    const { event, payload } = message;
    pubsub.emit(event, payload, outcome => ws.send(outcome));
  });
});

 pubsub.on('test', (payload, send) => {
   send(payload);
});