fix: improve typing for pipeline

Author: luin

lib/DataHandler.ts

@@ -1,4 +1,4 @@
-import { NetStream, ICommandItem, ICommand } from "./types";
+import { NetStream, CommandItem, ICommand } from "./types";
 import Deque = require("denque");
 import { EventEmitter } from "events";
 import Command from "./command";
@@ -25,11 +25,11 @@ interface IDataHandledable extends EventEmitter {
   stream: NetStream;
   status: string;
   condition: ICondition;
-  commandQueue: Deque<ICommandItem>;
+  commandQueue: Deque<CommandItem>;
 
   disconnect(reconnect: boolean): void;
   recoverFromFatalError(commandError: Error, err: Error, options: any): void;
-  handleReconnection(err: Error, item: ICommandItem): void;
+  handleReconnection(err: Error, item: CommandItem): void;
 }
 
 interface IParserOptions {
@@ -119,7 +119,11 @@ export default class DataHandler {
       case "message":
         if (this.redis.listeners("message").length > 0) {
           // Check if there're listeners to avoid unnecessary `toString()`.
-          this.redis.emit("message", reply[1].toString(), reply[2] ? reply[2].toString() : '');
+          this.redis.emit(
+            "message",
+            reply[1].toString(),
+            reply[2] ? reply[2].toString() : ""
+          );
         }
         this.redis.emit("messageBuffer", reply[1], reply[2]);
         break;
@@ -208,7 +212,7 @@ export default class DataHandler {
     return true;
   }
 
-  private shiftCommand(reply: ReplyData | Error): ICommandItem | null {
+  private shiftCommand(reply: ReplyData | Error): CommandItem | null {
     const item = this.redis.commandQueue.shift();
     if (!item) {
       const message =

lib/Redis.ts

@@ -10,7 +10,12 @@ import {
   RedisOptions,
 } from "./redis/RedisOptions";
 import { addTransactionSupport, Transaction } from "./transaction";
-import { CallbackFunction, ICommandItem, NetStream } from "./types";
+import {
+  CallbackFunction,
+  CommandItem,
+  NetStream,
+  WriteableStream,
+} from "./types";
 import {
   CONNECTION_CLOSED_ERROR_MSG,
   Debug,
@@ -53,7 +58,7 @@ class Redis extends Commander {
 
   options: RedisOptions;
   status: RedisStatus = "wait";
-  commandQueue: Deque;
+  commandQueue: Deque<CommandItem>;
   offlineQueue: Deque;
   connectionEpoch = 0;
   connector: AbstractConnector;
@@ -360,7 +365,7 @@ class Redis extends Commander {
     this.disconnect(true);
   }
 
-  handleReconnection(err: Error, item: ICommandItem) {
+  handleReconnection(err: Error, item: CommandItem) {
     let needReconnect: ReturnType<ReconnectOnError> = false;
     if (this.options.reconnectOnError) {
       needReconnect = this.options.reconnectOnError(err);
@@ -575,7 +580,7 @@ class Redis extends Commander {
    * redis.sendCommand(set);
    * ```
    */
-  sendCommand(command: Command, stream?: NetStream): unknown {
+  sendCommand(command: Command, stream?: WriteableStream): unknown {
     if (this.status === "wait") {
       this.connect().catch(noop);
     }
@@ -648,7 +653,7 @@ class Redis extends Commander {
       }
 
       if (stream) {
-        if (stream.isPipeline) {
+        if ("isPipeline" in stream && stream.isPipeline) {
           stream.write(command.toWritable(stream.destination.redis.stream));
         } else {
           stream.write(command.toWritable(stream));

lib/cluster/index.ts

@@ -1,39 +1,41 @@
 import { EventEmitter } from "events";
-import ClusterAllFailedError from "../errors/ClusterAllFailedError";
-import { defaults, noop, Debug } from "../utils";
-import ConnectionPool from "./ConnectionPool";
-import {
-  NodeKey,
-  RedisOptions,
-  normalizeNodeOptions,
-  NodeRole,
-  getUniqueHostnamesFromOptions,
-  nodeKeyToRedisOptions,
-  groupSrvRecords,
-  weightSrvRecords,
-  getConnectionName,
-} from "./util";
-import ClusterSubscriber from "./ClusterSubscriber";
-import DelayQueue from "./DelayQueue";
-import ScanStream from "../ScanStream";
+import * as commands from "redis-commands";
 import { AbortError, RedisError } from "redis-errors";
 import asCallback from "standard-as-callback";
-import { CallbackFunction, NetStream } from "../types";
-import { ClusterOptions, DEFAULT_CLUSTER_OPTIONS } from "./ClusterOptions";
+import { Pipeline } from "..";
+import Command from "../command";
+import ClusterAllFailedError from "../errors/ClusterAllFailedError";
+import Redis from "../Redis";
+import ScanStream from "../ScanStream";
+import { CallbackFunction, WriteableStream } from "../types";
 import {
-  sample,
   CONNECTION_CLOSED_ERROR_MSG,
+  Debug,
+  defaults,
+  noop,
+  sample,
   shuffle,
   timeout,
   zipMap,
 } from "../utils";
-import * as commands from "redis-commands";
-import Command from "../command";
-import Redis from "../Redis";
+import applyMixin from "../utils/applyMixin";
 import Commander from "../utils/Commander";
+import { ClusterOptions, DEFAULT_CLUSTER_OPTIONS } from "./ClusterOptions";
+import ClusterSubscriber from "./ClusterSubscriber";
+import ConnectionPool from "./ConnectionPool";
+import DelayQueue from "./DelayQueue";
+import {
+  getConnectionName,
+  getUniqueHostnamesFromOptions,
+  groupSrvRecords,
+  NodeKey,
+  nodeKeyToRedisOptions,
+  NodeRole,
+  normalizeNodeOptions,
+  RedisOptions,
+  weightSrvRecords,
+} from "./util";
 import Deque = require("denque");
-import { Pipeline } from "..";
-import applyMixin from "../utils/applyMixin";
 
 const debug = Debug("cluster");
 
@@ -573,7 +575,7 @@ class Cluster extends Commander {
       : nodeKey;
   }
 
-  sendCommand(command: Command, stream?: NetStream, node?: any): unknown {
+  sendCommand(command: Command, stream?: WriteableStream, node?: any): unknown {
     if (this.status === "wait") {
       this.connect().catch(noop);
     }

lib/command.ts

@@ -9,7 +9,7 @@ import {
   convertObjectToArray,
 } from "./utils";
 import { flatten } from "./utils/lodash";
-import { CallbackFunction, ICommand, CommandParameter, NetStream } from "./types";
+import { CallbackFunction, ICommand, CommandParameter } from "./types";
 
 export type ArgumentType =
   | string
@@ -273,7 +273,7 @@ export default class Command implements ICommand {
    * @see {@link Redis#sendCommand}
    * @public
    */
-  public toWritable(socket: NetStream): string | Buffer {
+  public toWritable(_socket: object): string | Buffer {
     let bufferMode = false;
     for (const arg of this.args) {
       if (arg instanceof Buffer) {

lib/pipeline.ts

@@ -4,9 +4,9 @@ import asCallback from "standard-as-callback";
 import { deprecate } from "util";
 import Redis, { Cluster } from ".";
 import Command from "./command";
-import Commander from "./utils/Commander";
-import { CallbackFunction } from "./types";
+import { CallbackFunction, PipelineWriteableStream } from "./types";
 import { noop } from "./utils";
+import Commander from "./utils/Commander";
 
 /*
   This function derives from the cluster-key-slot implementation.
@@ -344,7 +344,7 @@ Pipeline.prototype.exec = function (
     }
     let bufferMode = false;
 
-    const stream = {
+    const stream: PipelineWriteableStream = {
       isPipeline: true,
       destination: _this.isCluster ? node : { redis: _this.redis },
       write: function (writable) {

lib/redis/event_handler.ts

@@ -4,7 +4,7 @@ import Deque = require("denque");
 import { AbortError } from "redis-errors";
 import Command from "../command";
 import { MaxRetriesPerRequestError } from "../errors";
-import { ICommandItem, ICommand } from "../types";
+import { CommandItem, ICommand } from "../types";
 import { Debug, noop, CONNECTION_CLOSED_ERROR_MSG } from "../utils";
 import DataHandler from "../DataHandler";
 
@@ -115,7 +115,7 @@ function abortError(command: ICommand) {
 // the connection close and those pipelined commands must be aborted. For
 // example, if the queue looks like this: [2, 3, 4, 0, 1, 2] then after
 // aborting and purging we'll have a queue that looks like this: [0, 1, 2]
-function abortIncompletePipelines(commandQueue: Deque<ICommandItem>) {
+function abortIncompletePipelines(commandQueue: Deque<CommandItem>) {
   let expectedIndex = 0;
   for (let i = 0; i < commandQueue.length; ) {
     const command = commandQueue.peekAt(i).command as Command;
@@ -135,7 +135,7 @@ function abortIncompletePipelines(commandQueue: Deque<ICommandItem>) {
 // If only a partial transaction result was received before connection close,
 // we have to abort any transaction fragments that may have ended up in the
 // offline queue
-function abortTransactionFragments(commandQueue: Deque<ICommandItem>) {
+function abortTransactionFragments(commandQueue: Deque<CommandItem>) {
   for (let i = 0; i < commandQueue.length; ) {
     const command = commandQueue.peekAt(i).command as Command;
     if (command.name === "multi") {

lib/script.ts

@@ -1,10 +1,10 @@
 import { createHash } from "crypto";
 import Command from "./command";
 import asCallback from "standard-as-callback";
-import { CallbackFunction, NetStream } from "./types";
+import { CallbackFunction } from "./types";
 export default class Script {
   private sha: string;
-  private Command;
+  private Command: new (...args: any[]) => Command;
 
   constructor(
     private lua: string,
@@ -17,7 +17,7 @@ export default class Script {
     const sha = this.sha;
     const socketHasScriptLoaded = new WeakSet();
     this.Command = class CustomScriptCommand extends Command {
-      public toWritable(socket: NetStream): string | Buffer {
+      public toWritable(socket: object): string | Buffer {
         const origReject = this.reject;
         this.reject = (err) => {
           if (err.toString().indexOf("NOSCRIPT") !== -1) {

lib/types.ts

@@ -15,8 +15,16 @@ export interface ICommand {
   reject(error: Error): void;
 }
 
-export interface ICommandItem {
+export interface PipelineWriteableStream {
+  isPipeline: true;
+  write(data: string | Buffer): unknown;
+  destination: { redis: { stream: NetStream } };
+}
+
+export type WriteableStream = NetStream | PipelineWriteableStream;
+
+export interface CommandItem {
   command: ICommand;
-  stream: NetStream;
+  stream: WriteableStream;
   select: number;
 }

lib/utils/Commander.ts

@@ -5,7 +5,7 @@ import {
 } from "../autoPipelining";
 import Command, { ArgumentType } from "../command";
 import Script from "../script";
-import { CallbackFunction, NetStream } from "../types";
+import { CallbackFunction, WriteableStream } from "../types";
 import RedisCommander, { ClientContext } from "./RedisCommander";
 
 export interface CommanderOptions {
@@ -93,7 +93,11 @@ class Commander<Context extends ClientContext = { type: "default" }> {
     );
   }
 
-  sendCommand(command: Command, stream?: NetStream, node?: unknown): unknown {
+  sendCommand(
+    command: Command,
+    stream?: WriteableStream,
+    node?: unknown
+  ): unknown {
     throw new Error('"sendCommand" is not implemented');
   }
 }