Merge pull request #193 from quirrel-dev/reduce-redis-client-number

concentrate all subscriptions onto one client

Author: Skn0tt

package-lock.json

@@ -34,12 +34,12 @@
     "@types/chai": "4.2.22",
     "@types/debug": "4.1.7",
     "@types/ioredis": "4.27.5",
+    "@types/minimatch": "^3.0.5",
     "@types/mocha": "9.0.0",
     "@types/node": "16.6.1",
     "@types/pino": "6.3.11",
     "chai": "4.3.4",
     "delay": "5.0.0",
-    "glob": "7.2.0",
     "mocha": "9.1.2",
     "nyc": "15.1.0",
     "ts-node": "10.2.1",
@@ -48,6 +48,7 @@
   "dependencies": {
     "ioredis": "^4.27.1",
     "ioredis-mock": "^5.5.6",
+    "minimatch": "^3.0.4",
     "opentracing": "^0.14.5",
     "pino": "^6.11.3"
   }

package.json

@@ -2,7 +2,7 @@ import { Redis } from "ioredis";
 import { Closable } from "../Closable";
 import { decodeRedisKey, encodeRedisKey } from "../encodeRedisKey";
 import { Job } from "../Job";
-import { Producer } from "../producer/producer";
+import minimatch from "minimatch";
 
 /**
  * Like String.split, but has a maximum number of delimiters it picks up.
@@ -34,19 +34,21 @@ function splitEvent(message: string, maxParts: number, delimiter = ":") {
 }
 
 export type SubscriptionOptions = { queue?: string; id?: string };
-export type OnActivity = (event: OnActivityEvent) => Promise<void> | void;
+export type OnActivity<ScheduleType extends string> = (
+  event: OnActivityEvent<ScheduleType>
+) => Promise<void> | void;
 
-export type OnActivityEvent =
-  | ScheduledEvent
+export type OnActivityEvent<ScheduleType extends string> =
+  | ScheduledEvent<ScheduleType>
   | DeletedEvent
   | RequestedEvent
   | InvokedEvent
   | RescheduledEvent
   | AcknowledgedEvent;
 
-interface ScheduledEvent {
+interface ScheduledEvent<ScheduleType extends string> {
   type: "scheduled";
-  job: Job;
+  job: Job<ScheduleType>;
 }
 
 interface InvokedEvent {
@@ -81,35 +83,34 @@ interface AcknowledgedEvent {
 }
 
 export class Activity<ScheduleType extends string> implements Closable {
-  private redis;
-  private producer;
-
+  private readonly pattern: string;
   constructor(
-    redisFactory: () => Redis,
-    private readonly onEvent: OnActivity,
+    private readonly redis: Redis,
+    private readonly onEvent: OnActivity<ScheduleType>,
     options: SubscriptionOptions = {}
   ) {
-    this.redis = redisFactory();
-    this.producer = new Producer<ScheduleType>(redisFactory, null as any);
-
     this.redis.on("pmessage", (_pattern, channel, message) =>
       this.handleMessage(channel, message)
     );
 
-    if (options.queue) {
-      options.queue = encodeRedisKey(options.queue);
-    }
+    const queue = options.queue ? encodeRedisKey(options.queue) : "*";
+    const id = options.id ? encodeRedisKey(options.id) : "*";
+    this.pattern = `${queue}:${id}`;
 
-    if (options.id) {
-      options.id = encodeRedisKey(options.id);
-    }
+    this.redis.psubscribe(this.pattern);
+  }
 
-    this.redis.psubscribe(`${options.queue ?? "*"}:${options.id ?? "*"}`);
+  private matchesPattern(channel: string) {
+    return minimatch(channel, this.pattern);
   }
 
   private async handleMessage(channel: string, message: string) {
+    if (!this.matchesPattern(channel)) {
+      return;
+    }
+
     const [_type, ...args] = splitEvent(message, 9);
-    const type = _type as OnActivityEvent["type"];
+    const type = _type as OnActivityEvent<ScheduleType>["type"];
 
     const channelParts = channel.split(":").map(decodeRedisKey);
     if (channelParts.length !== 2) {
@@ -141,7 +142,7 @@ export class Activity<ScheduleType extends string> implements Closable {
           retry: JSON.parse(retryJson),
           schedule: schedule_type
             ? {
-                type: schedule_type,
+                type: schedule_type as ScheduleType,
                 meta: schedule_meta,
                 times: max_times ? Number(max_times) : undefined,
               }
@@ -166,7 +167,6 @@ export class Activity<ScheduleType extends string> implements Closable {
   }
 
   async close() {
-    await this.redis.quit();
-    await this.producer.close();
+    await this.redis.punsubscribe(this.pattern);
   }
 }

src/activity/activity.ts

@@ -61,15 +61,16 @@ export default class Owl<ScheduleType extends string> {
     );
   }
 
+  private activityRedis?: Redis;
+
   public createActivity(
-    onEvent: OnActivity,
+    onEvent: OnActivity<ScheduleType>,
     options: SubscriptionOptions = {}
   ) {
-    return new Activity<ScheduleType>(
-      this.redisFactory,
-      onEvent,
-      options
-    );
+    if (!this.activityRedis) {
+      this.activityRedis = this.redisFactory();
+    }
+    return new Activity(this.activityRedis, onEvent, options);
   }
 
   public async runMigrations() {

src/index.ts

@@ -19,6 +19,16 @@ describeAcrossBackends("Activity", (backend) => {
   }
 
   it("publishes all relevant information", async () => {
+    const secondActivityEvents = [];
+    const secondActivity = env.owl.createActivity(
+      (event) => {
+        secondActivityEvents.push(event);
+      },
+      {
+        id: "nonexistent",
+        queue: "nada",
+      }
+    );
     const currentDate = new Date();
 
     await env.producer.enqueue({
@@ -139,5 +149,9 @@ describeAcrossBackends("Activity", (backend) => {
       ],
       200
     );
+
+    expect(secondActivityEvents).to.be.empty;
+
+    await secondActivity.close();
   });
 });

test/functional/activity.test.ts

@@ -157,7 +157,7 @@ export function makeActivityEnv(backend: Backend, fail?: WorkerFailPredicate) {
 
   const activityEnv: typeof workerEnv & {
     activity: Activity<"every">;
-    events: OnActivityEvent[];
+    events: OnActivityEvent<"every">[];
   } = workerEnv as any;
 
   activityEnv.activity = null as any;