WorkQueueProcessor extends EventLoopProcessor<E> { static int bestEffortMaxSubscribers(ExecutorService executor) { int maxSubscribers = Integer.MIN_VALUE; if (executor instanceof ThreadPoolExecutor) { maxSubscribers = ((ThreadPoolExecutor) executor).getMaximumPoolSize(); } else if (executor instanceof ForkJoinPool) { maxSubscribers = ((ForkJoinPool) executor).getParallelism(); } return maxSubscribers; } @SuppressWarnings("unchecked")  WorkQueueProcessor(
			@Nullable ThreadFactory threadFactory,
			@Nullable ExecutorService executor,
			ExecutorService requestTaskExecutor,
			int bufferSize, WaitStrategy waitStrategy, boolean share,
	                                boolean autoCancel); final static Builder<T> builder(); static WorkQueueProcessor<E> create(); static WorkQueueProcessor<E> create(String name, int bufferSize); static WorkQueueProcessor<E> share(String name, int bufferSize); @Override void subscribe(final CoreSubscriber<? super E> actual); @Override Flux<E> drain(); @Override long getPending(); @Override void run();  }
@Test public void testBestEffortMaxSubscribers() { int expectedUnbounded = Integer.MAX_VALUE; int expectedUnknown = Integer.MIN_VALUE; ExecutorService executorService1 = Executors.newSingleThreadExecutor(); ScheduledExecutorService executorService2 = Executors.newSingleThreadScheduledExecutor(); ExecutorService executorService3 = Executors.newCachedThreadPool(); ExecutorService executorService4 = Executors.newFixedThreadPool(2); ScheduledExecutorService executorService5 = Executors.newScheduledThreadPool(3); ExecutorService executorService6 = Executors.newWorkStealingPool(4); ExecutorService executorService7 = Executors.unconfigurableExecutorService(executorService4); ExecutorService executorService8 = Executors.unconfigurableScheduledExecutorService(executorService5); int maxSub1 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService1); int maxSub2 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService2); int maxSub3 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService3); int maxSub4 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService4); int maxSub5 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService5); int maxSub6 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService6); int maxSub7 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService7); int maxSub8 = WorkQueueProcessor.bestEffortMaxSubscribers(executorService8); executorService1.shutdown(); executorService2.shutdown(); executorService3.shutdown(); executorService4.shutdown(); executorService5.shutdown(); executorService6.shutdown(); executorService7.shutdown(); executorService8.shutdown(); assertEquals("newSingleThreadExecutor", expectedUnknown, maxSub1); assertEquals("newSingleThreadScheduledExecutor", expectedUnknown, maxSub2); assertEquals("newCachedThreadPool", expectedUnbounded, maxSub3); assertEquals("newFixedThreadPool(2)", 2, maxSub4); assertEquals("newScheduledThreadPool(3)", expectedUnbounded, maxSub5); assertEquals("newWorkStealingPool(4)", 4, maxSub6); assertEquals("unconfigurableExecutorService", expectedUnknown, maxSub7); assertEquals("unconfigurableScheduledExecutorService", expectedUnknown, maxSub8); }