feat: pass redis connection as option

Author: Boris

src/connectRedis.ts example/src/connectRedis.tssrc/connectRedis.ts renamed to example/src/connectRedis.ts

@@ -1,6 +1,6 @@
 import { BULL_REDIS_URI, BULL_HOST_ID } from '../config';
 import { Queue, Worker, QueueScheduler } from 'bullmq';
-import { createBullConnection } from '../../../src/connectRedis';
+import { createBullConnection } from '../connectRedis';
 
 if (!BULL_REDIS_URI) {
   throw new Error(`Env var BULL_REDIS_URI is empty. Cannot init task ${__filename}.`);

example/src/demo_queues/fetchMetrics.ts

@@ -1,5 +1,6 @@
 import { composeBull } from '../../src';
 import { schemaComposer } from 'graphql-compose';
+import { createBullConnection } from './connectRedis';
 
 const { queryFields, mutationFields } = composeBull({
   schemaComposer,
@@ -9,6 +10,7 @@ const { queryFields, mutationFields } = composeBull({
     name: 'fetch_metrics',
     prefix: 'bull.demo',
   },
+  //redis: createBullConnection('queue'),
 });
 
 schemaComposer.Query.addFields({

example/src/schema.ts

@@ -10,7 +10,7 @@ export type Options = {
   };
   redis?:
     | {
-        uri: string;
+        uri?: string;
         opts?: RedisOptions;
       }
     | Redis.Redis;

src/definitions.ts

@@ -0,0 +1,22 @@
+import { Options } from './../definitions';
+import Redis from 'ioredis';
+
+let connection: Redis.Redis;
+
+export function getBullConnection(opts: Options): Redis.Redis {
+  if (connection) {
+    return connection;
+  }
+
+  if (opts.redis instanceof Redis) {
+    connection = opts.redis;
+  } else if (opts?.redis?.uri) {
+    connection = new Redis(opts.redis.uri);
+  } else if (opts?.redis?.opts) {
+    connection = new Redis(opts.redis.opts);
+  } else {
+    connection = new Redis();
+  }
+
+  return connection;
+}

src/helpers/getBullConnection.ts

@@ -1,16 +1,17 @@
 import { Title } from './queueTitles';
 import { Queue } from 'bullmq';
-import { createBullConnection } from '../connectRedis';
 import { MutationError, ErrorCodeEnum } from './MutationError';
+import { getBullConnection } from './getBullConnection';
+import { Options } from '../definitions';
 
-export function getQueues(titles: Array<Title>): Array<Queue> {
-  return titles.map((title) => getQueue(title.prefix, title.queueName));
+export function getQueues(titles: Array<Title>, opts: Options): Array<Queue> {
+  return titles.map((title) => getQueue(title.prefix, title.queueName, opts));
 }
 
-export function getQueue(prefix: string, queueName: string): Queue {
+export function getQueue(prefix: string, queueName: string, opts: Options): Queue {
   const queue = new Queue(queueName, {
     prefix,
-    connection: createBullConnection('queue'),
+    connection: getBullConnection(opts),
   });
 
   return queue;
@@ -19,9 +20,10 @@ export function getQueue(prefix: string, queueName: string): Queue {
 export async function findQueue(
   prefix: string,
   queueName: string,
+  opts: Options,
   checkExistence: boolean = true
 ): Promise<Queue> {
-  const connection = createBullConnection('custom');
+  const connection = getBullConnection(opts);
 
   if (checkExistence) {
     const queueExists = await connection.exists([prefix, queueName, 'meta'].join(':'));
@@ -33,7 +35,7 @@ export async function findQueue(
 
   const queue = new Queue(queueName, {
     prefix,
-    connection: createBullConnection('queue'),
+    connection,
   });
 
   return queue;

src/helpers/gettingQueues.ts

@@ -1,10 +1,11 @@
-import { createBullConnection } from '../connectRedis';
+import { Options } from './../definitions';
 import { normalizePrefixGlob } from './normalizePrefixGlob';
+import { getBullConnection } from './getBullConnection';
 
 export type Title = { prefix: string; queueName: string };
 
-export async function fetchQueueTitles(prefix: string): Promise<Array<Title>> {
-  const connection = createBullConnection('custom');
+export async function fetchQueueTitles(prefix: string, opts: Options): Promise<Array<Title>> {
+  const connection = getBullConnection(opts);
   const keys = await connection.keys(normalizePrefixGlob(prefix));
 
   return keys.map((key) => {

src/helpers/queueTitles.ts

@@ -39,7 +39,7 @@ export function createJobAddFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
-      const queue = await findQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {
         job,

src/mutation/jobAdd.ts

@@ -43,7 +43,7 @@ export function createJobAddBulkFC(
         .getTypePlural(),
     },
     resolve: async (_, { prefix, queueName, jobs }) => {
-      const queue = await findQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName, opts);
       const jobsRes = await queue.addBulk(jobs);
       return {
         jobs: jobsRes,

src/mutation/jobAddBulk.ts

@@ -48,7 +48,7 @@ export function createJobAddCronFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
-      const queue = await findQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {
         job,