Pass push stream through Express middleware/route handler stack

[WVmf]

Hi,

This is probably more of a (double) question than an issue, but I'm trying to run a push stream (created with Response.push(path, opts) through an existing stack of Express middlewares and route handlers. This way I would be able to use the existing stack (which for example handles compression) instead of having to manually create a file stream and pipe it (i.e. fs.createReadStream("somePathToFile").pipe(pushStream);).

But there are two problems that I've encountered in trying to get this to work:

Firstly the Response.push() method expects both the request and response headers, and will send both of them out immediately. But this means that I would need to know which headers will be set by the middlewares and route handlers before actually passing the push stream through app._router.handle(). In other words since the http/2 headers frame was already sent by Response.push(), the headers set (and sent?) by the Express middlewares and route handlers will no longer be "seen". One way to circumvent this issue is to go through the Express stack twice, once for collecting the headers (without outputting any data), and once to actually send the data. But this is far from optimal, as for example Express's static middleware will read the file from disk twice (resulting in extra delay). So my question here is: Can I somehow tell the Response.push() method to not send the response headers, and let them be sent by the Express middlewares and route handlers.

Secondly, the app._router.handle() expects a request and response object. Constructing a fake request object was fairly straightforward. But what is the proper way to construct a ServerResponse object using the push stream object that is returned by spdy's Response.push()? Currently I try to do this as shown below by making the ServerResponse "inherit" the push stream object and overriding some methods. But this approach isn't really "clean" and only partially works (e.g., piping doesn't work yet). _Minor edit: Piping etc. did work, but I forgot to disable sending the headers in my writeHeadOverride() which caused the headers to be sent twice. The second header frame was interpreted by Chrome as a (malformed, e.g., it doesn't have the END_STREAM flag) trailing header, which caused an HTTP2_STREAMERROR (which didn't seem to propagate to my Node code?). My two questions still remain though...

let pushPath = "/path/to/file.txt";
let pushStream = res.push(pushPath, {
    status: 200, // default 200
    method: "GET",
    request: { "accept": "*/*" },
    response: { "content-type": mime.lookup(pushPath) }
});

let pushFakeReq = new http.IncomingMessage();
pushFakeReq.url = pushPath;
pushFakeReq.method = pushStream.method;
Object.assign(pushFakeReq.headers, pushStream.headers);
pushFakeReq.headers["host"] = pushStream.host;

let pushFakeRes = new http.ServerResponse({
    "method": pushStream.method,
    "httpVersionMajor": 1,
    "httpVersionMinor": 1
});

// Make the ServerResponse object "inherit" the pushStream's properties.
for ( let k in pushStream ) {
    pushFakeRes[k] = pushStream[k];
}
pushFakeRes.writeHead = writeHeadOverride;
pushFakeRes.writeContinue = writeContinueOverride;
pushFakeRes.orgEnd = pushFakeRes.end;
pushFakeRes.end = endOverride;
let pushOutCb = function (err) {
    if ( err ) {
        console.error("Error for: " + pushPath, err);
    } else {
        console.log("No error seemed to have occurred for " + pushPath);
    }
};
app._router.handle(pushFakeReq, pushFakeRes, pushOutCb);

The override functions:

Click to expand

// Based on spdy::response.js
function writeHeadOverride(statusCode, reason, obj) {
    var headers;
    if ( typeof reason === "string" ) {
        // writeHead(statusCode, reasonPhrase[, headers])
        this.statusMessage = reason;
    } else {
        // writeHead(statusCode[, headers])
        this.statusMessage = this.statusMessage || "unknown";
        obj = reason;
    }
    this.statusCode = statusCode;

    if ( this._headers ) {
        // Slow-case: when progressive API and header fields are passed.
        if ( obj ) {
            var keys = Object.keys(obj);
            for ( var i = 0; i < keys.length; i++ ) {
                var k = keys[i];
                if ( k ) this.setHeader(k, obj[k]);
            }
        }
        // only progressive api is used
        headers = this._renderHeaders();
    } else {
        // only writeHead() called
        headers = obj;
    }

    if ( statusCode === 204 || statusCode === 304 || (100 <= statusCode && statusCode <= 199) ) {
        // RFC 2616, 10.2.5:
        // The 204 response MUST NOT include a message-body, and thus is always
        // terminated by the first empty line after the header fields.
        // RFC 2616, 10.3.5:
        // The 304 response MUST NOT contain a message-body, and thus is always
        // terminated by the first empty line after the header fields.
        // RFC 2616, 10.1 Informational 1xx:
        // This class of status code indicates a provisional response,
        // consisting only of the Status-Line and optional headers, and is
        // terminated by an empty line.
        this._hasBody = false;
    }

    // don't keep alive connections where the client expects 100 Continue
    // but we sent a final status; they may put extra bytes on the wire.
    if ( this._expect_continue && !this._sent100 ) {
        this.shouldKeepAlive = false;
    }

    // Implicit headers sent!
    this._header = true;
    this._headerSent = true;

    //if ( this.socket._handle )
    //  this.socket._handle._spdyState.stream.respond(this.statusCode, headers);
    // EDIT: Headers were already sent during push stream creation, don't incorrectly send
    // them twice here.
    //this.respond(this.statusCode, headers);
}

function writeContinueOverride(callback) {
    //if ( this.socket._handle )
    //  this.socket._handle._spdyState.stream.respond(100, {}, callback);
    this.respond(100, {}, callback);
}

function endOverride(data, encoding, callback) {
    if ( !this._headerSent )
        this.writeHead(this.statusCode);

    //if ( !this.socket._handle )
    //  return;

    // Compatibility with Node.js core
    this.finished = true;

    //var self = this;
    //var handle = this.socket._handle;
    //handle._spdyState.ending = true;
    //this.socket.end(data, encoding, function() {
    //  self.constructor.prototype.end.call(self, '', 'utf8', callback);
    //});
    //this.connection.socket.end(data, encoding);
    this.orgEnd(data, encoding, callback);
}

Any feedback would be greatly appreciated.

[WVmf]

Update regarding the first question: By making a small modification to the callback function of _checkPush() in the pushFrame() method of spdy-transport/protocol/http2/framer.js, I was able to conditionally prevent the response headers from being sent immediately. The change is as follows:

  this._checkPush(function (err) {
    if (err) {
      return callback(err)
    }

    var pairs = {
      promise: [],
      response: []
    }

    self._defaultHeaders(frame, pairs.promise)
    pairs.response.push({ name: ':status', value: (frame.status || 200) + '' })

    compress(frame.headers, pairs.promise, function (promiseChunks) {
      sendPromise(promiseChunks)
+      // Don't send push response headers if 'response' was explicitly set to 'false'.
+      if (frame.response === false)
+        return callback(null)
      compress(frame.response, pairs.response, function (responseChunks) {
        sendResponse(responseChunks, callback)
      })
    })
  })

By also updating my own code as shown below, I was able to get push working (with the correct headers) without the need for an extra "header collection"-pass through the Express stack.

let pushPath = "/path/to/file.txt";
let pushStream = res.push(pushPath, {
    status: 200, // default 200
    method: "GET",
    request: { "accept": "*/*" },
-   response: { "content-type": mime.lookup(pushPath) }
+   response: false
});

let pushFakeReq = new http.IncomingMessage();
pushFakeReq.url = pushPath;
pushFakeReq.method = pushStream.method;
Object.assign(pushFakeReq.headers, pushStream.headers);
pushFakeReq.headers["host"] = pushStream.host;

let pushFakeRes = new http.ServerResponse({
    "method": pushStream.method,
    "httpVersionMajor": 1,
    "httpVersionMinor": 1
});

// Make the ServerResponse object "inherit" the pushStream's properties.
for ( let k in pushStream ) {
    pushFakeRes[k] = pushStream[k];
}
+pushFakeRes.orgWrite = pushFakeRes.write;
+pushFakeRes.write = function (chunk, encoding, cb) {
+   if ( !this._headerSent ) {
+       this.writeHead(this.statusCode);
+   }
+   this.orgWrite(chunk, encoding, cb);
+}
pushFakeRes.writeHead = writeHeadOverride;
pushFakeRes.writeContinue = writeContinueOverride;
pushFakeRes.orgEnd = pushFakeRes.end;
pushFakeRes.end = endOverride;
let pushOutCb = function (err) {
    if ( err ) {
        console.error("Error for: " + pushPath, err);
    } else {
        console.log("No error seemed to have occurred for " + pushPath);
    }
};
app._router.handle(pushFakeReq, pushFakeRes, pushOutCb);

Since the change to framer.js solves my first question, would it be possible to integrate the provided change into the official _spdytransport module? The other question (i.e. what is the proper way to construct ServerResponse from a push stream) still remains though.