Skip to content

Commit

Permalink
feat: add support for the ioredis package
Browse files Browse the repository at this point in the history
Related: #2
  • Loading branch information
darrachequesne committed Feb 13, 2024
1 parent 65d0539 commit 58faa1d
Show file tree
Hide file tree
Showing 5 changed files with 306 additions and 48 deletions.
42 changes: 16 additions & 26 deletions lib/adapter.ts
Original file line number Diff line number Diff line change
@@ -1,14 +1,13 @@
import type { PrivateSessionId, Session } from "socket.io-adapter";
import { decode, encode } from "@msgpack/msgpack";
import { commandOptions } from "redis";
import {
ClusterAdapter,
ClusterAdapterOptions,
ClusterMessage,
MessageType,
} from "./cluster-adapter";
import debugModule from "debug";
import { hasBinary } from "./util";
import { hasBinary, XADD, XREAD } from "./util";

const debug = debugModule("socket.io-redis-streams-adapter");

Expand Down Expand Up @@ -61,23 +60,15 @@ export function createAdapter(
);
let offset = "$";
let polling = false;
let shouldClose = false;

async function poll() {
try {
let response = await redisClient.xRead(
commandOptions({
isolated: true,
}),
[
{
key: options.streamName,
id: offset,
},
],
{
COUNT: options.readCount,
BLOCK: 5000,
}
let response = await XREAD(
redisClient,
options.streamName,
offset,
options.readCount
);

if (response) {
Expand All @@ -98,7 +89,7 @@ export function createAdapter(
debug("something went wrong while consuming the stream: %s", e.message);
}

if (namespaceToAdapters.size > 0 && redisClient.isOpen) {
if (namespaceToAdapters.size > 0 && !shouldClose) {
poll();
} else {
polling = false;
Expand All @@ -111,6 +102,7 @@ export function createAdapter(

if (!polling) {
polling = true;
shouldClose = false;
poll();
}

Expand All @@ -119,6 +111,10 @@ export function createAdapter(
adapter.close = () => {
namespaceToAdapters.delete(nsp.name);

if (namespaceToAdapters.size === 0) {
shouldClose = true;
}

defaultClose.call(adapter);
};

Expand All @@ -141,17 +137,11 @@ class RedisStreamsAdapter extends ClusterAdapter {
override doPublish(message: ClusterMessage) {
debug("publishing %o", message);

return this.#redisClient.xAdd(
return XADD(
this.#redisClient,
this.#opts.streamName,
"*",
RedisStreamsAdapter.encode(message),
{
TRIM: {
strategy: "MAXLEN",
strategyModifier: "~",
threshold: this.#opts.maxLen,
},
}
this.#opts.maxLen
);
}

Expand Down
91 changes: 91 additions & 0 deletions lib/util.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { randomBytes } from "crypto";
import { commandOptions } from "redis";

export function hasBinary(obj: any, toJSON?: boolean): boolean {
if (!obj || typeof obj !== "object") {
Expand Down Expand Up @@ -34,3 +35,93 @@ export function hasBinary(obj: any, toJSON?: boolean): boolean {
export function randomId() {
return randomBytes(8).toString("hex");
}

/**
* Whether the client comes from the `redis` package
*
* @param redisClient
*
* @see https://github.com/redis/node-redis
*/
function isRedisV4Client(redisClient: any) {
return typeof redisClient.sSubscribe === "function";
}

/**
* @see https://redis.io/commands/xread/
*/
export function XREAD(
redisClient: any,
streamName: string,
offset: string,
readCount: number
) {
if (isRedisV4Client(redisClient)) {
return redisClient.xRead(
commandOptions({
isolated: true,
}),
[
{
key: streamName,
id: offset,
},
],
{
COUNT: readCount,
BLOCK: 5000,
}
);
} else {
return redisClient
.xread("BLOCK", 100, "COUNT", readCount, "STREAMS", streamName, offset)
.then((results) => {
if (results === null) {
return null;
}
return [
{
messages: results[0][1].map((result) => {
const id = result[0];
const inlineValues = result[1];
const message = {};
for (let i = 0; i < inlineValues.length; i += 2) {
message[inlineValues[i]] = inlineValues[i + 1];
}
return {
id,
message,
};
}),
},
];
});
}
}

/**
* @see https://redis.io/commands/xadd/
*/
export function XADD(
redisClient: any,
streamName: string,
payload: any,
maxLenThreshold: number
) {
if (isRedisV4Client(redisClient)) {
return redisClient.xAdd(streamName, "*", payload, {
TRIM: {
strategy: "MAXLEN",
strategyModifier: "~",
threshold: maxLenThreshold,
},
});
} else {
const args = [streamName, "MAXLEN", "~", maxLenThreshold, "*"];
Object.keys(payload).forEach((k) => {
args.push(k, payload[k]);
});

return redisClient.xadd.call(redisClient, args);
}
}

0 comments on commit 58faa1d

Please sign in to comment.