Method connect

lib/queue.js:92–145

()

  }

  connect() {
    return new Promise((resolve, reject) => {
      try {
        if (this._isReady) return resolve(this._isReady);

        const getEventPromise = () => {
          if (this.settings.getEvents || this.settings.activateDelayedJobs) {
            return this.makeClient('eclient', true).then(() => {
              this.eclient.on('message', this._onMessage.bind(this));
              const channels = [];
              if (this.settings.getEvents) {
                channels.push(this.toKey('events'));
              }
              if (this.settings.activateDelayedJobs) {
                channels.push(this.toKey('earlierDelayed'));
              }
              return Promise.all(
                channels.map((channel) =>
                  helpers.callAsync((done) =>
                    this.eclient.subscribe(channel, done)
                  )
                )
              );
            });
          }

          return null;
        };

        const eventsPromise = getEventPromise();
        // Wait for Lua scripts and client connections to load. Also wait for
        // bclient and eclient/subscribe if they're needed.
        this._ready = Promise.all([
          // Make the clients
          this.makeClient('client', false),
          this.settings.isWorker ? this.makeClient('bclient', true) : null,
          eventsPromise,
        ])
          .then(() => {
            if (this.settings.ensureScripts) {
              return lua.buildCache(this.client);
            }
          })
          .then(() => {
            this._isReady = true;
            setImmediate(() => this.emit('ready'));
            resolve(this._isReady);
            return this;
          });
      } catch (err) {
        reject(err);
      }
    });
  }

  _onMessage(channel, message) {
    if (channel === this.toKey('earlierDelayed')) {