Function executeWithAutoPipelining

lib/autoPipelining.ts:134–222

(
  client,
  functionName: string,
  commandName: string,
  args: ArgumentType[],
  callback
)

}

export function executeWithAutoPipelining(
  client,
  functionName: string,
  commandName: string,
  args: ArgumentType[],
  callback
) {
  // On cluster mode let's wait for slots to be available
  if (client.isCluster && !client.slots.length) {
    if (client.status === "wait") client.connect().catch(noop);
    return asCallback(
      new Promise(function (resolve, reject) {
        client.delayUntilReady((err) => {
          if (err) {
            reject(err);
            return;
          }

          executeWithAutoPipelining(
            client,
            functionName,
            commandName,
            args,
            null
          ).then(resolve, reject);
        });
      }),
      callback
    );
  }

  // If we have slot information, we can improve routing by grouping slots served by the same subset of nodes
  const prefix = client.options.keyPrefix || "";
  let slotKey = client.isCluster
    ? client.slots[
        calculateSlot(`${prefix}${getFirstKeyForCommand(commandName, args)}`)
      ].join(",")
    : "main";

  // When scaleReads is enabled, separate read and write commands into different pipelines
  // so they can be routed to replicas and masters respectively
  if (client.isCluster && client.options.scaleReads !== "master") {
    const isReadOnly = exists(commandName) && hasFlag(commandName, "readonly");
    slotKey += isReadOnly ? ":read" : ":write";
  }

  if (!client._autoPipelines.has(slotKey)) {
    const pipeline = client.pipeline();
    pipeline[kExec] = false;
    pipeline[kCallbacks] = [];
    client._autoPipelines.set(slotKey, pipeline);
  }

  const pipeline = client._autoPipelines.get(slotKey);

  /* 
    Mark the pipeline as scheduled.
    The symbol will make sure that the pipeline is only scheduled once per tick.