WorkQueueProcessor extends EventLoopProcessor<E> { public static <E> WorkQueueProcessor<E> create() { return WorkQueueProcessor.<E>builder().build(); } @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 createDefault() { WorkQueueProcessor<Integer> processor = WorkQueueProcessor.create(); assertProcessor(processor, false, null, null, null, null, null, null); }
@Test public void createOverrideNameBufferSize() { String name = "nameOverride"; int bufferSize = 1024; WorkQueueProcessor<Integer> processor = WorkQueueProcessor.create(name, bufferSize); assertProcessor(processor, false, name, bufferSize, null, null, null, null); }
@Test public void scanInner() { WorkQueueProcessor<String> main = WorkQueueProcessor.create("name", 16); CoreSubscriber<String> subscriber = new SomeSubscriber(); WorkQueueProcessor.WorkQueueInner<String> test = new WorkQueueProcessor.WorkQueueInner<>(subscriber, main); Assertions.assertThat(test.scan(Scannable.Attr.PARENT)) .isSameAs(main); Assertions.assertThat(test.scan(Scannable.Attr.ACTUAL)) .isSameAs(subscriber); Assertions.assertThat(test.scan(Scannable.Attr.PREFETCH)) .isEqualTo(Integer.MAX_VALUE); Assertions.assertThat(test.scan(Scannable.Attr.REQUESTED_FROM_DOWNSTREAM)) .isEqualTo(0L); test.pendingRequest.set(123); Assertions.assertThat(test.scan(Scannable.Attr.REQUESTED_FROM_DOWNSTREAM)) .isEqualTo(123L); Assertions.assertThat(test.scan(Scannable.Attr.TERMINATED)) .isFalse(); Assertions.assertThat(test.scan(Scannable.Attr.CANCELLED)) .isFalse(); main.terminated = 1; Assertions.assertThat(test.scan(Scannable.Attr.TERMINATED)) .isTrue(); Assertions.assertThat(test.scan(Scannable.Attr.CANCELLED)) .isFalse(); test.cancel(); Assertions.assertThat(test.scan(Scannable.Attr.CANCELLED)) .isTrue(); }
@Test public void scanProcessor() { WorkQueueProcessor<String> test = WorkQueueProcessor.create("name", 16); Subscription subscription = Operators.emptySubscription(); test.onSubscribe(subscription); Assertions.assertThat(test.scan(Scannable.Attr.PARENT)) .isEqualTo(subscription); Assertions.assertThat(test.scan(Scannable.Attr.CAPACITY)) .isEqualTo(16); Assertions.assertThat(test.scan(Scannable.Attr.TERMINATED)) .isFalse(); Assertions.assertThat(test.scan(Scannable.Attr.ERROR)) .isNull(); test.onError(new IllegalStateException("boom")); Assertions.assertThat(test.scan(Scannable.Attr.ERROR)) .hasMessage("boom"); Assertions.assertThat(test.scan(Scannable.Attr.TERMINATED)) .isTrue(); }