Implement both multi and pipeline.

fenying · fenying · commit 28cde56c989e · 2019-06-29T16:32:19.000+08:00
diff --git a/commands.json b/commands.json
diff --git a/docs/zh-CN/Pipeline.md b/docs/zh-CN/Pipeline.md
@@ -0,0 +1,69 @@
+# Pipeline 管道原理
+
+LiteRT/Redis 的客户端实现中，Pipeline 有两种使用方式：**事务**和**管线**。
+
+## 1. 事务
+
+事务是指通过 `WATCH` 和 `MULTI` 指令配合实现的原子操作，其实现完全基于 Redis 服务器。
+
+```
+Application               Client                      Server
+     |                       |                           |
+     |  Send MULIT command   |    Send MULIT command     |
+     |---------------------->|-------------------------->|
+     |                       |                           |
+     |       Reply OK        |         Reply OK          |
+     |<----------------------|<--------------------------|
+     |                       |                           |
+     |   Send SET command    |     Send SET command      |
+     |---------------------->|-------------------------->|
+     |                       |                           |
+     |      Reply QUEUED     |         Reply QUEUED      |
+     |<----------------------|<--------------------------|
+     |                       |                           |
+     |   Send GET command    |     Send GET command      |
+     |---------------------->|-------------------------->|
+     |                       |                           |
+     |      Reply QUEUED     |         Reply QUEUED      |
+     |<----------------------|<--------------------------|
+     |                       |                           |
+     |   Send EXEC command   |     Send EXEC command     |
+     |---------------------->|-------------------------->|  Server execute
+     |                       |                           |  all queued commands。
+     |                       |                           |
+     | Reply list of result  |   Reply list of result    |
+     |<----------------------|<--------------------------|
+     |                       |                           |
+   Over                    over                         over
+```
+
+## 2. 管道
+
+管道是指将一系列命令缓存在本地，然后一次性发出去，由于 Redis 的单线程模型，这些命令会连续不间断地（不会被其他连接发送的命令插队）执行。
+
+```
+Application               Client                      Server
+     |                       |                           |
+     |   Send SET command    |                           |
+     |---------------------->|                           |
+     |                       |                           |
+     |      Reply QUEUED     |                           |
+     |<----------------------|                           |
+     |                       |                           |
+     |   Send GET command    |                           |
+     |---------------------->|                           |
+     |                       |                           |
+     |      Reply QUEUED     |                           |
+     |<----------------------|                           |
+     |                       |                           |
+     |   Send EXEC command   | Send all queued command   |
+     |---------------------->|-------------------------->|  Server execute
+     |                       |                           |  all the series of
+     |                       |                           |  commands
+     |                       |                           |
+     | Reply list of result  |   Reply list of result    |
+     |<----------------------|<--------------------------|
+     |                       |                           |
+   Over                    over                         over
+```
+
diff --git a/src/examples/debug.ts b/src/examples/debug.ts
@@ -21,6 +21,7 @@ import * as L from "@litert/core";
 
     await cli.connect();
     await cli.auth("hello");
+
     console.log(await cli.ping(""));
     console.log(await cli.set("a", "123"));
     console.log(await cli.incr("a", 23));
@@ -49,6 +50,53 @@ import * as L from "@litert/core";
 
     await L.Async.sleep(2000);
 
+    const pipeline = await cli.pipeline();
+
+    // Multi Mode
+    await pipeline.multi();
+    await pipeline.get("a");
+
+    await pipeline.set("ccc", "g");
+
+    await pipeline.mGet(["a", "ccc"]);
+
+    await pipeline.hSet("h", "name", "Mick");
+    await pipeline.hMSet("h", {
+        "age": 123,
+        "title": "Mr."
+    });
+
+    await pipeline.hMGet("h", ["age", "title"]);
+    await pipeline.hGetAll("h");
+    console.log(JSON.stringify(await pipeline.scan(0), null, 2));
+
+    await pipeline.incr("a", 123);
+
+    console.log(JSON.stringify(await pipeline.exec(), null, 2));
+
+    // Pipeline Mode
+    await pipeline.get("a");
+
+    await pipeline.set("ccc", "g");
+
+    await pipeline.mGet(["a", "ccc"]);
+
+    await pipeline.hSet("h", "name", "Mick");
+    await pipeline.hMSet("h", {
+        "age": 123,
+        "title": "Mr."
+    });
+
+    await pipeline.hMGet("h", ["age", "title"]);
+    await pipeline.hGetAll("h");
+    console.log(JSON.stringify(await pipeline.scan(0), null, 2));
+
+    await pipeline.incr("a", 123);
+
+    console.log(JSON.stringify(await pipeline.exec(), null, 2));
+
+    await pipeline.shutdown();
+
     await cli.shutdown();
     await sub.shutdown();
 
diff --git a/src/lib/BaseClient.ts b/src/lib/BaseClient.ts
@@ -9,12 +9,13 @@ extends ProtocolClient {
     public constructor(
         host: string,
         port: number,
-        decoder: C.IDecoder,
-        encoder: C.IEncoder,
-        subscribeMode?: boolean
+        decoder: C.TDecoderFactory,
+        encoder: C.TEncoderFactory,
+        subscribeMode?: boolean,
+        pipeline?: boolean
     ) {
 
-        super(host, port, decoder, encoder, subscribeMode);
+        super(host, port, decoder, encoder, subscribeMode, pipeline);
     }
 
     protected _onConnected(callback: C.ICallbackA): void {
diff --git a/src/lib/CommandClient.ts b/src/lib/CommandClient.ts
@@ -11,11 +11,11 @@ implements C.ICommandClient {
     public constructor(
         host: string,
         port: number,
-        _decoder: C.IDecoder,
-        _encoder: C.IEncoder
+        private _createDecoder: C.TDecoderFactory,
+        private _createEncoder: C.TEncoderFactory
     ) {
 
-        super(host, port, _decoder, _encoder);
+        super(host, port, _createDecoder, _createEncoder);
     }
 
     public strLen(k: string): Promise<number> {
@@ -822,7 +822,14 @@ implements C.ICommandClient {
 
     public async pipeline(): Promise<C.IPipelineClient> {
 
-        const cli = new PipelineClient(this.host, this.port, this._decoder, this._encoder);
+        const cli: C.IPipelineClient = new PipelineClient(
+            this.host,
+            this.port,
+            this._createDecoder,
+            this._createEncoder
+        );
+
+        await cli.connect();
 
         if (this._password) {
 
diff --git a/src/lib/Commands.ts b/src/lib/Commands.ts
@@ -20,7 +20,7 @@ export interface ICommand {
 
 function isStringOK(data: any): boolean {
 
-    return "OK" === data;
+    return "OK" === (data && data.toString());
 }
 
 function isIntegerOne(data: any): boolean {
@@ -542,7 +542,10 @@ export const COMMANDS: Record<keyof C.ICommandAPIs, ICommand> = {
                 args: keys
             };
         },
-        process: U.pairList2NullableStringDict
+        process(data, args) {
+
+            return U.list2NullableStringDict(args, data);
+        }
     },
 
     /**
@@ -556,7 +559,10 @@ export const COMMANDS: Record<keyof C.ICommandAPIs, ICommand> = {
                 args: keys
             };
         },
-        process: U.pairList2NullableBufferDict
+        process(data, args) {
+
+            return U.list2NullableBufferDict(args, data);
+        }
     },
 
     /**
diff --git a/src/lib/Common.ts b/src/lib/Common.ts
@@ -549,7 +549,7 @@ export interface ICommandAPIs {
      * Command: hMSet
      * @see https://redis.io/commands/hMSet
      */
-    hMSet(key: string, kv: Record<string, string | Buffer>): Promise<void>;
+    hMSet(key: string, kv: Record<string, string | Buffer | number>): Promise<void>;
 
     /**
      * Command: hStrLen
@@ -978,3 +978,7 @@ export interface ISubscriberClient extends IProtocolClient {
      */
     pUnsubscribe(patterns: NonEmptyArray<string>): Promise<void>;
 }
+
+export type TEncoderFactory = () => IEncoder;
+
+export type TDecoderFactory = () => IDecoder;
diff --git a/src/lib/Errors.ts b/src/lib/Errors.ts
@@ -53,3 +53,9 @@ export const E_SUBSCRIBE_FAILURE = ErrorHub.define(
     "E_SUBSCRIBE_FAILURE",
     "Failed to subscribe subjects."
 );
+
+export const E_PIPELINING = ErrorHub.define(
+    null,
+    "E_PIPELINING",
+    "Some commands queued in pipeline, can not use MULTI."
+);
diff --git a/src/lib/ModuleAPI.ts b/src/lib/ModuleAPI.ts
@@ -19,8 +19,8 @@ export function createEncoder(): C.IEncoder {
 export function createCommandClient(
     host: string,
     port: number = DEFUALT_PORT,
-    decoder: C.IDecoder = createDecoder(),
-    encoder: C.IEncoder = createEncoder()
+    decoder: C.TDecoderFactory = createDecoder,
+    encoder: C.TEncoderFactory = createEncoder
 ): C.ICommandClient {
 
     return new CommandClient(
@@ -34,8 +34,8 @@ export function createCommandClient(
 export function createProtocolClient(
     host: string,
     port: number = DEFUALT_PORT,
-    decoder: C.IDecoder = createDecoder(),
-    encoder: C.IEncoder = createEncoder()
+    decoder: C.TDecoderFactory = createDecoder,
+    encoder: C.TEncoderFactory = createEncoder
 ): C.IProtocolClient {
 
     return new ProtocolClient(
@@ -49,8 +49,8 @@ export function createProtocolClient(
 export function createSubscriberClient(
     host: string,
     port: number = DEFUALT_PORT,
-    decoder: C.IDecoder = createDecoder(),
-    encoder: C.IEncoder = createEncoder()
+    decoder: C.TDecoderFactory = createDecoder,
+    encoder: C.TEncoderFactory = createEncoder
 ): C.ISubscriberClient {
 
     return new SubscriberClient(
diff --git a/src/lib/PipelineClient.ts b/src/lib/PipelineClient.ts
diff --git a/src/lib/ProtocolClient.ts b/src/lib/ProtocolClient.ts
diff --git a/src/lib/Subscriber.ts b/src/lib/Subscriber.ts
