Unverified Commit 1d51c69c authored by Hong Minhee (洪 民憙)'s avatar Hong Minhee (洪 民憙) Committed by GitHub
Browse files

Merge pull request #368 from mike182uk/redis-cluster-support

parents 7c78ce1a 2f1a39f5

CHANGES.md

+10 −0
Original line number Diff line number Diff line
@@ -64,6 +64,9 @@ To be released. 
    hanging on slow or unresponsive servers.

    [[#258] by Hyunchae Kim]



[#353]: https://github.com/fedify-dev/fedify/issues/353

[#365]: https://github.com/fedify-dev/fedify/pull/365



### @fedify/next



 -  Created [Next.js] integration as the *@fedify/next* package.
@@ -72,6 +75,13 @@ To be released. 
[Next.js]: https://nextjs.org/

[#313]: https://github.com/fedify-dev/fedify/issues/313



### @fedify/redis



 -  Added support for Redis Cluster to the *@fedify/redis* package.

    [[#368] by Michael Barrett]



[#368]: https://github.com/fedify-dev/fedify/pull/368





Version 1.8.5

-------------

packages/redis/README.md

+16 −1
Original line number Diff line number Diff line
@@ -16,12 +16,27 @@ implementations for Redis: 
~~~~ typescript

import { createFederation } from "@fedify/fedify";

import { RedisKvStore, RedisMessageQueue } from "@fedify/redis";

import { Redis } from "ioredis";

import { Redis, Cluster } from "ioredis";



// Using a standalone Redis instance:

const federation = createFederation({

  kv: new RedisKvStore(new Redis()),

  queue: new RedisMessageQueue(() => new Redis()),

});



// Using a Redis Cluster:

const federation = createFederation({

  kv: new RedisKvStore(new Cluster([

    { host: "127.0.0.1", port: 7000 },

    { host: "127.0.0.1", port: 7001 },

    { host: "127.0.0.1", port: 7002 },

  ])),

  queue: new RedisMessageQueue(() => new Cluster([

    { host: "127.0.0.1", port: 7000 },

    { host: "127.0.0.1", port: 7001 },

    { host: "127.0.0.1", port: 7002 },

  ])),

});

~~~~



[JSR]: https://jsr.io/@fedify/redis

packages/redis/src/kv.ts

+17 −5
Original line number Diff line number Diff line 
import type { KvKey, KvStore, KvStoreSetOptions } from "@fedify/fedify";

import type { Redis, RedisKey } from "ioredis";

import type { Cluster, Redis, RedisKey } from "ioredis";

import { Buffer } from "node:buffer";

import { type Codec, JsonCodec } from "./codec.ts";
@@ -27,26 +27,38 @@ export interface RedisKvStoreOptions { 
 * ```ts ignore

 * import { createFederation } from "@fedify/fedify";

 * import { RedisKvStore } from "@fedify/redis";

 * import { Redis } from "ioredis";

 * import { Redis, Cluster } from "ioredis";

 *

 * // Using a standalone Redis instance:

 * const federation = createFederation({

 *   // ...

 *   kv: new RedisKvStore(new Redis()),

 * });

 *

 * // Using a Redis Cluster:

 * const cluster = new Cluster([

 *   { host: "127.0.0.1", port: 7000 },

 *   { host: "127.0.0.1", port: 7001 },

 *   { host: "127.0.0.1", port: 7002 },

 * ]);

 * const federation = createFederation({

 *   // ...

 *   kv: new RedisKvStore(cluster),

 * });

 * ```

 */

export class RedisKvStore implements KvStore {

  #redis: Redis;

  #redis: Redis | Cluster;

  #keyPrefix: RedisKey;

  #codec: Codec;

  #textEncoder = new TextEncoder();



  /**

   * Creates a new Redis key–value store.

   * @param redis The Redis client to use.

   * @param redis The Redis client (standalone or cluster) to use.

   * @param options The options for the key–value store.

   */

  constructor(redis: Redis, options: RedisKvStoreOptions = {}) {

  constructor(redis: Redis | Cluster, options: RedisKvStoreOptions = {}) {

    this.#redis = redis;

    this.#keyPrefix = options.keyPrefix ?? "fedify::";

    this.#codec = options.codec ?? new JsonCodec();

packages/redis/src/mq.ts

+36 −7
Original line number Diff line number Diff line
@@ -5,7 +5,7 @@ import type { 
  MessageQueueListenOptions,

} from "@fedify/fedify";

import { getLogger } from "@logtape/logtape";

import type { Redis, RedisKey } from "ioredis";

import type { Cluster, Redis, RedisKey } from "ioredis";

import { type Codec, JsonCodec } from "./codec.ts";



const logger = getLogger(["fedify", "redis", "mq"]);
@@ -63,17 +63,28 @@ export interface RedisMessageQueueOptions { 
 * ```ts ignore

 * import { createFederation } from "@fedify/fedify";

 * import { RedisMessageQueue } from "@fedify/redis";

 * import { Redis } from "ioredis";

 * import { Redis, Cluster } from "ioredis";

 *

 * // Using a standalone Redis instance:

 * const federation = createFederation({

 *   // ...

 *   queue: new RedisMessageQueue(() => new Redis()),

 * });

 *

 * // Using a Redis Cluster:

 * const federation = createFederation({

 *   // ...

 *   queue: new RedisMessageQueue(() => new Cluster([

 *     { host: "127.0.0.1", port: 7000 },

 *     { host: "127.0.0.1", port: 7001 },

 *     { host: "127.0.0.1", port: 7002 },

 *   ])),

 * });

 * ```

 */

export class RedisMessageQueue implements MessageQueue, Disposable {

  #redis: Redis;

  #subRedis: Redis;

  #redis: Redis | Cluster;

  #subRedis: Redis | Cluster;

  #workerId: string;

  #channelKey: RedisKey;

  #queueKey: RedisKey;
@@ -87,7 +98,10 @@ export class RedisMessageQueue implements MessageQueue, Disposable { 
   * @param redis The Redis client factory.

   * @param options The options for the message queue.

   */

  constructor(redis: () => Redis, options: RedisMessageQueueOptions = {}) {

  constructor(

    redis: () => Redis | Cluster,

    options: RedisMessageQueueOptions = {},

  ) {

    this.#redis = redis();

    this.#subRedis = redis();

    this.#workerId = options.workerId ?? crypto.randomUUID();
@@ -196,9 +210,24 @@ export class RedisMessageQueue implements MessageQueue, Disposable { 
      }

    };

    const promise = this.#subRedis.subscribe(this.#channelKey, () => {

      this.#subRedis.on("message", poll);

      /**

       * Cast to Redis for event methods. Both Redis and Cluster extend EventEmitter

       * and get the same methods via applyMixin at runtime, but their TypeScript

       * interfaces are incompatible:

       * - Redis declares specific overloads: on(event: "message", cb: (channel, message) => void)

       * - Cluster only has generic: on(event: string | symbol, listener: Function)

       *

       * This makes the union type Redis | Cluster incompatible for these method calls.

       * The cast is safe because both classes use applyMixin(Class, EventEmitter) which

       * copies all EventEmitter prototype methods, giving them identical pub/sub functionality.

       *

       * @see https://github.com/redis/ioredis/blob/main/lib/Redis.ts#L863 (has specific overloads)

       * @see https://github.com/redis/ioredis/blob/main/lib/cluster/index.ts#L1110 (empty interface)

       */

      const subRedis = this.#subRedis as Redis;

      subRedis.on("message", poll);

      signal?.addEventListener("abort", () => {

        this.#subRedis.off("message", poll);

        subRedis.off("message", poll);

      });

    });

    signal?.addEventListener(