common-utils/src/main/java/org/warp/commonutils/concurrency/executor/BoundedExecutorService.java

package org.warp.commonutils.concurrency.executor;

import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import org.jetbrains.annotations.Nullable;

public class BoundedExecutorService {

	private static final int MAX_BLOCKING_QUEUE_SIZE = 50000;

	private BoundedExecutorService() {

	}

	@Deprecated
	public static ExecutorService createUnbounded(
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus) {
		return create(0, corePoolSize, maxPoolSize, keepAliveTime, unit, Executors.defaultThreadFactory(), queueSizeStatus);
	}

	public static ExecutorService createUnbounded(
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			ThreadFactory threadFactory,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus) {
		return createCustom(0, corePoolSize, maxPoolSize, keepAliveTime, unit, threadFactory, Duration.ofDays(100000), queueSizeStatus, new LinkedBlockingQueue<>(MAX_BLOCKING_QUEUE_SIZE));
	}

	public static ExecutorService createUnbounded(
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			ThreadFactory threadFactory,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus,
			BlockingQueue<Runnable> queue) {
		return createCustom(0, corePoolSize, maxPoolSize, keepAliveTime, unit, threadFactory, Duration.ofDays(100000), queueSizeStatus, queue);
	}

	@Deprecated
	public static ExecutorService create(
			int maxQueueSize,
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus) {
		return create(maxQueueSize, corePoolSize, maxPoolSize, keepAliveTime, unit, Executors.defaultThreadFactory(), queueSizeStatus);
	}

	public static ExecutorService create(
			int maxQueueSize,
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			ThreadFactory threadFactory,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus) {
		return createCustom(maxQueueSize, corePoolSize, maxPoolSize, keepAliveTime, unit, threadFactory, Duration.ofDays(100000), queueSizeStatus, new LinkedBlockingQueue<>(MAX_BLOCKING_QUEUE_SIZE));
	}

	public static ExecutorService create(
			int maxQueueSize,
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			ThreadFactory threadFactory,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus,
			BlockingQueue<Runnable> queue) {
		return createCustom(maxQueueSize, corePoolSize, maxPoolSize, keepAliveTime, unit, threadFactory, Duration.ofDays(100000), queueSizeStatus, queue);
	}

	public static ExecutorService createCustom(
			int maxQueueSize,
			int corePoolSize,
			int maxPoolSize,
			long keepAliveTime,
			TimeUnit unit,
			ThreadFactory threadFactory,
			Duration queueItemTtl,
			@Nullable BiConsumer<Boolean, Integer> queueSizeStatus,
			BlockingQueue<Runnable> queue) {
		ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize,
				maxPoolSize,
				keepAliveTime,
				unit,
				queue,
				threadFactory
		);
		threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		return new BlockingOnFullQueueExecutorServiceDecorator(threadPoolExecutor,
				maxQueueSize,
				queueItemTtl,
				queue::size,
				queueSizeStatus
		);
	}
}