iovalkey

While Valkey.Cluster supports transactions and pipelines, observe these constraints:

1. Key Locality: All keys in a pipeline must belong to slots served by the same node, as iovalkey sends the entire pipeline to a single node.
2. Multi Constraint: You cannot use multi without a pipeline (i.e., use cluster.multi({ pipeline: true })). Without a pipeline, the client cannot determine which node should receive the multi command.
3. Automatic Resending: If a pipeline receives MOVED or ASK errors, iovalkey will automatically resend the whole pipeline to the correct node ONLY if:
   • All errors in the pipeline are identical (e.g., all point to the same new node).
   • All successfully executed commands are read-only (to prevent side effects during resending).

You can use iovalkey with either Node.js callbacks or Native Promises. If the last argument to a method is not a function, it returns a Promise. All arguments are passed directly to the Valkey server, meaning all Valkey commands are supported via method names in lowercase.

const Valkey = require("iovalkey");
const valkey = new Valkey();

// Using Promises
await valkey.set("mykey", "value");
const result = await valkey.get("mykey");

// Using Callbacks
valkey.get("mykey", (err, result) => {
  if (err) console.error(err);
  else console.log(result);
});

// Direct command mapping (equivalent to CLI: SET mykey hello EX 10)
valkey.set("mykey", "hello", "EX", 10);

iovalkey supports binary data out of the box. You can send Buffer objects directly using standard commands like set. To retrieve data as a Buffer instead of a UTF-8 string, use the command variant with the Buffer suffix (e.g., getBuffer).

// Sending binary data
valkey.set("foo", Buffer.from([0x62, 0x75, 0x66]));

// Receiving binary data
const result = await valkey.getBuffer("foo");
// result is <Buffer 62 75 66>

// Using Buffer suffix to get old value during SET
const oldVal = await valkey.setBuffer("foo", "new value", "GET");

When performing asynchronous operations (like migrating data to another database) inside a data event handler, you should use stream.pause() and stream.resume() to prevent the stream from overwhelming the client or the destination. This ensures the scanning process waits for your async tasks to complete before fetching the next batch of keys.

const stream = valkey.scanStream();
stream.on("data", (resultKeys) => {
  // Pause the stream from scanning more keys until we've migrated the current keys.
  stream.pause();

  Promise.all(resultKeys.map(migrateKeyToMySQL)).then(() => {
    // Resume the stream here.
    stream.resume();
  });
});

stream.on("end", () => {
  console.log("done migration");
});

AWS ElastiCache for Valkey (Clustered Mode) requires TLS. To avoid invalid certificate errors, you must provide a dnsLookup function that returns the address as-is and enable tls in redisOptions.

const cluster = new Valkey.Cluster(
  [
    {
      host: "clustercfg.myCluster.abcdefg.xyz.cache.amazonaws.com",
      port: 6379,
    },
  ],
  {
    dnsLookup: (address, callback) => callback(null, address),
    redisOptions: {
      tls: {},
    },
  }
);

A Valkey instance creates a connection upon instantiation. You can connect using a port number, a host/port combination, a Unix socket, a configuration object, or a connection URI (supporting valkey://, valkeys:// for TLS, and legacy redis:///rediss:// schemes).

// Default: 127.0.0.1:6379
new Valkey();

// By port
new Valkey(6380);

// By host and port
new Valkey(6379, "192.168.1.1");

// By Unix socket
new Valkey("/tmp/valkey.sock");

// By configuration object
new Valkey({
  port: 6379,
  host: "127.0.0.1",
  username: "default",
  password: "my-top-secret",
  db: 0
});

// By URI
new Valkey("valkey://username:authpassword@127.0.0.1:6380/4");

Install the iovalkey package using npm. If you are working in a TypeScript project, you should also install the Node.js type definitions.

npm install iovalkey

# For TypeScript projects
npm install --save-dev @types/node

You can use xread to consume messages from a stream. When using the block argument, the client will wait for new data. To continuously listen, pass the ID of the last received message back into the next xread call.

const Valkey = require("iovalkey");
const valkey = new Valkey();

async function listenForMessage(lastID = "$") {
  // results is an array of [key, messages]
  const results = await valkey.xread("block", 0, "STREAMS", "mystream", lastID);
  const [key, messages] = results[0];

  messages.forEach(msg => {
    console.log("ID: %s. Data: %O", msg[0], msg[1]);
  });

  // Use the last ID from the results for the next iteration
  await listenForMessage(messages[messages.length - 1][0]);
}

listenForMessage();