Note: Generally it's much slower than other {@code Collectors}.

@param streamingCollector
@return
@see Stream#observe(BlockingQueue, Predicate, long)
@see Stream#asyncCall(Try.Function)
431490
@SuppressWarnings("rawtypes")
    public static <T, R> Collector<T, ?, R> streaming(final Function<? super Stream<T>, R> streamingCollector) {
        return streaming((Supplier) queueSupplier, streamingCollector);
    }
42328
public static boolean synchronousSync(CuratorFramework curator, Duration timeout) {
        try {
            // Curator sync() is always a background operation.  Use a latch to block until it finishes.
            final CountDownLatch latch = new CountDownLatch(1);
            curator.sync().inBackground(new BackgroundCallback() {
                @Override
                public void processResult(CuratorFramework curator, CuratorEvent event) throws Exception {
                    if (event.getType() == CuratorEventType.SYNC) {
                        latch.countDown();
                    }
                }
            }).forPath(curator.getNamespace().isEmpty() ? "/" : "");

            // Wait for sync to complete.
            return latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            return false;
        } catch (Exception e) {
            throw Throwables.propagate(e);
        }
    }
72243
@Override
	public final void coGroup(Iterable<IN1> first, Iterable<IN2> second, Collector<OUT> out) throws Exception {
		streamer.streamBufferWithGroups(first.iterator(), second.iterator(), out);
	}
1190616
public synchronized Iterator receiveListenerIterator()
	{
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "receiveListenerIterator");
		
      final LinkedList<ReceiveListener>linkedList = new LinkedList<ReceiveListener>();
		final Iterator iterator = table.values().iterator();

		while(iterator.hasNext())
		{
			final RequestIdTableEntry tableEntry = (RequestIdTableEntry)iterator.next();
         if (tableEntry.receiveListener != null) linkedList.add(tableEntry.receiveListener);
		}

      final Iterator result = linkedList.iterator();
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "receiveListenerIterator", result);   	
		return result;
	}
616678
public static <T> Stream<T> parallelConcatt(final Collection<? extends Iterator<? extends T>> c) {
        return parallelConcatt(c, DEFAULT_READING_THREAD_NUM);
    }
849519
public CompletableFuture<Pair<List<String>, String>> listStreams(final String scope, final String token, final int limit) {
        Exceptions.checkNotNullOrEmpty(scope, "scope");
        return streamStore.listStream(scope, token, limit, executor);
    }
991601
public static <T> Collector<T, ?, Seq<T>> minAll(Comparator<? super T> comparator) {
        return minAllBy(t -> t, comparator);
    }
1195135
@Override
  public void collect(SpanCollector collector) throws IOException {
    for (QueueItem item : queueSpans) {
      item.sequenceSpans.spans.collect(collector);
    }
  }
725381
public static <T> Stream<T> parallelConcat(final Collection<? extends Stream<? extends T>> c) {
        return parallelConcat(c, DEFAULT_READING_THREAD_NUM);
    }
1328008
public static <T> Collector<T, ?, Optional<T>> max(Comparator<? super T> comparator) {
        return maxBy(t -> t, comparator);
    }
1106744
public static <T> StreamObserver<T> asyncUnimplementedStreamingCall(
      MethodDescriptor<?, ?> methodDescriptor, StreamObserver<?> responseObserver) {
    // NB: For streaming call we want to do the same as for unary call. Fail-fast by setting error
    // on responseObserver and then return no-op observer.
    asyncUnimplementedUnaryCall(methodDescriptor, responseObserver);
    return new NoopStreamObserver<>();
  }
1070829
public static <T> ObservableTransformer<T,T> doOnCompleteCount(Consumer<Integer> onComplete) {
        return obs -> obs.lift(new OperatorEmissionCounter<>(new CountObserver(null,onComplete,null)));
    }
920314
@SuppressWarnings("rawtypes")
    public static <T> Collector<T, ?, Optional<T>> onlyOne() {
        final Supplier<Holder<Optional<T>>> supplier = (Supplier) onlyOne_supplier;
        final BiConsumer<Holder<Optional<T>>, T> accumulator = (BiConsumer) onlyOne_accumulator;
        final BinaryOperator<Holder<Optional<T>>> combiner = (BinaryOperator) onlyOne_combiner;
        final Function<Holder<Optional<T>>, Optional<T>> finisher = (Function) onlyOne_finisher;

        return new CollectorImpl<>(supplier, accumulator, combiner, finisher, CH_UNORDERED_NOID);
    }
844571
public static <T> Collector<T, List<T>> toList() {
    return new Collector<T, List<T>>() {
      @Override
      public List<T> collect(Stream<? extends T> stream) {
        return Lists.newArrayList(stream.iterator());
      }
    };
  }
53175
public static <T> Observable<Response<T>> from(@NotNull final ApolloQueryWatcher<T> watcher) {
    checkNotNull(watcher, "watcher == null");
    return Observable.create(new ObservableOnSubscribe<Response<T>>() {
      @Override public void subscribe(final ObservableEmitter<Response<T>> emitter) throws Exception {
        cancelOnObservableDisposed(emitter, watcher);

        watcher.enqueueAndWatch(new ApolloCall.Callback<T>() {
          @Override public void onResponse(@NotNull Response<T> response) {
            if (!emitter.isDisposed()) {
              emitter.onNext(response);
            }
          }

          @Override public void onFailure(@NotNull ApolloException e) {
            Exceptions.throwIfFatal(e);
            if (!emitter.isDisposed()) {
              emitter.onError(e);
            }
          }
        });
      }
    });
  }
1198210
@Override
    public <X> LazyCollectionX<X> fromStream(final ReactiveSeq<X> stream) {
        return ListX.fromIterable(stream.collect(Collectors.toList()));
    }
959189
@Override
    public java.util.concurrent.Future<GetShardIteratorResult> getShardIteratorAsync(String streamName, String shardId, String shardIteratorType,
            String startingSequenceNumber, com.amazonaws.handlers.AsyncHandler<GetShardIteratorRequest, GetShardIteratorResult> asyncHandler) {

        return getShardIteratorAsync(new GetShardIteratorRequest().withStreamName(streamName).withShardId(shardId).withShardIteratorType(shardIteratorType)
                .withStartingSequenceNumber(startingSequenceNumber), asyncHandler);
    }