Expose RequestHandler

[artur-ma]

Hey, I want to implement something like "fire and forget", this is some sort of POC

const client = new Pool(`https://google.com`)

class CustomRequestHandler extends RequestHandler {
  constructor(opts, callback) {
    super(opts, callback)
  }

  onBodySent(data) {
    this.callback(null, data, true);
  }
}

const request = (opts) => {
  const emitter = new events.EventEmitter();

  const chunkPromise = new Promise((resolve, reject) => { 
    emitter.on('error', reject)
    emitter.on('chunk', resolve)
    emitter.on('done', resolve)
  });

  const responsePromise = new Promise((resolve, reject) => { 
    emitter.on('error', reject)
    emitter.on('done', resolve)
  });

  client.dispatch(opts, new CustomRequestHandler(opts, (err, data, isChunk) => {
    if (err) emitter.emit('error')
    else if (isChunk) emitter.emit('chunk', data)
    else emitter.emit('done', data)
  }));

  return [chunkPromise, responsePromise]
}

const transports = [() => request({ path: '/', method: 'POST', body: 'test' })]

const send = async (cb) => {
  for (const transport of transports) {
    try {
      const [chunk, request] = transport();
      cb(await chunk)
      await request
      console.log('done');
    } catch(err) {
      console.error(err);
    }
  }
}

send((sentData) => console.log(`sent ${sentData}`))

maybe there is a simple way to implement, it, but in my example I need the RequestHandler to be exported, so I could extend it

ps. my idea is to resolve the promise once data was sent, but still wait for the response (to have a fallback if needed )

[mcollina]

I'm unsure if exposing those is something we should do - our API surface is getting bigger. I also have a feeling dispatch() might be too low level and we should not really expose it for end-users.

Wdyt @ronag?

[szmarczak]

We'll use dispatch in Got

[szmarczak]

Actually I think you're right, we can use request instead...

[artur-ma]

found another way to do it.
extend Agent + request

const events = require('events');
const { request, Agent } = require('undici')

function onBodySent (data) {
  // access to internal value `callback` of request handler
  this.callback(null, data, true);
}

class CustomAgent extends Agent {
  dispatch(opts, handler) {
    // patching RequestHandler instance
    handler.onBodySent = onBodySent;
    return super.dispatch(opts, handler)
  }
}

const url = new URL('https://google.com');
const dispatcher = new CustomAgent();

const httpRequest = () => {
  const emitter = new events.EventEmitter();

  const chunkPromise = new Promise((resolve, reject) => { 
    emitter.on('error', reject)
    emitter.on('chunk', resolve)
    emitter.on('done', resolve)
  });

  const responsePromise = new Promise((resolve, reject) => { 
    emitter.on('error', reject)
    emitter.on('done', resolve)
  });

  request(url, { path: '/', method: 'POST', body: 'test', dispatcher }, (err, data, isChunk) => {
    if (err) emitter.emit('error')
    else if (isChunk) emitter.emit('chunk', data)
    else emitter.emit('done', data)
  });

  return [chunkPromise, responsePromise]
}

const transports = [httpRequest]

const send = async (cb) => {
  for (const transport of transports) {
    try {
      const [chunk, request] = transport();
      cb(await chunk)
      await request
      break;
    } catch(err) {
      console.error(err);
    }
  }
}

send((sentData) => console.log(`sent ${sentData}`))

but, here im patching "RequestHandler" instance and i'm trying to access it's internal values (the callback)

[artur-ma]

if someone is looking for a solution, I ended up with this

class RequestHandlerWrap {
  constructor (originalHandler, initialTransferCallback) {
    this.originalHandler = originalHandler
    this.initialTransferCallback = initialTransferCallback
  }

  onConnect (abort, context) {
    return this.originalHandler.onConnect(abort, context)
  }

  onBodySent (chunk) {
    this.initialTransferCallback(chunk)

    if (typeof this.originalHandler.onBodySent === 'function') {
      return this.originalHandler.onBodySent(chunk)
    }
  }

  onHeaders (statusCode, headers, resume) {
    return this.originalHandler.onHeaders(statusCode, headers, resume)
  }

  onData (chunk) {
    return this.originalHandler.onData(chunk)
  }

  onComplete (trailers) {
    return this.originalHandler.onComplete(trailers)
  }

  onError (err) {
    return this.originalHandler.onError(err)
  }
}

class AgentWrap {
  constructor (originalDispatcher) {
    this.originalDispatcher = originalDispatcher
  }

  dispatch (opts, handler) {
    const { initialTransferCallback, ...originalOptions } = opts
    return this.originalDispatcher.dispatch(originalOptions, new RequestHandlerWrap(handler, initialTransferCallback))
  }

  get closed () {
    return this.originalDispatcher.closed
  }

  get destroyed () {
    return this.originalDispatcher.destroyed
  }

  close (callback) {
    return this.originalDispatcher.close(callback)
  }

  destroy (err, callback) {
    return this.originalDispatcher.close(err, callback)
  }
}


function _call(url, cb) {
    let isTransferStarted = false
    
    return new Promise((resolve, reject) => {
      request(url, {
        method,
        body,
        dispatcher: new AgentWrap(getGlobalDispatcher())
        initialTransferCallback: (data) => {
          if (!isTransferStarted) {
            isTransferStarted = true
            cb(null, data)
          }
        }
      }, (err, data) => {
        if (!isTransferStarted) {
          isTransferStarted = true
          cb(err, data)
        }

        if (err) {
          reject(err)
          return
        }

        handleResponse(data, path).catch(reject).then(resolve)
      })
    })
}