onBodySent is not called in mockDispatch

[tjhiggins]

Bug Description

onBodySent is not called when using the MockAgent

Reproducible By

import { DecoratorHandler, Dispatcher, fetch, MockAgent } from 'undici';

class LoggingHandler extends DecoratorHandler implements Dispatcher.DispatchHandlers {
  private requestBody = '';

  constructor(
    protected requestOptions: Dispatcher.DispatchOptions,
    protected handler: Dispatcher.DispatchHandlers,
  ) {
    super(handler);
  }

  onConnect(abort: (err?: Error) => void): void {
    if (this.handler.onConnect) {
      return this.handler.onConnect(abort);
    }
  }

  onBodySent(chunk: any, totalBytesSent: number) {
    try {
      this.requestBody += Buffer.from(chunk).toString();
    } catch (err) {
      console.debug('Failed to parse request body:', err);
    }
    if (this.handler.onBodySent) {
      this.handler.onBodySent(chunk, totalBytesSent);
    }
  }

  onComplete(trailers: string[] | null) {
    console.log('onComplete', this.requestBody); // In reality this would log to something like GCP with headers/response/status etc
    return this.handler.onComplete!(trailers);
  }
}

function createLoggingInterceptor() {
  return (dispatch: Dispatcher['dispatch']) => {
    return function Cache(requestOptions: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandlers) {
      return dispatch(requestOptions, new LoggingHandler(requestOptions, handler));
    };
  };
}

describe('onBodySent', () => {
  it('onBodySent should be called when using MockAgent', async () => {
    const mockAgent = new MockAgent();
    mockAgent.disableNetConnect();
    const mockPool = mockAgent.get('http://mock.localhost');

    const onConnectSpy = jest.spyOn(LoggingHandler.prototype, 'onConnect');
    const onBodySentSpy = jest.spyOn(LoggingHandler.prototype, 'onBodySent');

    const dispatcher = mockAgent.compose(createLoggingInterceptor());

    const bodyString = JSON.stringify({
      amount: '100',
    });

    mockPool
      .intercept({
        path: '/test',
        method: 'POST',
        body: () => {
          return true;
        },
      })
      .reply(200, {
        test: 1,
      });

    const res = await fetch('http://mock.localhost/test', {
      dispatcher,
      method: 'POST',
      body: bodyString,
    });
    expect(await res.json()).toEqual({ test: 1 });
    expect(onConnectSpy).toHaveBeenCalledTimes(1);
    expect(onBodySentSpy).toHaveBeenCalledTimes(1); // fails
  });
});

Expected Behavior

onBodySent would be called when making a POST request

Additional context

Looks like it should be called here: https://github.com/nodejs/undici/blob/a73fd0970f7b30cdb38b099e71ff2e354500c3b5/lib/mock/mock-utils.js#L317

onBodySent future?

Concerned about these future changes which remove onBodySent. Not sure how I'd easily get the request body in an interceptor in the next branch. Mostly creating this issue to figure out what I should be doing instead going forward. https://github.com/nodejs/undici/pull/2723#issuecomment-2484624946

[mcollina]

Thanks for reporting! Would you like to send a Pull Request to address this issue? Remember to add unit tests.

[metcoder95]

FWIW, This can be achieved in the same way with the current interceptor setup. e.g.

return (dispatch: Dispatcher['dispatch']) => {
    return function intercept(requestOptions: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandlers) {
	  // if body is stream, attach to it
      requestOptions.body.on('data', () => { /* do something */ })

      return dispatch(requestOptions, new LoggingHandler(requestOptions, handler));
    };

[tjhiggins]

FWIW, This can be achieved in the same way with the current interceptor setup. e.g.

return (dispatch: Dispatcher['dispatch']) => {
    return function intercept(requestOptions: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandlers) {
	  // if body is stream, attach to it
      requestOptions.body.on('data', () => { /* do something */ })

      return dispatch(requestOptions, new LoggingHandler(requestOptions, handler));
    };

@metcoder95 I appreciate the quick response. I don't think this works for fetch since it's requestOptions.body is an AsyncGenerator: https://github.com/nodejs/undici/blob/444a0afaa545a7009e21380e96c8b947ede56bb7/lib/web/fetch/index.js#L1845

I tried wrapping in a Readable.from, but ran into other issues with content-length being set in the headers.

[tjhiggins]

This feels pretty bad, but it seems to work:

export function createLoggingInterceptor({ projectId, name }: LoggingOptions) {
  const logging = new Logging({ projectId, autoRetry: true, maxRetries: 3 });
  const log = logging.log(name);
  return (dispatch: Dispatcher['dispatch']) => {
    return function Logging(requestOptions: Dispatcher.DispatchOptions, handler: Dispatcher.DispatchHandlers) {
      let body1: PassThrough | undefined;
      let body2: PassThrough | undefined;
      if (requestOptions.body) {
        body1 = new PassThrough();
        body2 = new PassThrough();
        const body = Readable.from(requestOptions.body); // requestOptions.body is an AsyncGenerator and can only be read once
        body.pipe(body1);
        body.pipe(body2);
      }
      return dispatch(
        { ...requestOptions, body: body1 },
        new LoggingHandler({ log }, { ...requestOptions, body: body2 }, handler),
      );
    };
  };
}

[metcoder95]

Yeah, AsyncGenerators cannot be re-used/consumed in parallel.

Can you elaborate why bad? The interceptors API has been made for these kind of purposes

[tjhiggins]

Yeah, AsyncGenerators cannot be re-used/consumed in parallel.

Can you elaborate why bad? The interceptors API has been made for these kind of purposes

The onBodySent callback is more straightforward than having to tee the body that is passed into dispatch/handler. Advocating for keeping the event since request/fetch have different request bodies.

onBodySent(chunk: any, totalBytesSent: number) {
  this.requestBody.push(chunk);
}

[metcoder95]

cc: @ronag

[mcollina]

I think we have dropped onBodySent in v7.

[tjhiggins]

@metcoder95 @mcollina could we add onBodySent back? What was the rationale for removing in the first place?

[metcoder95]

In pro of an improved lifecycle hooks API. I believe something like onRequestSent (that is triggered only after the full Request is sent) could be beneficial (I'll add onRequestError on top for situations where the transmission of the request failed due to other side ended or networking issue).

Contributions are always welcomed

[tjhiggins]

@metcoder95 both of those hooks would be great and make it a lot easier to build interceptors. I'm going to hold off on upgrading to v7 until they are added.