6

Estoy haciendo una especie de canal de comunicaciones, que envía peticiones y recibe respuestas.

La función que envía una petición, quiero que devuelva una Promise, que será resuelta por la función de recepción (interna de la clase).

Y ahí me he atascado:

class IOChannel {
  constructor( channel ) {
    this.$channel = channel;
    this.$queue = { };

    channel.on( 'message', ( data ) => {
      if( 'error' in data ) {
        // Es una respuesta con un error. // <- ¿ ?
      } else if( 'result' in data ) {
        // Es una respuesta. // <- ¿ ?
    } );
  }

  $nextId( ) { ... }

  async sendRequest( params ) {
    this.$channel.send( {
      id: this.$nextId( ),
      params: params
    } );

    return new Promise( ); // <- ¿ ?
  }
}

Tanto las solicitudes salientes como las respuestas entrantes están relacionadas mediante el atributo id del mensaje. La idea era usar la variable this.$queue como punto de unión entre las solicitudes salientes y las respuestas entrantes, basándome en ese atributo id.

Pero no se me ocurre como relacionar la función callback de respuesta a un message con la Promise generada en la función sendRequest( ).

¿ Como tengo que modificar mi código para que la función sendMessage( ) devuelva una promesa válida, que será resuelta o rechazada por mi función callback ?

Edito

La intención es invocar esa función sendRequest( ) desde manejadores de rutas de express:

const express = require( 'express' );

const app = express( );
const ioChannel = new IOChannel( process );

app.post( '/login', body-parser, function( req, res ) {
  const view = await ioChannel.send( {
    login: req.body.userId,
    pass: req.body.pass
  } );

  res.status( view ? 200 : 401 ).end( );
}
10
  • Quieres que sendRequest devuelva una promesa para luego resolverla o rechazarla dentro del callback de onmessage? Commented el 7 ago. 2020 a las 18:26
  • @RicardoCuaresma Exacto. Estoy muy mucho verde con las promesas, era feliz con las antiguas callbacks :-(
    – Trauma
    Commented el 7 ago. 2020 a las 18:27
  • Podrías mostrar a donde invocas a sendRequest?, las funciones async una vez las invocas devuelven una promesa, cuando el método async termina de ejecutarse.si no lanzas una excepcion, esa promesa se termina resolviendo, eso por el objeto Promise que devuelves en tu codigo actual, lo digo por si aun no lo has tenido en cuenta Commented el 7 ago. 2020 a las 18:33
  • 1
    @JoseHermosillaRodrigo Bueno, channel pertenece a la API de node y no lo puedo cambiar :-) Le echo un vistazo al enlace que indicas, gracias :-)
    – Trauma
    Commented el 7 ago. 2020 a las 19:18
  • 1
    Cierto! Creo que entendí bien tú pregunta ahora. Creo que esta genial respuesta en el sitio de Inglés es lo que buscas. Muy parecido a la idea que tienes de guardar un id.stackoverflow.com/a/48444103 Commented el 7 ago. 2020 a las 19:41

2 respuestas 2

2

Creo que lo que necesitas es esto:

class IOChannel {
  constructor( channel ) {
    this.$channel = channel;
    this.$queue = { };
    
  }

  $nextId( ) { ... }

  async sendRequest( params ) {
    this.$channel.send( {
      id: this.$nextId( ),
      params: params
    } );

    return new Promise((resolve, reject) => {
      this.$channel.on( 'message', ( data ) => {
      if( 'error' in data ) {
        // Es una respuesta con un error. // <- ¿ ?
        reject(data.error);
      } else if( 'result' in data ) {
        // Es una respuesta. // <- ¿ ?
        resolve(data.result);
    } );
    });
  }
}

1

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 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...');

Tu Respuesta

By clicking “Publica tu respuesta”, you agree to our terms of service and acknowledge you have read our privacy policy.

¿No es la respuesta que buscas? Examina otras preguntas con la etiqueta o formula tu propia pregunta.