code: @Override
      public T next(long systemCurrentTimeMillis) {
          // drain all the waiting messages from the source (up to 10k)
          while (delayed.size() < 10000) {
              T event = source.next(systemCurrentTimeMillis);
              if (event == null) {
                  break;
              }
              transformAndQueue(event, systemCurrentTimeMillis);
          }
  
          return delayed.nextReady(systemCurrentTimeMillis);
      }
docs: Return the next event that is safe for delivery or null
  if there are no safe objects to deliver.
  
  Null response could mean no events, or could mean all events
  are scheduled for the future.
  
  @param systemCurrentTimeMillis The current time.