bassemZohdy

import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import java.util.function.Predicate;

public class RetryRunnable implements Runnable {
	private final long sleep;
	private final Runnable r;
	private final int times;
	private final Predicate<Throwable> p;
	private BiConsumer<Integer, Throwable> log;

bassemZohdy

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class ConsumerStream<T> implements Consumer<T> {

bassemZohdy

import java.util.function.Consumer;
import java.util.function.Predicate;

public interface Router<T> {

	Router<T> chain(Predicate<T> p, Consumer<T> c);

	Consumer<T> toConsumer();
}

bassemZohdy

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Predicate;

public class ConsumingRouter<T> implements Consumer<T> {
	private Map<Predicate<T>, Consumer<T>> map = new ConcurrentHashMap<>();

	public void sub(Predicate<T> p, Consumer<T> c) {
		map.put(p, c);

bassemZohdy

import java.util.Comparator;
import java.util.Iterator;
import java.util.Optional;
import java.util.Spliterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.BinaryOperator;
import java.util.function.Consumer;

bassemZohdy

import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;

import org.junit.Test;

public class ContConsumingTest {

bassemZohdy

import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class Processor<T> {
	final private Supplier<T> supplier;

bassemZohdy

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.function.Supplier;

public class ContConsuming<T> {
	final private Supplier<T> supplier;

bassemZohdy

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.function.Supplier;

public class RetrySupplyAsync<T> implements Supplier<CompletableFuture<T>> {
	private final Supplier<T> supplier;
	private final int tries;
	private final T defaultValue;
	private final long sleep;
	private final Class<? extends Throwable> throwableClass;

bassemZohdy

import java.util.function.Supplier;

public class BlockingSupplierWithRetry<T> implements Supplier<T> {
	private final Supplier<T> supplier;
	private final int tries;
	private final T defaultValue;
	private final long sleep;
	private final Class<? extends Throwable> throwableClass;

	private BlockingSupplierWithRetry(Supplier<T> supplier, int tries,