code: public static <T> FlowableTransformer<T, List<T>> buffer(final int maxSize,
              final Function<? super T, ? extends Long> duration, final TimeUnit unit, final Scheduler scheduler) {
  
          final BiPredicate<List<T>, MyOptional<T>> condition = new BiPredicate<List<T>, MyOptional<T>>() {
              @Override
              public boolean test(List<T> list, MyOptional<T> x) throws Exception {
                  return list.size() < maxSize && x.isPresent();
              }
          };
          Function<MyOptional<T>, Long> timeout = new Function<MyOptional<T>, Long>() {
              @Override
              public Long apply(MyOptional<T> t) throws Exception {
                  return duration.apply(t.get());
              }
          };
          final FlowableTransformer<MyOptional<T>, MyOptional<T>> insert = insert(timeout, unit,
                  Functions.constant(MyOptional.<T>empty()), scheduler);
  
          final FlowableTransformer<MyOptional<T>, List<T>> collectWhile = collectWhile( //
                  // factory
                  ListFactoryHolder.<T>factory(), //
                  // add function
                  MyOptional.<T>addIfPresent(), //
                  // condition
                  condition);
  
          return new FlowableTransformer<T, List<T>>() {
              @Override
              public Publisher<List<T>> apply(Flowable<T> source) {
  
                  return source //
                          .map(MyOptional.<T>of()) //
                          .compose(insert) //
                          .compose(collectWhile)
                          // need this filter because sometimes nothing gets added to list
                          .filter(MyOptional.<T>listHasElements()); //
              }
          };
      }
docs: Buffers the source {@link Flowable} into {@link List}s, emitting Lists when
  the size of a list reaches {@code maxSize} or if the elapsed time since last
  emission from the source reaches the given duration.
  
  @param maxSize
  max size of emitted lists
  @param duration
  function that based on the last emission calculates the elapsed
  time to be used before emitting a buffered list
  @param unit
  unit of {@code duration}
  @param scheduler
  scheduler to use to schedule emission of a buffer (as a list) if
  the time since last emission from the source reaches duration
  @param <T>
  type of the source stream items
  @return source with operator applied