feat: move async initialization to constructors

src/classes/queue-base.ts

@@ -48,6 +48,12 @@ export class QueueBase extends EventEmitter {
       keys[key] = this.toKey(key);
     });
     this.keys = keys;
+
+    this.initializing = this.connection.init();
+
+    this.waitUntilReady()
+      .then(client => client.on('error', this.emit.bind(this)))
+      .catch(err => this.emit('error'));
   }
 
   toKey(type: string) {
@@ -59,7 +65,7 @@ export class QueueBase extends EventEmitter {
       this.initializing = this.connection.init();
     }
 
-    this.client = await this.initializing;
+    return (this.client = await this.initializing);
   }
 
   protected base64Name() {

src/classes/queue-events.ts

@@ -16,7 +16,8 @@ export class QueueEvents extends QueueBase {
       this.opts,
     );
 
-    this.consumeEvents();
+    // tslint:disable: no-floating-promises
+    this.consumeEvents().catch(err => this.emit('error'));
   }
 
   private async consumeEvents() {
@@ -63,9 +64,9 @@ export class QueueEvents extends QueueBase {
         }
       } catch (err) {
         if (err.message !== 'Connection is closed.') {
-          await delay(5000);
           throw err;
         }
+        await delay(5000);
       }
     }
   }

src/classes/queue-scheduler.ts

@@ -26,27 +26,25 @@ export class QueueScheduler extends QueueBase {
 
   constructor(protected name: string, opts: QueueSchedulerOptions = {}) {
     super(name, { maxStalledCount: 1, stalledInterval: 30000, ...opts });
+
+    // tslint:disable: no-floating-promises
+    this.run();
   }
 
-  async waitUntilReady() {
-    await super.waitUntilReady();
+  private async run() {
+    await this.waitUntilReady();
 
-    const [nextTimestamp, streamLastId] = await Scripts.updateDelaySet(
-      this,
-      Date.now(),
-    );
+    const key = this.keys.delay;
+    const opts = this.opts as QueueSchedulerOptions;
+    const delaySet = await Scripts.updateDelaySet(this, Date.now());
+
+    const [nextTimestamp] = delaySet;
+    let streamLastId = delaySet[1] || '0-0';
 
     if (nextTimestamp) {
       this.nextTimestamp = nextTimestamp;
     }
 
-    this.run(streamLastId);
-  }
-
-  private async run(streamLastId = '0-0') {
-    const key = this.keys.delay;
-    const opts = this.opts as QueueSchedulerOptions;
-
     while (!this.closing) {
       // Check if at least the min stalled check time has passed.
       await this.moveStalledJobsToWait();

src/classes/queue.ts

@@ -25,6 +25,7 @@ export class Queue extends QueueGetters {
 
     this.jobsOpts = get(opts, 'defaultJobOptions');
 
+    // tslint:disable: no-floating-promises
     this.waitUntilReady().then(() => {
       this.client.hset(
         this.keys.meta,

src/classes/redis-connection.ts

@@ -7,7 +7,7 @@ export class RedisConnection {
   static minimumVersion = '5.0.0';
   client: IORedis.Redis;
 
-  constructor(private opts?: ConnectionOptions) {
+  constructor(private opts?: ConnectionOptions | IORedis.Redis) {
     if (!(opts instanceof IORedis)) {
       this.opts = Object.assign(
         {

src/classes/worker.ts

@@ -59,6 +59,7 @@ export class Worker extends QueueBase {
 
     this.repeat = new Repeat(name, opts);
 
+    /* tslint:disable: no-floating-promises */
     this.run();
   }