code: @Override
      public void call(Subscriber<? super Job<K, V>> subscriber) {
  
          log.debug("onSubscribe()");
          if (subscriber.isUnsubscribed()) {
              return;
          }
  
          String subscriberId = getClass().getSimpleName() + "-" + id + "-" + subscriberIds.incrementAndGet();
          subscriber.onStart();
  
          try {
              Scheduler.Worker worker = scheduler.createWorker();
  
              GetJobsAction<K, V> getJobsAction = new GetJobsAction<K, V>(disqueConnectionSupplier, subscriberId, subscriber,
                      jobLocalityTracking, getJobsArgs);
  
              actions.add(getJobsAction);
              Subscription subscription = worker.schedulePeriodically(getJobsAction, 0, 10, TimeUnit.MILLISECONDS);
              getJobsAction.setSelfSubscription(subscription);
  
              if (improveLocalityTimeUnit != null && improveLocalityInterval > 0 && reconnectTrigger == null) {
                  reconnectTrigger = worker.schedulePeriodically(new Action0() {
                      @Override
                      public void call() {
                          switchNodes();
                      }
                  }, improveLocalityInterval, improveLocalityInterval, improveLocalityTimeUnit);
              }
          } catch (Exception e) {
              if (log.isDebugEnabled()) {
                  log.debug("QueueListener.call caught an exception: {}", e.getMessage(), e);
              }
              subscriber.onError(e);
          }
      }
docs: Setup subscriptions when the Observable subscription is set up.
  
  @param subscriber the subscriber