code: public static <T> Function<Flux<T>, Flux<T>> timespan(Comparator<T> comparator, Duration timespan) {
          return source -> {
              Queue<Tuple2<Long, T>> accumulator = new PriorityQueue<>((o1, o2) -> comparator.compare(o1.getT2(), o2.getT2()));
  
              Object monitor = new Object();
  
              DirectProcessor<Void> d = DirectProcessor.create();
  
              Disposable disposable = source
                  .timestamp()
                  .subscribe(item -> {
                      synchronized (monitor) {
                          accumulator.add(item);
                      }
                  }, d::onError, d::onComplete);
  
              return Flux
                  .interval(timespan)
                  .takeUntilOther(d)
                  .flatMap(n -> getItems(accumulator, monitor, timespan), null, () -> getItems(accumulator, monitor, Duration.ZERO))
                  .doOnCancel(disposable::dispose);
          };
      }
docs: Sorts the elements of a {@link Flux} within a sliding time window.  This sorter should be used when element order may be scrambled, but that scrambling has a certain 'temporal locality' to it.
  This assumption means that sorting can be limited to elements that arrive temporally close to one another without risking a latecomer being sorted incorrectly.
  
  @param comparator a {@link Comparator} to use when sorting the elements within the window
  @param timespan   the duration of the 'temporal locality'
  @param <T>        The type of the elements to be sorted
  @return a {@link Flux} providing the sorted elements