Fix autopipeling for buffer function + added built in commands

lib/autoPipelining.ts

@@ -17,7 +17,11 @@ export const notAllowedAutoPipelineCommands = [
   "unpsubscribe",
 ];
 
-function findAutoPipeline(client, _commandName, ...args: Array<string>): string {
+function findAutoPipeline(
+  client,
+  _commandName,
+  ...args: Array<string>
+): string {
   if (!client.isCluster) {
     return "main";
   }
@@ -68,8 +72,13 @@ function executeAutoPipeline(client, slotKey: string) {
   });
 }
 
-export function shouldUseAutoPipelining(client, commandName: string): boolean {
+export function shouldUseAutoPipelining(
+  client,
+  functionName: string,
+  commandName: string
+): boolean {
   return (
+    functionName &&
     client.options.enableAutoPipelining &&
     !client.isPipeline &&
     !notAllowedAutoPipelineCommands.includes(commandName) &&
@@ -79,6 +88,7 @@ export function shouldUseAutoPipelining(client, commandName: string): boolean {
 
 export function executeWithAutoPipelining(
   client,
+  functionName: string,
   commandName: string,
   args: string[],
   callback
@@ -94,10 +104,13 @@ export function executeWithAutoPipelining(
           return;
         }
 
-        executeWithAutoPipelining(client, commandName, args, callback).then(
-          resolve,
-          reject
-        );
+        executeWithAutoPipelining(
+          client,
+          functionName,
+          commandName,
+          args,
+          callback
+        ).then(resolve, reject);
       });
     });
   }
@@ -138,7 +151,7 @@ export function executeWithAutoPipelining(
       resolve(value);
     });
 
-    pipeline[commandName](...args);
+    pipeline[functionName](...args);
   });
 
   return asCallback(autoPipelinePromise, callback);

lib/commander.ts

@@ -31,6 +31,7 @@ export default function Commander() {
     showFriendlyErrorStack: false,
   });
   this.scriptsSet = {};
+  this.addedBuiltinSet = new Set();
 }
 
 const commands = require("redis-commands").list.filter(function (command) {
@@ -57,21 +58,43 @@ Commander.prototype.getBuiltinCommands = function () {
  */
 Commander.prototype.createBuiltinCommand = function (commandName) {
   return {
-    string: generateFunction(commandName, "utf8"),
-    buffer: generateFunction(commandName, null),
+    string: generateFunction(null, commandName, "utf8"),
+    buffer: generateFunction(null, commandName, null),
   };
 };
 
+/**
+ * Create add builtin command
+ *
+ * @param {string} commandName - command name
+ * @return {object} functions
+ * @public
+ */
+Commander.prototype.addBuiltinCommand = function (commandName) {
+  this.addedBuiltinSet.add(commandName);
+  this[commandName] = generateFunction(commandName, commandName, "utf8");
+  this[commandName + "Buffer"] = generateFunction(
+    commandName + "Buffer",
+    commandName,
+    null
+  );
+};
+
 commands.forEach(function (commandName) {
-  Commander.prototype[commandName] = generateFunction(commandName, "utf8");
+  Commander.prototype[commandName] = generateFunction(
+    commandName,
+    commandName,
+    "utf8"
+  );
   Commander.prototype[commandName + "Buffer"] = generateFunction(
+    commandName + "Buffer",
     commandName,
     null
   );
 });
 
-Commander.prototype.call = generateFunction("utf8");
-Commander.prototype.callBuffer = generateFunction(null);
+Commander.prototype.call = generateFunction("call", "utf8");
+Commander.prototype.callBuffer = generateFunction("callBuffer", null);
 // eslint-disable-next-line @typescript-eslint/camelcase
 Commander.prototype.send_command = Commander.prototype.call;
 
@@ -93,8 +116,13 @@ Commander.prototype.defineCommand = function (name, definition) {
     definition.readOnly
   );
   this.scriptsSet[name] = script;
-  this[name] = generateScriptingFunction(name, script, "utf8");
-  this[name + "Buffer"] = generateScriptingFunction(name, script, null);
+  this[name] = generateScriptingFunction(name, name, script, "utf8");
+  this[name + "Buffer"] = generateScriptingFunction(
+    name + "Buffer",
+    name,
+    script,
+    null
+  );
 };
 
 /**
@@ -105,9 +133,17 @@ Commander.prototype.defineCommand = function (name, definition) {
  */
 Commander.prototype.sendCommand = function () {};
 
-function generateFunction(_encoding: string);
-function generateFunction(_commandName: string | void, _encoding: string);
-function generateFunction(_commandName?: string, _encoding?: string) {
+function generateFunction(functionName: string | null, _encoding: string);
+function generateFunction(
+  functionName: string | null,
+  _commandName: string | void,
+  _encoding: string
+);
+function generateFunction(
+  functionName: string | null,
+  _commandName?: string,
+  _encoding?: string
+) {
   if (typeof _encoding === "undefined") {
     _encoding = _commandName;
     _commandName = null;
@@ -139,18 +175,29 @@ function generateFunction(_commandName?: string, _encoding?: string) {
     }
 
     // No auto pipeline, use regular command sending
-    if (!shouldUseAutoPipelining(this, commandName)) {
+    if (!shouldUseAutoPipelining(this, functionName, commandName)) {
       return this.sendCommand(
         new Command(commandName, args, options, callback)
       );
     }
 
     // Create a new pipeline and make sure it's scheduled
-    return executeWithAutoPipelining(this, commandName, args, callback);
+    return executeWithAutoPipelining(
+      this,
+      functionName,
+      commandName,
+      args,
+      callback
+    );
   };
 }
 
-function generateScriptingFunction(name, script, encoding) {
+function generateScriptingFunction(
+  functionName,
+  commandName,
+  script,
+  encoding
+) {
   return function () {
     let length = arguments.length;
     const lastArgIndex = length - 1;
@@ -183,11 +230,17 @@ function generateScriptingFunction(name, script, encoding) {
     }
 
     // No auto pipeline, use regular command sending
-    if (!shouldUseAutoPipelining(this, name)) {
+    if (!shouldUseAutoPipelining(this, functionName, commandName)) {
       return script.execute(this, args, options, callback);
     }
 
     // Create a new pipeline and make sure it's scheduled
-    return executeWithAutoPipelining(this, name, args, callback);
+    return executeWithAutoPipelining(
+      this,
+      functionName,
+      commandName,
+      args,
+      callback
+    );
   };
 }

lib/pipeline.ts

@@ -47,6 +47,11 @@ export default function Pipeline(redis) {
     this[name + "Buffer"] = redis[name + "Buffer"];
   });
 
+  redis.addedBuiltinSet.forEach((name) => {
+    this[name] = redis[name];
+    this[name + "Buffer"] = redis[name + "Buffer"];
+  });
+
   const Promise = PromiseContainer.get();
   this.promise = new Promise((resolve, reject) => {
     this.resolve = resolve;

test/functional/autopipelining.ts

@@ -1,5 +1,7 @@
 import { expect, use } from "chai";
 import Redis from "../../lib/redis";
+import { ReplyError } from "redis-errors";
+import * as sinon from "sinon";
 
 use(require("chai-as-promised"));
 
@@ -44,6 +46,15 @@ describe("autoPipelining for single node", function () {
     await promise;
   });
 
+  it("should support buffer commands", async () => {
+    const redis = new Redis({ enableAutoPipelining: true });
+    const buffer = Buffer.from("bar");
+    await redis.setBuffer("foo", buffer);
+    const promise = redis.getBuffer("foo");
+    expect(redis.autoPipelineQueueSize).to.eql(1);
+    expect(await promise).to.eql(buffer);
+  });
+
   it("should support custom commands", async () => {
     const redis = new Redis({ enableAutoPipelining: true });
 

test/functional/cluster/autopipelining.ts

@@ -1,5 +1,5 @@
 import { expect, use } from "chai";
-import * as calculateKeySlot from 'cluster-key-slot';
+import * as calculateKeySlot from "cluster-key-slot";
 
 import { default as Cluster } from "../../../lib/cluster";
 import MockServer from "../../helpers/mock_server";
@@ -400,11 +400,11 @@ describe("autoPipelining for cluster", function () {
     const promise2 = cluster.set("foo5", "bar");
     const promise3 = cluster.set("foo2", "bar");
     const promise4 = cluster.set("foo6", "bar");
-    
+
     // Override slots to induce a failure
-    const key1Slot = calculateKeySlot('foo1');
-    const key2Slot = calculateKeySlot('foo2');
-    const key5Slot = calculateKeySlot('foo5');
+    const key1Slot = calculateKeySlot("foo1");
+    const key2Slot = calculateKeySlot("foo2");
+    const key5Slot = calculateKeySlot("foo5");
     cluster.slots[key1Slot] = cluster.slots[key2Slot];
     cluster.slots[key2Slot] = cluster.slots[key5Slot];
 
@@ -492,9 +492,9 @@ describe("autoPipelining for cluster", function () {
       expect(cluster.autoPipelineQueueSize).to.eql(4);
 
       // Override slots to induce a failure
-      const key1Slot = calculateKeySlot('foo1');
-      const key2Slot = calculateKeySlot('foo2');
-      const key5Slot = calculateKeySlot('foo5');
+      const key1Slot = calculateKeySlot("foo1");
+      const key2Slot = calculateKeySlot("foo2");
+      const key5Slot = calculateKeySlot("foo5");
       cluster.slots[key1Slot] = cluster.slots[key2Slot];
       cluster.slots[key2Slot] = cluster.slots[key5Slot];
     });
@@ -541,8 +541,8 @@ describe("autoPipelining for cluster", function () {
 
       expect(cluster.autoPipelineQueueSize).to.eql(3);
 
-      const key1Slot = calculateKeySlot('foo1');
-      const key2Slot = calculateKeySlot('foo2');
+      const key1Slot = calculateKeySlot("foo1");
+      const key2Slot = calculateKeySlot("foo2");
       cluster.slots[key1Slot] = cluster.slots[key2Slot];
     });
   });

test/functional/pipeline.ts

@@ -1,5 +1,6 @@
 import Redis from "../../lib/redis";
 import { expect } from "chai";
+import * as sinon from "sinon";
 
 describe("pipeline", function () {
   it("should return correct result", function (done) {
@@ -138,6 +139,16 @@ describe("pipeline", function () {
       });
   });
 
+  it("should include added built in commands", async () => {
+    const redis = new Redis({ keyPrefix: "foo:" });
+    redis.addBuiltinCommand("someCommand");
+    sinon.stub(redis, "sendCommand").callsFake((command) => {
+      command.resolve(Buffer.from("OK"));
+    });
+    const result = await redis.pipeline().someCommand().exec();
+    expect(result).to.eql([[null, "OK"]]);
+  });
+
   describe("custom commands", function () {
     let redis;
 

test/unit/commander.ts

@@ -13,6 +13,28 @@ describe("Commander", function () {
     });
   });
 
+  describe("#addBuiltinCommand()", () => {
+    beforeEach(() => sinon.spy(Commander.prototype, "sendCommand"));
+    afterEach(() => sinon.restore());
+    it("adds string command", () => {
+      const c = new Commander();
+      c.addBuiltinCommand("someCommand");
+      c.someCommand();
+      const command = Commander.prototype.sendCommand.getCall(0).args[0];
+      expect(command.name).to.eql("someCommand");
+      expect(command.replyEncoding).to.eql("utf8");
+    });
+
+    it("adds buffer command", () => {
+      const c = new Commander();
+      c.addBuiltinCommand("someCommand");
+      c.someCommandBuffer();
+      const command = Commander.prototype.sendCommand.getCall(0).args[0];
+      expect(command.name).to.eql("someCommand");
+      expect(command.replyEncoding).to.eql(null);
+    });
+  });
+
   it("should pass the correct arguments", function () {
     sinon.stub(Commander.prototype, "sendCommand").callsFake((command) => {
       return command;