import * as assert from 'assert'; import * as http2 from 'http2'; import {range} from 'lodash'; import * as stream from 'stream'; import {CallCredentials} from '../src/call-credentials'; import {Http2CallStream} from '../src/call-stream'; import {Channel, Http2Channel} from '../src/channel'; import {CompressionFilterFactory} from '../src/compression-filter'; import {Status} from '../src/constants'; import {FilterStackFactory} from '../src/filter-stack'; import {Metadata} from '../src/metadata'; import {assert2, mockFunction} from './common'; interface DataFrames { payload: Buffer; frameLengths: number[]; } const {HTTP2_HEADER_STATUS} = http2.constants; function serialize(data: string): Buffer { const header: Buffer = Buffer.alloc(5); header.writeUInt8(0, 0); // TODO: Uncompressed only header.writeInt32BE(data.length, 1); return Buffer.concat([header, Buffer.from(data, 'utf8')]); } class ClientHttp2StreamMock extends stream.Duplex implements http2.ClientHttp2Stream { constructor(private readonly dataFrames: DataFrames) { super(); } emitResponse(responseCode: number, metadata?: Metadata) { this.emit('response', { [HTTP2_HEADER_STATUS]: responseCode, ...metadata ? metadata.toHttp2Headers() : {} }); } bytesRead = 0; dataFrame = 0; aborted = false; closed = false; destroyed = false; endAfterHeaders = false; pending = false; rstCode = 0; // tslint:disable:no-any session: http2.Http2Session = {} as any; state: http2.StreamState = {} as any; // tslint:enable:no-any close = mockFunction; priority = mockFunction; rstStream = mockFunction; rstWithNoError = mockFunction; rstWithProtocolError = mockFunction; rstWithCancel = mockFunction; rstWithRefuse = mockFunction; rstWithInternalError = mockFunction; setTimeout = mockFunction; _read() { if (this.dataFrame === this.dataFrames.frameLengths.length) { if (this.bytesRead < this.dataFrames.payload.length) { this.push(this.dataFrames.payload.slice( this.bytesRead, this.dataFrames.payload.length)); } this.push(null); return; } const from = this.bytesRead; this.bytesRead += this.dataFrames.frameLengths[this.dataFrame++]; this.push(this.dataFrames.payload.slice(from, this.bytesRead)); } _write(chunk: Buffer, encoding: string, cb: Function) { this.emit('write', chunk); cb(); } } describe('CallStream', () => { const callStreamArgs = {deadline: Infinity, flags: 0, host: '', parentCall: null}; /* A CompressionFilter is now necessary to frame and deframe messages. * Currently the channel is unused, so we can replace it with an empty object, * but this might break if we start checking channel arguments, in which case * we will need a more sophisticated fake */ const filterStackFactory = new FilterStackFactory([new CompressionFilterFactory({} as Channel)]); const message = 'eat this message'; // 16 bytes beforeEach(() => { assert2.clearMustCalls(); }); it('should emit a metadata event when it receives a response event', (done) => { const responseMetadata = new Metadata(); responseMetadata.add('key', 'value'); const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock( {payload: Buffer.alloc(0), frameLengths: []}); callStream.once('metadata', assert2.mustCall((metadata) => { assert.deepStrictEqual(metadata.get('key'), ['value']); })); callStream.attachHttp2Stream(http2Stream); http2Stream.emitResponse(200, responseMetadata); assert2.afterMustCallsSatisfied(done); }); describe('should end a call with an error if a stream was closed', () => { const c = http2.constants; const s = Status; const errorCodeMapping = { [c.NGHTTP2_NO_ERROR]: s.INTERNAL, [c.NGHTTP2_PROTOCOL_ERROR]: s.INTERNAL, [c.NGHTTP2_INTERNAL_ERROR]: s.INTERNAL, [c.NGHTTP2_FLOW_CONTROL_ERROR]: s.INTERNAL, [c.NGHTTP2_SETTINGS_TIMEOUT]: s.INTERNAL, [c.NGHTTP2_STREAM_CLOSED]: null, [c.NGHTTP2_FRAME_SIZE_ERROR]: s.INTERNAL, [c.NGHTTP2_REFUSED_STREAM]: s.UNAVAILABLE, [c.NGHTTP2_CANCEL]: s.CANCELLED, [c.NGHTTP2_COMPRESSION_ERROR]: s.INTERNAL, [c.NGHTTP2_CONNECT_ERROR]: s.INTERNAL, [c.NGHTTP2_ENHANCE_YOUR_CALM]: s.RESOURCE_EXHAUSTED, [c.NGHTTP2_INADEQUATE_SECURITY]: s.PERMISSION_DENIED }; const keys = Object.keys(errorCodeMapping).map(key => Number(key)); keys.forEach((key) => { const value = errorCodeMapping[key]; // A null value indicates: behavior isn't specified, so skip this test. const maybeSkip = (fn: typeof it) => value ? fn : fn.skip; maybeSkip(it)(`for error code ${key}`, () => { return new Promise((resolve, reject) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock( {payload: Buffer.alloc(0), frameLengths: []}); callStream.attachHttp2Stream(http2Stream); callStream.once('status', (status) => { try { assert.strictEqual(status.code, value); resolve(); } catch (e) { reject(e); } }); http2Stream.emit('close', Number(key)); }); }); }); }); it('should have functioning getters', (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); assert.strictEqual(callStream.getDeadline(), callStreamArgs.deadline); assert.strictEqual(callStream.getStatus(), null); const credentials = CallCredentials.createEmpty(); callStream.setCredentials(credentials); assert.strictEqual(callStream.getCredentials(), credentials); callStream.on('status', assert2.mustCall((status) => { assert.strictEqual(status.code, Status.CANCELLED); assert.strictEqual(status.details, ';)'); assert.strictEqual(callStream.getStatus(), status); })); callStream.cancelWithStatus(Status.CANCELLED, ';)'); // TODO: getPeer assert2.afterMustCallsSatisfied(done); }); describe('attachHttp2Stream', () => { it('should handle an empty message', (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock({payload: serialize(''), frameLengths: []}); callStream.once('data', assert2.mustCall((buffer) => { assert.strictEqual(buffer.toString('utf8'), ''); })); callStream.attachHttp2Stream(http2Stream); assert2.afterMustCallsSatisfied(done); }); [{description: 'all data is supplied in a single frame', frameLengths: []}, { description: 'frames are split along header field delimiters', frameLengths: [1, 4] }, { description: 'portions of header fields are split between different frames', frameLengths: [2, 1, 1, 4] }, { description: 'frames are split into bytes', frameLengths: range(0, 20).map(() => 1) }].forEach((testCase: {description: string, frameLengths: number[]}) => { it(`should handle a short message where ${testCase.description}`, (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock({ payload: serialize(message), // 21 bytes frameLengths: testCase.frameLengths }); callStream.once('data', assert2.mustCall((buffer) => { assert.strictEqual(buffer.toString('utf8'), message); })); callStream.once('end', assert2.mustCall(() => {})); callStream.attachHttp2Stream(http2Stream); assert2.afterMustCallsSatisfied(done); }); }); [{description: 'all data is supplied in a single frame', frameLengths: []}, { description: 'frames are split between delimited messages', frameLengths: [21] }, {description: 'frames are split within messages', frameLengths: [10, 22]}, { description: 'part of 2nd message\'s header is in first frame', frameLengths: [24] }, { description: 'frames are split into bytes', frameLengths: range(0, 41).map(() => 1) }].forEach((testCase: {description: string, frameLengths: number[]}) => { it(`should handle two messages where ${testCase.description}`, (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock({ payload: Buffer.concat( [serialize(message), serialize(message)]), // 42 bytes frameLengths: testCase.frameLengths }); callStream.once('data', assert2.mustCall((buffer) => { assert.strictEqual(buffer.toString('utf8'), message); })); callStream.once('data', assert2.mustCall((buffer) => { assert.strictEqual(buffer.toString('utf8'), message); })); callStream.once('end', assert2.mustCall(() => {})); callStream.attachHttp2Stream(http2Stream); assert2.afterMustCallsSatisfied(done); }); }); it('should send buffered writes', (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock( {payload: Buffer.alloc(0), frameLengths: []}); let streamFlushed = false; http2Stream.once('write', assert2.mustCall((chunk: Buffer) => { const dataLength = chunk.readInt32BE(1); const encodedMessage = chunk.slice(5).toString('utf8'); assert.strictEqual(dataLength, message.length); assert.strictEqual(encodedMessage, message); streamFlushed = true; })); callStream.write({message: Buffer.from(message)}, assert2.mustCall(() => { // Ensure this is called only after contents are written to http2Stream assert.ok(streamFlushed); })); callStream.end(assert2.mustCall(() => {})); callStream.attachHttp2Stream(http2Stream); assert2.afterMustCallsSatisfied(done); }); it('should cause data chunks in write calls afterward to be written to the given stream', (done) => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock( {payload: Buffer.alloc(0), frameLengths: []}); http2Stream.once('write', assert2.mustCall((chunk: Buffer) => { const dataLength = chunk.readInt32BE(1); const encodedMessage = chunk.slice(5).toString('utf8'); assert.strictEqual(dataLength, message.length); assert.strictEqual(encodedMessage, message); })); callStream.attachHttp2Stream(http2Stream); callStream.write( {message: Buffer.from(message)}, assert2.mustCall(() => {})); callStream.end(assert2.mustCall(() => {})); assert2.afterMustCallsSatisfied(done); }); it('should handle underlying stream errors', () => { const callStream = new Http2CallStream( 'foo', {} as Http2Channel, callStreamArgs, filterStackFactory); const http2Stream = new ClientHttp2StreamMock( {payload: Buffer.alloc(0), frameLengths: []}); callStream.once('status', assert2.mustCall((status) => { assert.strictEqual(status.code, Status.INTERNAL); })); callStream.attachHttp2Stream(http2Stream); http2Stream.emit('error'); }); }); });