Add a 'response-initiated' event

Author: pimterry

src/admin/mockttp-admin-model.ts

@@ -31,6 +31,7 @@ import { SubscribableEvent } from "../main";
 const graphqlSubscriptionPairs = Object.entries({
     'requestInitiated': 'request-initiated',
     'requestReceived': 'request',
+    'responseInitiated': 'response-initiated',
     'responseCompleted': 'response',
     'webSocketRequest': 'websocket-request',
     'webSocketAccepted': 'websocket-accepted',

src/admin/mockttp-schema.ts

@@ -20,6 +20,7 @@ export const MockttpSchema = gql`
     extend type Subscription {
         requestInitiated: InitiatedRequest!
         requestReceived: Request!
+        responseInitiated: InitiatedResponse!
         responseCompleted: Response!
         webSocketRequest: Request!
         webSocketAccepted: Response!
@@ -206,6 +207,18 @@ export const MockttpSchema = gql`
         error: Json
     }
 
+    type InitiatedResponse {
+        id: ID!
+        timingEvents: Json!
+        tags: [String!]!
+
+        statusCode: Int!
+        statusMessage: String!
+
+        headers: Json!
+        rawHeaders: Json!
+    }
+
     type Response {
         id: ID!
         timingEvents: Json!

src/client/mockttp-admin-request-builder.ts

@@ -244,6 +244,16 @@ export class MockttpAdminRequestBuilder {
                     tags
                 }
             }`,
+            'response-initiated': gql`subscription OnResponseInitiated {
+                responseInitiated {
+                    id
+                    statusCode
+                    statusMessage
+                    rawHeaders
+                    timingEvents
+                    tags
+                }
+            }`,
             response: gql`subscription OnResponse {
                 responseCompleted {
                     id

src/mockttp.ts

@@ -22,7 +22,8 @@ import {
     AbortedRequest,
     RuleEvent,
     RawPassthroughEvent,
-    RawPassthroughDataEvent
+    RawPassthroughDataEvent,
+    InitiatedResponse
 } from "./types";
 import type { RequestRuleData } from "./rules/requests/request-rule";
 import type { WebSocketRuleData } from "./rules/websockets/websocket-rule";
@@ -364,6 +365,21 @@ export interface Mockttp {
      */
     on(event: 'request', callback: (req: CompletedRequest) => void): Promise<void>;
 
+    /**
+     * Subscribe to hear about response details as soon as the initial response (the
+     * status code & headers) are sent, without waiting for the body.
+     *
+     * This is only useful in some niche use cases, such as logging all requests seen
+     * by the server independently of the rules defined.
+     *
+     * The callback will be called asynchronously from request handling. This function
+     * returns a promise, and the callback is not guaranteed to be registered until
+     * the promise is resolved.
+     *
+     * @category Events
+     */
+    on(event: 'response-initiated', callback: (req: InitiatedResponse) => void): Promise<void>;
+
     /**
      * Subscribe to hear about response details when the response is completed.
      *
@@ -885,6 +901,7 @@ export interface MockttpOptions {
 export type SubscribableEvent =
     | 'request-initiated'
     | 'request'
+    | 'response-initiated'
     | 'response'
     | 'websocket-request'
     | 'websocket-accepted'

src/server/mockttp-server.ts

@@ -32,7 +32,8 @@ import {
     RawTrailers,
     RawPassthroughEvent,
     RawPassthroughDataEvent,
-    RawHeaders
+    RawHeaders,
+    InitiatedResponse
 } from "../types";
 import { DestroyableServer } from "destroyable-server";
 import {
@@ -82,7 +83,8 @@ import {
     buildInitiatedRequest,
     tryToParseHttpRequest,
     buildBodyReader,
-    parseRawHttpResponse
+    parseRawHttpResponse,
+    buildInitiatedResponse
 } from "../util/request-utils";
 import { asBuffer } from "../util/buffer-utils";
 import {
@@ -327,6 +329,7 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
 
     public on(event: 'request-initiated', callback: (req: InitiatedRequest) => void): Promise<void>;
     public on(event: 'request', callback: (req: CompletedRequest) => void): Promise<void>;
+    public on(event: 'response-initiated', callback: (req: InitiatedResponse) => void): Promise<void>;
     public on(event: 'response', callback: (req: CompletedResponse) => void): Promise<void>;
     public on(event: 'abort', callback: (req: InitiatedRequest) => void): Promise<void>;
     public on(event: 'websocket-request', callback: (req: CompletedRequest) => void): Promise<void>;
@@ -380,6 +383,21 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
         .catch(console.error);
     }
 
+    private announceInitialResponseAsync(response: OngoingResponse) {
+        if (this.eventEmitter.listenerCount('response-initiated') === 0) return;
+
+        setImmediate(() => {
+            const initiatedRes = buildInitiatedResponse(response);
+            this.eventEmitter.emit('response-initiated', Object.assign(
+                initiatedRes,
+                {
+                    timingEvents: _.clone(initiatedRes.timingEvents),
+                    tags: _.clone(initiatedRes.tags)
+                }
+            ));
+        });
+    }
+
     private announceResponseAsync(response: OngoingResponse | CompletedResponse) {
         if (this.eventEmitter.listenerCount('response') === 0) return;
 
@@ -790,7 +808,10 @@ export class MockttpServer extends AbstractMockttp implements Mockttp {
             rawResponse,
             request.timingEvents,
             request.tags,
-            { maxSize: this.maxBodySize }
+            {
+                maxSize: this.maxBodySize,
+                onWriteHead: () => this.announceInitialResponseAsync(response)
+            }
         );
         response.id = request.id;
         response.on('error', (error) => {

src/types.ts

@@ -321,17 +321,20 @@ export interface OngoingResponse extends http.ServerResponse {
     tags: string[];
 }
 
-export interface CompletedResponse {
+export interface InitiatedResponse {
     id: string;
     statusCode: number;
     statusMessage: string;
     headers: Headers;
     rawHeaders: RawHeaders;
+    timingEvents: TimingEvents;
+    tags: string[];
+}
+
+export interface CompletedResponse extends InitiatedResponse {
     body: CompletedBody;
     rawTrailers: RawTrailers;
     trailers: Trailers;
-    timingEvents: TimingEvents;
-    tags: string[];
 }
 
 export interface WebSocketMessage {

src/util/request-utils.ts

@@ -24,7 +24,8 @@ import {
     TimingEvents,
     InitiatedRequest,
     RawHeaders,
-    Destination
+    Destination,
+    InitiatedResponse
 } from "../types";
 
 import {
@@ -354,7 +355,10 @@ export function trackResponse(
     response: http.ServerResponse,
     timingEvents: TimingEvents,
     tags: string[],
-    options: { maxSize: number }
+    options: {
+        maxSize: number,
+        onWriteHead: () => void
+    }
 ): OngoingResponse {
     let trackedResponse = <OngoingResponse> response;
 
@@ -378,6 +382,8 @@ export function trackResponse(
     trackedResponse.writeHead = function (this: typeof trackedResponse, ...args: any) {
         if (!timingEvents.headersSentTimestamp) {
             timingEvents.headersSentTimestamp = now();
+            // Notify listeners that the head is being written:
+            options.onWriteHead();
         }
 
         // HTTP/2 responses shouldn't have a status message:
@@ -466,41 +472,47 @@ export function trackResponse(
 }
 
 /**
- * Build a completed response: the external representation of a response
- * that's been completely written out and sent back to the client.
+ * Build an initiated response: the external representation of a response
+ * that's just started.
  */
-export async function waitForCompletedResponse(
-    response: OngoingResponse | CompletedResponse
-): Promise<CompletedResponse> {
-    // Ongoing response has 'getHeaders' - completed has 'headers'.
-    if ('headers' in response) return response;
-
-    const body = await waitForBody(response.body, response.getHeaders());
-    response.timingEvents.responseSentTimestamp = response.timingEvents.responseSentTimestamp || now();
-
-    const completedResponse: CompletedResponse = _(response).pick([
+export function buildInitiatedResponse(response: OngoingResponse): InitiatedResponse {
+    const initiatedResponse: InitiatedResponse = _(response).pick([
         'id',
         'statusCode',
         'timingEvents',
         'tags'
     ]).assign({
         statusMessage: '',
-
         headers: response.getHeaders(),
         rawHeaders: response.getRawHeaders(),
-
-        body: body,
-
-        rawTrailers: response.getRawTrailers(),
-        trailers: rawHeadersToObject(response.getRawTrailers())
     }).valueOf();
 
     if (!(response instanceof http2.Http2ServerResponse)) {
         // H2 has no status messages, and generates a warning if you look for one
-        completedResponse.statusMessage = response.statusMessage;
+        initiatedResponse.statusMessage = response.statusMessage;
     }
 
-    return completedResponse;
+    return initiatedResponse;
+}
+
+/**
+ * Build a completed response: the external representation of a response
+ * that's been completely written out and sent back to the client.
+ */
+export async function waitForCompletedResponse(
+    response: OngoingResponse | CompletedResponse
+): Promise<CompletedResponse> {
+    // Ongoing response has 'getHeaders' - completed has 'headers'.
+    if ('headers' in response) return response;
+
+    const body = await waitForBody(response.body, response.getHeaders());
+    response.timingEvents.responseSentTimestamp = response.timingEvents.responseSentTimestamp || now();
+
+    return Object.assign(buildInitiatedResponse(response), {
+        body: body,
+        rawTrailers: response.getRawTrailers(),
+        trailers: rawHeadersToObject(response.getRawTrailers())
+    });
 }
 
 // Take raw HTTP request bytes received, have a go at parsing something useful out of them.

test/integration/subscriptions/response-events.spec.ts

@@ -1,4 +1,5 @@
 import * as _ from 'lodash';
+import { PassThrough } from 'stream';
 import * as http from 'http';
 import * as zlib from 'zlib';
 
@@ -7,7 +8,8 @@ import {
     CompletedRequest,
     CompletedResponse,
     TimingEvents,
-    AbortedRequest
+    AbortedRequest,
+    InitiatedResponse
 } from "../../..";
 import {
     expect,
@@ -19,6 +21,55 @@ import {
     makeAbortableRequest
 } from "../../test-utils";
 
+describe("Response initiated subscriptions", () => {
+    describe("with a local HTTP server", () => {
+        let server = getLocal();
+
+        beforeEach(() => server.start());
+        afterEach(() => server.stop());
+
+        it("should notify with response details as soon as they're ready", async () => {
+            let seenResponsePromise = getDeferred<InitiatedResponse>();
+            await server.on('response-initiated', (r) => seenResponsePromise.resolve(r));
+
+            const bodyStream = new PassThrough();
+            await server.forAnyRequest().thenStream(400, bodyStream, {
+                'a': 'b',
+                'access-control-allow-origin': '*',
+                'access-control-expose-headers': '*'
+            });
+
+            const realResponse = await fetch(server.urlFor("/mocked-endpoint"));
+            const realResponseHeaders = Object.fromEntries(realResponse.headers as any);
+
+            let seenResponse = await seenResponsePromise;
+            expect(seenResponse.statusCode).to.equal(400);
+            expect(seenResponse.statusMessage).to.equal('Bad Request');
+
+            expect(seenResponse.headers).to.deep.equal(realResponseHeaders);
+            expect(seenResponse.rawHeaders).to.deep.equal(Object.entries(realResponseHeaders));
+
+            expect((seenResponse as any).body).to.equal(undefined); // No body included yet
+            expect((seenResponse as any).trailers).to.equal(undefined); // No trailers yet
+            expect((seenResponse as any).rawTrailers).to.equal(undefined);
+
+            expect(seenResponse.id).to.be.a('string');
+            expect(seenResponse.tags).to.deep.equal([]);
+
+            const timingEvents = seenResponse.timingEvents;
+            expect(timingEvents.startTimestamp).to.be.a('number');
+            expect(timingEvents.bodyReceivedTimestamp).to.be.a('number');
+            expect(timingEvents.headersSentTimestamp).to.be.a('number');
+
+            expect(timingEvents.bodyReceivedTimestamp).to.be.greaterThan(timingEvents.startTimestamp);
+            expect(timingEvents.headersSentTimestamp).to.be.greaterThan(timingEvents.startTimestamp);
+
+            expect(timingEvents.responseSentTimestamp).to.equal(undefined);
+            expect(timingEvents.abortedTimestamp).to.equal(undefined);
+        });
+    });
+});
+
 describe("Response subscriptions", () => {
 
     describe("with an HTTP server", () => {