Home / Questions / 380419
Accepted
Trauma
Asked: 2020-08-08 10:20:08 +0800 CST 2020-08-08 10:20:08 +0800 CST

Resolving promises from different functions

  • 772

I am making a kind of communications channel , which sends requests and receives responses.

The function that sends a request, I want it to return a Promise, which will be resolved by the receiving function (internal to the class).

And that's where I got stuck:

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( ); // <- ¿ ?
  }
}

Both outgoing requests and incoming responses are related by the idmessage attribute. The idea was to use the variable this.$queueas a join point between outgoing requests and incoming responses, based on that attribute id.

But I can't think of how to relate the response callbackmessage function to the one Promisegenerated in the function sendRequest( ).

How do I have to modify my code so that the function sendMessage( )returns a valid promise, which will be resolved or rejected by my callback function ?

edit

The intent is to call that function sendRequest( )from route handlers 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( );
}
javascript

2 Answers

  1. I think what you need is this:

    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);
    } );
    });
  }
}

    • 2
  2. Best Answer

    I've created a simple implementation with the help of this answer that I think may be useful, although it may be far from something usable in production.

    The idea is to encapsulate the queue logic "resolvers"so that it doesn't have to be handled outside of the public API #on.

    It also takes into account that there may be different ones listenersfor each event and the promise that the method returns sendresolves when all the listeners finish their task.

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

    • 1