index.js

'use strict';
const {PassThrough} = require('stream');
const ManyKeysMap = require('many-keys-map');
const {extend: gotExtend} = require('got');
const http2 = require('../../source/index.js'); // Note: using the local version

class PushAgent extends http2.Agent {
	constructor(options) {
		super(options);
		this.settings.enablePush = true;

		this.on('session', session => {
			const pushCache = new ManyKeysMap();
			session.pushCache = pushCache;

			session.on('stream', (stream, requestHeaders) => {
				const proxy = new PassThrough({highWaterMark: 1024 * 1024}); // 1MB

				proxy.session = {
					socket: stream.session.socket
				};

				// Hacky override to avoid implementing a Duplex stream
				const end = proxy.end.bind(proxy);
				proxy.end = (...args) => {
					end(...args);

					// Node 13 throws when double-ending
					proxy.end = () => {};
				};

				stream.pipe(proxy);

				const parsedPushHeaders = PushAgent._parsePushHeaders(undefined, requestHeaders);

				if (pushCache.has(parsedPushHeaders)) {
					stream.close(http2.constants.NGHTTP2_REFUSED_STREAM);
					return;
				}

				stream.once('push', pushHeaders => {
					pushCache.set(parsedPushHeaders, {stream: proxy, pushHeaders});
				});
			});
		});
	}

	request(origin, sessionOptions, headers, streamOptions) {
		return new Promise((resolve, reject) => {
			// The code after `await agent.getSession()` isn't executed immediately after calling `resolve()`,
			// so we need to use semi-callback style to support the `maxFreeSessions` option mechanism.

			// For further information please see the source code of the `processListeners` function (`source/agent.js` file).
			// A solution to avoid this hack will be greatly appreciated!

			this.getSession(origin, sessionOptions, [{
				reject,
				resolve: session => {
					const normalizedAuthority = (new URL(origin)).origin;

					const parsedPushHeaders = PushAgent._parsePushHeaders(normalizedAuthority, headers);
					const cache = session.pushCache.get(parsedPushHeaders);
					if (cache) {
						const {stream, pushHeaders} = cache;
						session.pushCache.delete(parsedPushHeaders);

						setImmediate(() => {
							stream.emit('response', pushHeaders);
						});

						resolve(stream);
						return;
					}

					resolve(session.request(headers, streamOptions));
				}
			}]);
		});
	}

	static _parsePushHeaders(authority, headers) {
		return [
			headers[':authority'] || authority,
			headers[':path'] || '/',
			headers[':method'] || 'GET'
		];
	}
}

(async () => {
	const agent = new PushAgent();

	const got = gotExtend({
		prefixUrl: 'https://localhost:3000',
		request: http2.auto,
		http2: true,
		rejectUnauthorized: false,
		agent: {
			http2: agent
		}
	});

	const response = await got('');
	console.log('/', response.body, response.headers);

	const pushResponse = await got('push');
	console.log('/push', pushResponse.body, pushResponse.headers);
})();