WorkQueueProcessor extends EventLoopProcessor<E> { public final static <T> Builder<T> builder() { return new Builder<>(); } @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 createDefaultExecutorOverrideAll() { String name = "nameOverride"; int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().name(name) .bufferSize(bufferSize) .waitStrategy(waitStrategy) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, name, bufferSize, waitStrategy, autoCancel, null, null); }
@Test public void createOverrideAll() { ExecutorService executor = Executors.newSingleThreadExecutor(); ExecutorService requestTaskExecutor = Executors.newSingleThreadExecutor(); int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .requestTaskExecutor( requestTaskExecutor) .bufferSize(bufferSize) .waitStrategy(waitStrategy) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, bufferSize, waitStrategy, autoCancel, executor, requestTaskExecutor); }
@Test public void createOverrideAutoCancel() { boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, null, null, autoCancel, null, null); }
@Test public void createOverrideExecutor() { ExecutorService executor = Executors.newSingleThreadExecutor(); WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .build(); assertProcessor(processor, false, null, null, null, null, executor, null); }
@Test public void createOverrideExecutorAutoCancel() { ExecutorService executor = Executors.newSingleThreadExecutor(); boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, null, null, autoCancel, executor, null); }
@Test public void createOverrideExecutorBufferSize() { ExecutorService executor = Executors.newSingleThreadExecutor(); int bufferSize = 1024; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .bufferSize(bufferSize) .build(); assertProcessor(processor, false, null, bufferSize, null, null, executor, null); }
@Test public void createOverrideExecutorBufferSizeAutoCancel() { ExecutorService executor = Executors.newSingleThreadExecutor(); int bufferSize = 1024; boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .bufferSize(bufferSize) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, bufferSize, null, autoCancel, executor, null); }
@Test public void createOverrideExecutorBufferSizeWaitStrategy() { ExecutorService executor = Executors.newSingleThreadExecutor(); int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .bufferSize(bufferSize) .waitStrategy(waitStrategy) .build(); assertProcessor(processor, false, null, bufferSize, waitStrategy, null, executor, null); }
@Test public void createOverrideExecutorBufferSizeWaitStrategyAutoCancel() { ExecutorService executor = Executors.newSingleThreadExecutor(); int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().executor(executor) .bufferSize(bufferSize) .waitStrategy(waitStrategy) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, bufferSize, waitStrategy, autoCancel, executor, null); }
@Test public void createOverrideName() { String name = "nameOverride"; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().name(name) .build(); assertProcessor(processor, false, name, null, null, null, null, null); }
@Test public void createOverrideNameBufferSizeAutoCancel() { String name = "nameOverride"; int bufferSize = 1024; boolean autoCancel = false; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().name(name) .bufferSize(bufferSize) .autoCancel(autoCancel) .build(); assertProcessor(processor, false, name, bufferSize, null, autoCancel, null, null); }
@Test public void createOverrideNameBufferSizeWaitStrategy() { String name = "nameOverride"; int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); WorkQueueProcessor<Integer> processor = WorkQueueProcessor.<Integer>builder().name(name) .bufferSize(bufferSize) .waitStrategy(waitStrategy) .build(); assertProcessor(processor, false, name, bufferSize, waitStrategy, null, null, null); }
@Test(expected = IllegalArgumentException.class) public void failNegativeBufferSize() { WorkQueueProcessor.builder() .name("test") .bufferSize(-1); }
@Test(expected = IllegalArgumentException.class) public void failNonPowerOfTwo() { WorkQueueProcessor.builder() .name("test") .bufferSize(3); }
@Test(expected = IllegalArgumentException.class) public void failNullBufferSize() { WorkQueueProcessor.builder() .name("test") .bufferSize(0); }
@Test public void testWorkQueueProcessorGetters() { final int TEST_BUFFER_SIZE = 16; WorkQueueProcessor<Object> processor = WorkQueueProcessor.builder() .name("testProcessor") .bufferSize(TEST_BUFFER_SIZE) .build(); assertEquals(TEST_BUFFER_SIZE, processor.getAvailableCapacity()); processor.onNext(new Object()); assertEquals(TEST_BUFFER_SIZE - 1, processor.getAvailableCapacity()); processor.awaitAndShutdown(); }