New RESP2 parser (redis#1899)

* parser

* a new RESP parser :)

* clean code

* fix simple string and bulk string cursor

* performance improvements

* change typescript compiler target

* do not use stream.Transform

* Update decoder.ts

* fix for 1d09acb

* improve integer performance

* revert 1d09acb

* improve RESP2 decoder performance

* improve performance

* improve encode performance

* remove unused import

* upgrade benchmark deps

* clean code

* fix socket error handlers, reset parser on error

* fix redis#2080 - reset pubSubState on socket error

* reset decoder on socket error

* fix pubsub

* fix "RedisSocketInitiator"

* fix returnStringsAsBuffers

* fix merge

Author: leibale

benchmark/package-lock.json

@@ -0,0 +1,14 @@
+import { strict as assert } from 'assert';
+import BufferComposer from './buffer';
+
+describe('Buffer Composer', () => {
+    const composer = new BufferComposer();
+
+    it('should compose two buffers', () => {
+        composer.write(Buffer.from([0]));
+        assert.deepEqual(
+            composer.end(Buffer.from([1])),
+            Buffer.from([0, 1])
+        );
+    });
+});

packages/client/lib/client/RESP2/composers/buffer.spec.ts

@@ -0,0 +1,18 @@
+import { Composer } from './interface';
+
+export default class BufferComposer implements Composer<Buffer> {
+    private chunks: Array<Buffer> = [];
+
+    write(buffer: Buffer): void {
+        this.chunks.push(buffer);
+    }
+
+    end(buffer: Buffer): Buffer {
+        this.write(buffer);
+        return Buffer.concat(this.chunks.splice(0));
+    }
+
+    reset() {
+        this.chunks = [];
+    }
+}

packages/client/lib/client/RESP2/composers/buffer.ts

@@ -0,0 +1,7 @@
+export interface Composer<T> {
+    write(buffer: Buffer): void;
+
+    end(buffer: Buffer): T;
+
+    reset(): void;
+}

packages/client/lib/client/RESP2/composers/interface.ts

@@ -0,0 +1,14 @@
+import { strict as assert } from 'assert';
+import StringComposer from './string';
+
+describe('String Composer', () => {
+    const composer = new StringComposer();
+
+    it('should compose two strings', () => {
+        composer.write(Buffer.from([0]));
+        assert.deepEqual(
+            composer.end(Buffer.from([1])),
+            Buffer.from([0, 1]).toString()
+        );
+    });
+});

packages/client/lib/client/RESP2/composers/string.spec.ts

@@ -0,0 +1,22 @@
+import { StringDecoder } from 'string_decoder';
+import { Composer } from './interface';
+
+export default class StringComposer implements Composer<string> {
+    private decoder = new StringDecoder();
+
+    private string = '';
+
+    write(buffer: Buffer): void {
+        this.string += this.decoder.write(buffer);
+    }
+
+    end(buffer: Buffer): string {
+        const string = this.string + this.decoder.end(buffer);
+        this.string = '';
+        return string;
+    }
+
+    reset() {
+        this.string = '';
+    }
+}

packages/client/lib/client/RESP2/composers/string.ts

@@ -0,0 +1,195 @@
+import { strict as assert } from 'assert';
+import { SinonSpy, spy } from 'sinon';
+import RESP2Decoder from './decoder';
+import { ErrorReply } from '../../errors';
+
+interface DecoderAndSpies {
+    decoder: RESP2Decoder;
+    returnStringsAsBuffersSpy: SinonSpy;
+    onReplySpy: SinonSpy;
+}
+
+function createDecoderAndSpies(returnStringsAsBuffers: boolean): DecoderAndSpies {
+    const returnStringsAsBuffersSpy = spy(() => returnStringsAsBuffers),
+        onReplySpy = spy();
+
+    return {
+        decoder: new RESP2Decoder({
+            returnStringsAsBuffers: returnStringsAsBuffersSpy,
+            onReply: onReplySpy
+        }),
+        returnStringsAsBuffersSpy,
+        onReplySpy
+    };
+}
+
+function writeChunks(stream: RESP2Decoder, buffer: Buffer) {
+    let i = 0;
+    while (i < buffer.length) {
+        stream.write(buffer.slice(i, ++i));
+    }
+}
+
+type Replies = Array<Array<unknown>>;
+
+interface TestsOptions {
+    toWrite: Buffer;
+    returnStringsAsBuffers: boolean;
+    replies: Replies;
+}
+
+function generateTests({
+    toWrite,
+    returnStringsAsBuffers,
+    replies
+}: TestsOptions): void {
+    it('single chunk', () => {
+        const { decoder, returnStringsAsBuffersSpy, onReplySpy } =
+            createDecoderAndSpies(returnStringsAsBuffers);
+        decoder.write(toWrite);
+        assert.equal(returnStringsAsBuffersSpy.callCount, replies.length);
+        testReplies(onReplySpy, replies);
+    });
+
+    it('multiple chunks', () => {
+        const { decoder, returnStringsAsBuffersSpy, onReplySpy } =
+            createDecoderAndSpies(returnStringsAsBuffers);
+        writeChunks(decoder, toWrite);
+        assert.equal(returnStringsAsBuffersSpy.callCount, replies.length);
+        testReplies(onReplySpy, replies);
+    });
+}
+
+function testReplies(spy: SinonSpy, replies: Replies): void {
+    if (!replies) {
+        assert.equal(spy.callCount, 0);
+        return;
+    }
+
+    assert.equal(spy.callCount, replies.length);
+    for (const [i, reply] of replies.entries()) {
+        assert.deepEqual(
+            spy.getCall(i).args,
+            reply
+        );
+    }
+}
+
+describe('RESP2Parser', () => {
+    describe('Simple String', () => {
+        describe('as strings', () => {
+            generateTests({
+                toWrite: Buffer.from('+OK\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [['OK']]
+            });
+        });
+
+        describe('as buffers', () => {
+            generateTests({
+                toWrite: Buffer.from('+OK\r\n'),
+                returnStringsAsBuffers: true,
+                replies: [[Buffer.from('OK')]]
+            });
+        });
+    });
+
+    describe('Error', () => {
+        generateTests({
+            toWrite: Buffer.from('-ERR\r\n'),
+            returnStringsAsBuffers: false,
+            replies: [[new ErrorReply('ERR')]]
+        });
+    });
+
+    describe('Integer', () => {
+        describe('-1', () => {
+            generateTests({
+                toWrite: Buffer.from(':-1\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [[-1]]
+            });
+        });
+
+        describe('0', () => {
+            generateTests({
+                toWrite: Buffer.from(':0\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [[0]]
+            });
+        });
+    });
+
+    describe('Bulk String', () => {
+        describe('null', () => {
+            generateTests({
+                toWrite: Buffer.from('$-1\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [[null]]
+            });
+        });
+
+        describe('as strings', () => {
+            generateTests({
+                toWrite: Buffer.from('$2\r\naa\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [['aa']]
+            });
+        });
+
+        describe('as buffers', () => {
+            generateTests({
+                toWrite: Buffer.from('$2\r\naa\r\n'),
+                returnStringsAsBuffers: true,
+                replies: [[Buffer.from('aa')]]
+            });
+        });
+    });
+
+    describe('Array', () => {
+        describe('null', () => {
+            generateTests({
+                toWrite: Buffer.from('*-1\r\n'),
+                returnStringsAsBuffers: false,
+                replies: [[null]]
+            });
+        });
+
+        const arrayBuffer = Buffer.from(
+            '*5\r\n' +
+            '+OK\r\n' +
+            '-ERR\r\n' +
+            ':0\r\n' +
+            '$1\r\na\r\n' +
+            '*0\r\n'
+        );
+
+        describe('as strings', () => {
+            generateTests({
+                toWrite: arrayBuffer,
+                returnStringsAsBuffers: false,
+                replies: [[[
+                    'OK',
+                    new ErrorReply('ERR'),
+                    0,
+                    'a',
+                    []
+                ]]]
+            });
+        });
+
+        describe('as buffers', () => {
+            generateTests({
+                toWrite: arrayBuffer,
+                returnStringsAsBuffers: true,
+                replies: [[[
+                    Buffer.from('OK'),
+                    new ErrorReply('ERR'),
+                    0,
+                    Buffer.from('a'),
+                    []
+                ]]]
+            });
+        });
+    });
+});