Perform collection for stream collector.

@param stages stages to apply to collector
@param consumer consumer to apply
@param supplier supplier to provide result
@param <T> source type
@param <U> target type
191903
<T, U> Stage<U> doStreamCollect(
      final Collection<? extends Stage<? extends T>> stages, final Consumer<? super T> consumer,
      final Supplier<? extends U> supplier
  ) {
    final Completable<U> target = completable();

    final StreamCollectHelper<? super T, ? extends U> done =
        new StreamCollectHelper<>(caller, stages.size(), consumer, supplier, target);

    for (final Stage<? extends T> q : stages) {
      q.handle(done);
    }

    bindSignals(target, stages);
    return target;
  }
40243
void addConsumer(ConsumerSessionImpl consumer) {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "addConsumer", consumer);

        _consumerList.add(consumer);

        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "addConsumer");
    }
934558
public static <T> void forEach(final Iterable<T> iterable, final Consumer<? super T> consumer) {
        checkNotNull(iterable, "forEach must have a valid iterable");
        checkNotNull(consumer, "forEach must have a valid consumer");

        for (T t : iterable) {
            consumer.accept(t);
        }
    }
344486
private void processInitializedConduits(Source source) {
        String sourceId = CollectorManagerUtils.getSourceId(source);
        String sourceName = source.getSourceName();
        /*
         * Check that the current source initialized is Log/Message and Trace.
         * These are the unique special sources along with their respective Conduits/BufferManagers
         * that were started before OSGI and thusly were not configured via 'the osgi' way that CollectorManager
         * had in place before the JSON Logging work.
         */
        if (sourceName.equals(CollectorConstants.MESSAGES_SOURCE) || sourceName.equals(CollectorConstants.TRACE_SOURCE)) {
            //Make sure we have a bundleContext, we need this to play with osgi (i.e. register/listen to services)
            if (bundleContext == null) {
                retrieveBundleContext();
            }
            /*
             * We should really check if the source and conduit/bufferManager is already set into the
             * SourceManager and BufferManager maps. If so, no need to continue the rest of this method.
             */
            if (sourceMgrs.containsKey(sourceId) && bufferManagerMap.containsKey(sourceId)) {
                return;
            }

            ServiceReference<BufferManager>[] servRefs;
            try {
                servRefs = (ServiceReference<BufferManager>[]) bundleContext.getServiceReferences(BufferManager.class.getName(), null);
                if (servRefs != null) {
                    for (ServiceReference<BufferManager> servRef : servRefs) {
                        /*
                         * Ensure that the Conduit/BufferManager retrieved is the right one for the source
                         * by checking that the 'source' that the conduit/bufferManager was assinged to
                         * was for the Source that we are registering.
                         */
                        if (sourceName.equals(servRef.getProperty("source"))) {
                            /*
                             * Retrieve the actual Conduit/BufferManager fro mthe service Reference
                             * and put it into the bufferManagerMap.
                             */
                            Object object = bundleContext.getService(servRef);
                            BufferManager conduit = (BufferManager) object;
                            bufferManagerMap.put(sourceId, conduit);
                        }
                    }
                }

            } catch (InvalidSyntaxException e) {
                e.printStackTrace();
            }
        }
    }
383058
public static void forEachField(final List<Token> tokens, final BiConsumer<Token, Token> consumer)
    {
        for (int i = 0, size = tokens.size(); i < size;)
        {
            final Token fieldToken = tokens.get(i);
            if (fieldToken.signal() == Signal.BEGIN_FIELD)
            {
                final Token typeToken = tokens.get(i + 1);
                consumer.accept(fieldToken, typeToken);
                i += fieldToken.componentTokenCount();
            }
            else
            {
                ++i;
            }
        }
    }
543727
protected Supplier<CompletableFuture<ExecutionResult>> supplyAsync(
      Supplier<CompletableFuture<ExecutionResult>> supplier, Scheduler scheduler, FailsafeFuture<Object> future) {
    return () -> {
      ExecutionResult result = preExecute();
      if (result != null)
        return CompletableFuture.completedFuture(result);

      return supplier.get().thenCompose(s -> postExecuteAsync(s, scheduler, future));
    };
  }
512026
public static <E> Collection<E> createLazyCollectionFromStream(Supplier<Stream<E>> sequentialStreamSupplier, int size)
  {
    return new AbstractCollection<E>()
    {
      @Override
      public Iterator<E> iterator()
      {
        return sequentialStreamSupplier.get().iterator();
      }

      @Override
      public Spliterator<E> spliterator()
      {
        return sequentialStreamSupplier.get().spliterator();
      }

      @Override
      public Stream<E> stream()
      {
        return sequentialStreamSupplier.get();
      }

      @Override
      public Stream<E> parallelStream()
      {
        return sequentialStreamSupplier.get().parallel();
      }

      @Override
      public int size()
      {
        return size;
      }
    };
  }
1283014
public static <T> Consumer<T> pipeline(Consumer<T> consumer) {
        return new PipelinedConsumer<T>(Iterations.iterable(consumer));
    }
679730
@Override
   public void forEach(Consumer<? super List<String>> consumer) {
      try (Stream<List<String>> stream = stream()) {
         stream.forEach(consumer);
      }
   }
1184826
private static <T> void acceptResult(CompletableCompletionStage<T> s, Supplier<? extends T> supplier) {
        try {
            // exception can be thrown only by supplier. All callbacks are generated by us and they do not throw any exceptions
            s.complete(supplier.get());
        } catch (Throwable e) {
            handleFailure(s, e);
        }
    }
1262470
@SuppressWarnings({ "unchecked", "PMD.ShortVariable" })
    public <E extends EventBase<?>, C extends Channel> void forChannels(
            Class<C> type, BiConsumer<E, C> handler) {
        Arrays.stream(channels)
            .filter(c -> type.isAssignableFrom(c.getClass()))
            .forEach(c -> handler.accept((E) this, (C) c));
    }
1021090
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
            Executor executor) {
        return CompletableFuture
                .supplyAsync(handleSupplierWithException(supplier), executor);
    }
142281
@Override
    public StreamT<W,T> onEmptyGet(final Supplier<? extends T> supplier) {

        return (StreamT<W,T>) FoldableTransformerSeq.super.onEmptyGet(supplier);
    }
904863
@SuppressWarnings("unchecked")
    public void forEach(final EntryConsumer<V> consumer)
    {
        for (int i = 0, size = values.length; i < size; i++)
        {
            final Object value = values[i];
            if (null != value)
            {
                final long compoundKey = keys[i];
                final int keyPartA = (int)(compoundKey >>> 32);
                final int keyPartB = (int)(compoundKey & 0xFFFF_FFFFL);

                consumer.accept(keyPartA, keyPartB, (V)value);
            }
        }
    }
410798
@SuppressWarnings("rawtypes")
    public static <T, R> Collector<T, ?, R> streaming(final Supplier<? extends BlockingQueue<T>> queueSupplier,
            final Function<? super Stream<T>, ContinuableFuture<R>> streamingCollector, final long maxWaitIntervalInMillis) {
        final T NULL = (T) NONE;

        final Supplier<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>> supplier = new Supplier<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>>() {
            @Override
            public Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> get() {
                return Tuple.of((BlockingQueue<T>) queueSupplier.get(), MutableBoolean.of(true), MutableBoolean.of(false), new Holder<ContinuableFuture<R>>()); // _1 = queue, _2 = hasMore, _3 = isComplete, _4=future.
            }
        };

        final BiConsumer<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>, T> accumulator = new BiConsumer<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>, T>() {
            @Override
            public void accept(Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> xxx, T t) {
                if (xxx._4.value() == null) {
                    initStream(xxx, streamingCollector, maxWaitIntervalInMillis, NULL);
                }

                t = t == null ? NULL : t;

                if (xxx._3.isFalse() && xxx._1.offer(t) == false) {
                    try {
                        while (xxx._3.isFalse() && xxx._1.offer(t, maxWaitIntervalInMillis, TimeUnit.MILLISECONDS) == false) {
                        }
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        };

        final BinaryOperator<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>> combiner = new BinaryOperator<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>>() {
            @Override
            public Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> apply(
                    Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> t,
                    Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> u) {
                throw new UnsupportedOperationException("Should not happen");
            }
        };

        final Function<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>, R> finisher = new Function<Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>>, R>() {
            @Override
            public R apply(Tuple4<BlockingQueue<T>, MutableBoolean, MutableBoolean, Holder<ContinuableFuture<R>>> xxx) {
                if (xxx._4.value() == null) {
                    initStream(xxx, streamingCollector, maxWaitIntervalInMillis, NULL);
                }

                xxx._2.setFalse();

                try {
                    return xxx._4.value().get();
                } catch (InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            }
        };

        return new CollectorImpl(supplier, accumulator, combiner, finisher, CH_CONCURRENT_NOID);
    }
1026892
public static <T, A, R> Collector<T, ?, R> filter(Predicate<? super T> predicate, Collector<T, A, R> downstream) {
        return Collector.of(
            downstream.supplier(),
            (c, t) -> {
                if (predicate.test(t))
                    downstream.accumulator().accept(c, t);
            },
            downstream.combiner(),
            downstream.finisher()
        );
    }
54755
public void setCachedConsumer(CATMainConsumer consumer)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setCachedConsumer", consumer);
        this.cachedConsumer = consumer;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setCachedConsumer");
    }