Iterate over wheel so all active timers can be consumed without expiring them.

@param consumer to call for each active timer.
162198
public void forEach(final TimerConsumer consumer)
    {
        long numTimersLeft = timerCount;

        for (int j = currentTick, end = currentTick + wheel.length; j <= end; j++)
        {
            final long[] array = wheel[j & wheelMask];

            for (int i = 0, length = array.length; i < length; i++)
            {
                final long deadline = array[i];

                if (deadline != NULL_TIMER)
                {
                    consumer.accept(deadline, timerIdForSlot(j & wheelMask, i));

                    if (--numTimersLeft == 0)
                    {
                        return;
                    }
                }
            }
        }
    }
810855
public JMProgressiveManager<T, R> registerCompletedConsumer(
			Consumer<JMProgressiveManager<T, R>> completedConsumer) {
		Optional.ofNullable(completedConsumerList)
				.orElseGet(() -> completedConsumerList = new ArrayList<>())
				.add(completedConsumer);
		return this;
	}
95303
public void forEach(final IntObjConsumer<String> consumer)
    {
        int counterId = 0;

        final AtomicBuffer metaDataBuffer = this.metaDataBuffer;
        for (int i = 0, capacity = metaDataBuffer.capacity(); i < capacity; i += METADATA_LENGTH)
        {
            final int recordStatus = metaDataBuffer.getIntVolatile(i);

            if (RECORD_ALLOCATED == recordStatus)
            {
                consumer.accept(counterId, labelValue(metaDataBuffer, i));
            }
            else if (RECORD_UNUSED == recordStatus)
            {
                break;
            }

            counterId++;
        }
    }
935428
void removeConsumerSession(ConsumerSessionImpl consumer)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "removeConsumerSession", consumer);

        synchronized (_consumers)
        {
            _consumers.remove(consumer);
        }

        _messageProcessor.removeConsumer(consumer);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "removeConsumerSession");
    }
1286145
@Override
    public void dereferenceConsumerManager()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "dereferenceConsumerManager");
        setOutputHandler(null);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "dereferenceConsumerManager");
    }
633437
public void forEach(final CounterConsumer consumer)
    {
        int counterId = 0;

        final AtomicBuffer metaDataBuffer = this.metaDataBuffer;
        final AtomicBuffer valuesBuffer = this.valuesBuffer;

        for (int i = 0, capacity = metaDataBuffer.capacity(); i < capacity; i += METADATA_LENGTH)
        {
            final int recordStatus = metaDataBuffer.getIntVolatile(i);

            if (RECORD_ALLOCATED == recordStatus)
            {
                consumer.accept(
                    valuesBuffer.getLongVolatile(counterOffset(counterId)), counterId, labelValue(metaDataBuffer, i));
            }
            else if (RECORD_UNUSED == recordStatus)
            {
                break;
            }

            counterId++;
        }
    }
639082
protected void attachBifurcatedConsumer(BifurcatedConsumerSessionImpl consumer)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "attachBifurcatedConsumer", consumer);

        // Create a bifurcated list if required
        if (_bifurcatedConsumers == null)
        {
            synchronized (this)
            {
                if (_bifurcatedConsumers == null)
                    _bifurcatedConsumers = new LinkedList<BifurcatedConsumerSessionImpl>();
            }
        }

        synchronized (_bifurcatedConsumers)
        {
            _bifurcatedConsumers.add(consumer);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "attachBifurcatedConsumer");
    }
63825
public <T> void use(final Class<T> type, final Consumer<T> consumer) {
        if (uiThreadRunner.isOnUiThread()) {
            try {
                graph.use(type, Inject.class, consumer);
            } catch (PokeException e) {
                throw new MvcGraphException(e.getMessage(), e);
            }
        } else {
            uiThreadRunner.post(new Runnable() {
                @Override
                public void run() {
                    try {
                        graph.use(type, Inject.class, consumer);
                    } catch (PokeException e) {
                        throw new MvcGraphException(e.getMessage(), e);
                    }
                }
            });
        }
    }
1015311
public void forEach(final IntIntConsumer consumer)
    {
        final int[] entries = this.entries;
        final int initialValue = this.initialValue;
        @DoNotSub final int length = entries.length;

        for (@DoNotSub int i = 0; i < length; i += 2)
        {
            if (entries[i + 1] != initialValue) // lgtm [java/index-out-of-bounds]
            {
                consumer.accept(entries[i], entries[i + 1]); // lgtm [java/index-out-of-bounds]
            }
        }
    }
335725
public void forEach(IntBiConsumer consumer)
    {
        int len = length();
        for (int ii=0;ii<len;ii++)
        {
            consumer.accept(ii, get(ii));
        }
    }
290961
@Override
  public void consume(SerializableConsumer<R> consumer) {
    checkNotNull(consumer, "consumer cannot be null");

    ConsumerStreamlet<R> consumerStreamlet = new ConsumerStreamlet<>(this, consumer);
    addChild(consumerStreamlet);
  }
540364
public void forEach(final Consumer<V> consumer) {
        for (Map.Entry<Long, V> entry : map.entrySet()) {
            consumer.accept(entry.getValue());
        }
    }
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);
        }
    }
501471
public void dispatchStoppableConsumerSessionStopped(ConsumerSessionProxy consumerSessionProxy)
   {
      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dispatchStoppableConsumerSessionStopped", consumerSessionProxy);

      //Create a new StoppableAsynchConsumerCallbackThread and dispatch it.
      final StoppableAsynchConsumerCallbackThread thread = new StoppableAsynchConsumerCallbackThread(consumerSessionProxy);
      dispatchThread(thread);

      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dispatchStoppableConsumerSessionStopped");
   }
767171
public static void loopBlocks(int start , int endExclusive , IntRangeConsumer consumer ) {
		final ForkJoinPool pool = BoofConcurrency.pool;
		int numThreads = pool.getParallelism();

		int range = endExclusive-start;
		if( range == 0 ) // nothing to do here!
			return;
		if( range < 0 )
			throw new IllegalArgumentException("end must be more than start. "+start+" -> "+endExclusive);

		// Did some experimentation here. Gave it more threads than were needed or exactly what was needed
		// exactly seemed to do better in the test cases
		int blockSize = Math.max(1,range/numThreads);

		try {
			pool.submit(new IntRangeTask(start,endExclusive,blockSize,consumer)).get();
		} catch (InterruptedException | ExecutionException e) {
			throw new RuntimeException(e);
		}
	}
966312
public final void wakeUpConsumers() throws JMSException
    {
        synchronized (consumersMap)
        {
            Iterator<AbstractMessageConsumer> allConsumers = consumersMap.values().iterator();
            while (allConsumers.hasNext())
            {
                AbstractMessageConsumer consumer = allConsumers.next();
                consumer.wakeUp();
            }
        }
    }
883484
public void forEach(final EntryConsumer<V> consumer) {
        for (Map.Entry<Long, V> entry : map.entrySet()) {
            Long compoundKey = entry.getKey();
            final int keyPartA = (int) (compoundKey >>> Integer.SIZE);
            final int keyPartB = (int) (compoundKey & LOWER_INT_MASK);
            consumer.accept(keyPartA, keyPartB, entry.getValue());
        }
    }