docs: @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);
      }