Skip to main content
se añadieron 237 caracteres en el cuerpo
Origen Enlace
class EventEmitterAbstractEventEmitter {
  constructor() {
    this.listeners = {};
  }
  
  on() {}
  
  off() {}
  
  send() {}
}

// EventEmitter
// Simula lo que ya hace "process" mediante process.send y process.on
class EventEmitter extends AbstractEventEmitter {
  constructor() {
    super();
  }
  
  on(key, callback) {
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, callback];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(c => c !== callback);
    } 
  }
  
  send(key, ...args) {
    for (const callback of (this.listeners[key] || [])) {
      setTimeout(callback.bind(null, ...args), 0);
    }
  }
}

// AsyncEventEmitter
// PD: Quizás usar herencia aquí no tenga mucho sentido
class AsyncEventEmitter extends EventEmitterAbstractEventEmitter {
  constructor() {
    super();
    
    this.nextId = 0;
    this.queue = new Map;
  }
  
  on(key, callback) {
    const callbackWrapper = (nextId, ...args) => {
      const task = this.getTask(nextId);
            
      if (!task) {
        return;
      }
      
      const { resolve, reject } = task;
      
      try {
        const result = callback(...args);

        resolve(result);
      } catch (error) {
        reject(error);
      }
    };
    
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, { callback, callbackWrapper }];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(({ callback: c }) => c !== callback);
    }  
  }
  
  
  getTask(nextId) {
    const task = this.queue.get(nextId);
    this.queue.delete(nextId);
    return task;
  }
  
  send(key, ...args) {
    const taskIds = [];
    
    for (const { callbackWrapper } of (this.listeners[key] || [])) {
      const taskId = this.nextId++;
      setTimeout(callbackWrapper.bind(null, taskId, ...args), 0);
      taskIds.push(taskId);
    }
    
    if (!taskIds.length) {
      return Promise.resolve();
    }
    
    return this.waitForTasks(taskIds);
  }
  
  waitForTasks(taskIds) {
    return Promise.all(taskIds.map(this.getTaskPromise.bind(this)));
  }
  
  getTaskPromise(nextId) {
      return new Promise((resolve, reject) => this.queue.set(nextId, { resolve, reject }));
  }
}

// Testing

const process = new AsyncEventEmitter();
const key = Symbol('my-event');
const keyError = Symbol('my-error-event');

process.on(key, () => {
  for(let i = 0; i < 100000; i++) {
    // Do nothing
  }
  
  console.log('End for loop');
  return 1;
});

process.on(key, () => {
  for(let i = 0; i < 1000000; i++) {
    // Do nothing
  }
  
  console.log('End 2nd for loop');
  return 2;
});


const onError = () => {
  throw new Error('Lo siento, me he equivocado, no volverá a pasar :/');
};

process.on(keyError, onError);
process.on(keyError, (...args) => 'Este callback será ejecutado, pero no se ejecutará el método #then de la promise que devuelve #send');

//process.off(keyError, onError);

process
  .send(key, 'Hello', 'World')
  .then((...args) => console.log('Processed', ...args));

process
  .send(keyError, 'Hello', 'World')
  .then(console.log)
  .catch(console.error);

console.log('Waiting for process...');
class EventEmitter {
  constructor() {
    this.listeners = {};
  }
  
  on(key, callback) {
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, callback];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(c => c !== callback);
    } 
  }
  
  send(key, ...args) {
    for (const callback of (this.listeners[key] || [])) {
      setTimeout(callback.bind(null, ...args), 0);
    }
  }
}

// AsyncEventEmitter
// PD: Quizás usar herencia aquí no tenga mucho sentido
class AsyncEventEmitter extends EventEmitter {
  constructor() {
    super();
    
    this.nextId = 0;
    this.queue = new Map;
  }
  
  on(key, callback) {
    const callbackWrapper = (nextId, ...args) => {
      const task = this.getTask(nextId);
            
      if (!task) {
        return;
      }
      
      const { resolve, reject } = task;
      
      try {
        const result = callback(...args);

        resolve(result);
      } catch (error) {
        reject(error);
      }
    };
    
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, { callback, callbackWrapper }];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(({ callback: c }) => c !== callback);
    }  
  }
  
  
  getTask(nextId) {
    const task = this.queue.get(nextId);
    this.queue.delete(nextId);
    return task;
  }
  
  send(key, ...args) {
    const taskIds = [];
    
    for (const { callbackWrapper } of (this.listeners[key] || [])) {
      const taskId = this.nextId++;
      setTimeout(callbackWrapper.bind(null, taskId, ...args), 0);
      taskIds.push(taskId);
    }
    
    if (!taskIds.length) {
      return Promise.resolve();
    }
    
    return this.waitForTasks(taskIds);
  }
  
  waitForTasks(taskIds) {
    return Promise.all(taskIds.map(this.getTaskPromise.bind(this)));
  }
  
  getTaskPromise(nextId) {
      return new Promise((resolve, reject) => this.queue.set(nextId, { resolve, reject }));
  }
}

// Testing

const process = new AsyncEventEmitter();
const key = Symbol('my-event');
const keyError = Symbol('my-error-event');

process.on(key, () => {
  for(let i = 0; i < 100000; i++) {
    // Do nothing
  }
  
  console.log('End for loop');
  return 1;
});

process.on(key, () => {
  for(let i = 0; i < 1000000; i++) {
    // Do nothing
  }
  
  console.log('End 2nd for loop');
  return 2;
});


const onError = () => {
  throw new Error('Lo siento, me he equivocado, no volverá a pasar :/');
};

process.on(keyError, onError);
process.on(keyError, (...args) => 'Este callback será ejecutado, pero no se ejecutará el método #then de la promise que devuelve #send');

//process.off(keyError, onError);

process
  .send(key, 'Hello', 'World')
  .then((...args) => console.log('Processed', ...args));

process
  .send(keyError, 'Hello', 'World')
  .then(console.log)
  .catch(console.error);

console.log('Waiting for process...');
class AbstractEventEmitter {
  constructor() {
    this.listeners = {};
  }
  
  on() {}
  
  off() {}
  
  send() {}
}

// EventEmitter
// Simula lo que ya hace "process" mediante process.send y process.on
class EventEmitter extends AbstractEventEmitter {
  constructor() {
    super();
  }
  
  on(key, callback) {
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, callback];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(c => c !== callback);
    } 
  }
  
  send(key, ...args) {
    for (const callback of (this.listeners[key] || [])) {
      setTimeout(callback.bind(null, ...args), 0);
    }
  }
}

// AsyncEventEmitter
class AsyncEventEmitter extends AbstractEventEmitter {
  constructor() {
    super();
    
    this.nextId = 0;
    this.queue = new Map;
  }
  
  on(key, callback) {
    const callbackWrapper = (nextId, ...args) => {
      const task = this.getTask(nextId);
            
      if (!task) {
        return;
      }
      
      const { resolve, reject } = task;
      
      try {
        const result = callback(...args);

        resolve(result);
      } catch (error) {
        reject(error);
      }
    };
    
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, { callback, callbackWrapper }];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(({ callback: c }) => c !== callback);
    }  
  }
  
  
  getTask(nextId) {
    const task = this.queue.get(nextId);
    this.queue.delete(nextId);
    return task;
  }
  
  send(key, ...args) {
    const taskIds = [];
    
    for (const { callbackWrapper } of (this.listeners[key] || [])) {
      const taskId = this.nextId++;
      setTimeout(callbackWrapper.bind(null, taskId, ...args), 0);
      taskIds.push(taskId);
    }
    
    if (!taskIds.length) {
      return Promise.resolve();
    }
    
    return this.waitForTasks(taskIds);
  }
  
  waitForTasks(taskIds) {
    return Promise.all(taskIds.map(this.getTaskPromise.bind(this)));
  }
  
  getTaskPromise(nextId) {
      return new Promise((resolve, reject) => this.queue.set(nextId, { resolve, reject }));
  }
}

// Testing

const process = new AsyncEventEmitter();
const key = Symbol('my-event');
const keyError = Symbol('my-error-event');

process.on(key, () => {
  for(let i = 0; i < 100000; i++) {
    // Do nothing
  }
  
  console.log('End for loop');
  return 1;
});

process.on(key, () => {
  for(let i = 0; i < 1000000; i++) {
    // Do nothing
  }
  
  console.log('End 2nd for loop');
  return 2;
});


const onError = () => {
  throw new Error('Lo siento, me he equivocado, no volverá a pasar :/');
};

process.on(keyError, onError);
process.on(keyError, (...args) => 'Este callback será ejecutado, pero no se ejecutará el método #then de la promise que devuelve #send');

//process.off(keyError, onError);

process
  .send(key, 'Hello', 'World')
  .then((...args) => console.log('Processed', ...args));

process
  .send(keyError, 'Hello', 'World')
  .then(console.log)
  .catch(console.error);

console.log('Waiting for process...');
Origen Enlace

He creado una implementación sencilla con la ayuda de esta respuesta que creo que puede ser útil, aunque quizás esté lejos de ser algo usable en producción.

La idea es encapsular la lógica de la cola de "resolvers", de forma que no se tenga que tratar fuera de la API pública #on.

Además tiene en cuenta que pueden existir diferentes listeners para cada evento y la promesa que devuelve el método send resuelve cuando todos los listeners terminen su tarea.

class EventEmitter {
  constructor() {
    this.listeners = {};
  }
  
  on(key, callback) {
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, callback];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(c => c !== callback);
    } 
  }
  
  send(key, ...args) {
    for (const callback of (this.listeners[key] || [])) {
      setTimeout(callback.bind(null, ...args), 0);
    }
  }
}

// AsyncEventEmitter
// PD: Quizás usar herencia aquí no tenga mucho sentido
class AsyncEventEmitter extends EventEmitter {
  constructor() {
    super();
    
    this.nextId = 0;
    this.queue = new Map;
  }
  
  on(key, callback) {
    const callbackWrapper = (nextId, ...args) => {
      const task = this.getTask(nextId);
            
      if (!task) {
        return;
      }
      
      const { resolve, reject } = task;
      
      try {
        const result = callback(...args);

        resolve(result);
      } catch (error) {
        reject(error);
      }
    };
    
    const listeners = this.listeners[key] || [];
    this.listeners[key] = [...listeners, { callback, callbackWrapper }];
  }
  
  off(key, callback) {
    if (!callback) {
      return delete this.listeners[key];
    }
    
    if (this.listeners[key]) {
      this.listeners[key] = this.listeners[key].filter(({ callback: c }) => c !== callback);
    }  
  }
  
  
  getTask(nextId) {
    const task = this.queue.get(nextId);
    this.queue.delete(nextId);
    return task;
  }
  
  send(key, ...args) {
    const taskIds = [];
    
    for (const { callbackWrapper } of (this.listeners[key] || [])) {
      const taskId = this.nextId++;
      setTimeout(callbackWrapper.bind(null, taskId, ...args), 0);
      taskIds.push(taskId);
    }
    
    if (!taskIds.length) {
      return Promise.resolve();
    }
    
    return this.waitForTasks(taskIds);
  }
  
  waitForTasks(taskIds) {
    return Promise.all(taskIds.map(this.getTaskPromise.bind(this)));
  }
  
  getTaskPromise(nextId) {
      return new Promise((resolve, reject) => this.queue.set(nextId, { resolve, reject }));
  }
}

// Testing

const process = new AsyncEventEmitter();
const key = Symbol('my-event');
const keyError = Symbol('my-error-event');

process.on(key, () => {
  for(let i = 0; i < 100000; i++) {
    // Do nothing
  }
  
  console.log('End for loop');
  return 1;
});

process.on(key, () => {
  for(let i = 0; i < 1000000; i++) {
    // Do nothing
  }
  
  console.log('End 2nd for loop');
  return 2;
});


const onError = () => {
  throw new Error('Lo siento, me he equivocado, no volverá a pasar :/');
};

process.on(keyError, onError);
process.on(keyError, (...args) => 'Este callback será ejecutado, pero no se ejecutará el método #then de la promise que devuelve #send');

//process.off(keyError, onError);

process
  .send(key, 'Hello', 'World')
  .then((...args) => console.log('Processed', ...args));

process
  .send(keyError, 'Hello', 'World')
  .then(console.log)
  .catch(console.error);

console.log('Waiting for process...');