@Override
  public void run() {

    while (!this.closed) {
      // Grab the next available message source, if any
      final EventSource<T> nextSource = sources.poll();
      if (null != nextSource) {
        // Grab the next message from that source, if any
        final T message = nextSource.getNext();
        if (null != message) {
          // Add the source to the end of the queue again.
          sources.add(nextSource);
          // Send the message. Note that this may block depending on the underlying EventHandler.
          this.output.onNext(message);
        } else {
          // The message source has returned null as the next message. We drop the message source in that case.
          Logger.getLogger(Pull2Push.class.getName()).log(Level.INFO, "Droping message source {0} from the queue",
              nextSource.toString());
        }
      }
    }
  }
@ Override public void run ( ) { while ( ! this . closed ) { final EventSource < T > nextSource = sources . poll ( ) ; if ( null != nextSource ) { final T message = nextSource . getNext ( ) ; if ( null != message ) { sources . add ( nextSource ) ; this . output . onNext ( message ) ; } else { Logger . getLogger ( Pull2Push . class . getName ( ) ) . log ( Level . INFO , "Droping message source {0} from the queue" , nextSource . toString ( ) ) ; } } } }
Executes the message loop.