TopicProcessor extends EventLoopProcessor<E> { public static <E> TopicProcessor<E> create() { return TopicProcessor.<E>builder().build(); } TopicProcessor(
			@Nullable ThreadFactory threadFactory,
			@Nullable ExecutorService executor,
			ExecutorService requestTaskExecutor,
			int bufferSize,
			WaitStrategy waitStrategy,
			boolean shared,
			boolean autoCancel,
			@Nullable final Supplier<E> signalSupplier); static Builder<E> builder(); static TopicProcessor<E> create(); static TopicProcessor<E> create(String name, int bufferSize); static TopicProcessor<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() { TopicProcessor<Integer> processor = TopicProcessor.create(); assertProcessor(processor, false, null, null, null, null, null, null); }
@Test public void konamiCode() throws InterruptedException { final TopicProcessor<Integer> keyboardStream = TopicProcessor.create(); Mono<List<Boolean>> konamis = keyboardStream.skipWhile(key -> KeyEvent.VK_UP != key) .buffer(10, 1) .map(keys -> keys.size() == 10 && keys.get(0) == KeyEvent.VK_UP && keys.get( 1) == KeyEvent.VK_UP && keys.get(2) == KeyEvent.VK_DOWN && keys.get( 3) == KeyEvent.VK_DOWN && keys.get(4) == KeyEvent.VK_LEFT && keys.get( 5) == KeyEvent.VK_RIGHT && keys.get(6) == KeyEvent.VK_LEFT && keys.get( 7) == KeyEvent.VK_RIGHT && keys.get(8) == KeyEvent.VK_B && keys.get( 9) == KeyEvent.VK_A) .collectList(); keyboardStream.onNext(KeyEvent.VK_UP); keyboardStream.onNext(KeyEvent.VK_UP); keyboardStream.onNext(KeyEvent.VK_UP); keyboardStream.onNext(KeyEvent.VK_DOWN); keyboardStream.onNext(KeyEvent.VK_DOWN); keyboardStream.onNext(KeyEvent.VK_LEFT); keyboardStream.onNext(KeyEvent.VK_RIGHT); keyboardStream.onNext(KeyEvent.VK_LEFT); keyboardStream.onNext(KeyEvent.VK_RIGHT); keyboardStream.onNext(KeyEvent.VK_B); keyboardStream.onNext(KeyEvent.VK_A); keyboardStream.onNext(KeyEvent.VK_C); keyboardStream.onComplete(); List<Boolean> res = konamis.block(); Assert.assertTrue(res.size() == 12); Assert.assertFalse(res.get(0)); Assert.assertTrue(res.get(1)); Assert.assertFalse(res.get(2)); Assert.assertFalse(res.get(3)); Assert.assertFalse(res.get(4)); Assert.assertFalse(res.get(5)); Assert.assertFalse(res.get(6)); Assert.assertFalse(res.get(7)); Assert.assertFalse(res.get(8)); Assert.assertFalse(res.get(9)); Assert.assertFalse(res.get(10)); Assert.assertFalse(res.get(11)); }
@Test public void scanInner() { TopicProcessor<String> main = TopicProcessor.create("name", 16); RingBuffer.Sequence sequence = RingBuffer.newSequence(123); CoreSubscriber<String> activated = new SomeSubscriber(); TopicProcessor.TopicInner<String> test = new TopicProcessor.TopicInner<>(main, sequence, activated); assertThat(test.scan(Scannable.Attr.PARENT)).isSameAs(main); assertThat(test.scan(Scannable.Attr.ACTUAL)).isSameAs(activated); assertThat(test.scan(Scannable.Attr.PREFETCH)).isEqualTo(Integer.MAX_VALUE); assertThat(test.scan(Scannable.Attr.REQUESTED_FROM_DOWNSTREAM)).isEqualTo(123L); assertThat(test.scan(Scannable.Attr.TERMINATED)).isFalse(); assertThat(test.scan(Scannable.Attr.CANCELLED)).isFalse(); main.terminated = 1; assertThat(test.scan(Scannable.Attr.TERMINATED)).isTrue(); assertThat(test.scan(Scannable.Attr.CANCELLED)).isFalse(); test.cancel(); assertThat(test.scan(Scannable.Attr.CANCELLED)).isTrue(); }
@Test public void scanInnerBufferedLargeHasIntMinValue() { TopicProcessor<String> main = TopicProcessor.create("name", 16); RingBuffer.Sequence sequence = RingBuffer.newSequence(123); CoreSubscriber<String> sub = new BaseSubscriber<String>() { }; TopicProcessor.TopicInner<String> test = new TopicProcessor.TopicInner<>(main, sequence, sub); main.ringBuffer.getSequencer().cursor.set(Integer.MAX_VALUE + 5L); test.sequence.set(2L); assertThat(test.scan(Scannable.Attr.BUFFERED)).isEqualTo(Integer.MIN_VALUE); assertThat(test.scan(Scannable.Attr.LARGE_BUFFERED)).isEqualTo(Integer.MAX_VALUE + 3L); }
@Test public void scanInnerBufferedSmallHasIntRealValue() { TopicProcessor<String> main = TopicProcessor.create("name", 16); RingBuffer.Sequence sequence = RingBuffer.newSequence(123); CoreSubscriber<String> sub = new BaseSubscriber<String>() { }; TopicProcessor.TopicInner<String> test = new TopicProcessor.TopicInner<>(main, sequence, sub); main.ringBuffer.getSequencer().cursor.set(Integer.MAX_VALUE + 5L); test.sequence.set(6L); assertThat(test.scan(Scannable.Attr.BUFFERED)).isEqualTo(Integer.MAX_VALUE - 1); assertThat(test.scan(Scannable.Attr.LARGE_BUFFERED)).isEqualTo(Integer.MAX_VALUE - 1L); }
@Test public void scanProcessor() { TopicProcessor<String> test = TopicProcessor.create("name", 16); Subscription subscription = Operators.emptySubscription(); test.onSubscribe(subscription); assertThat(test.scan(Scannable.Attr.PARENT)).isEqualTo(subscription); assertThat(test.scan(Scannable.Attr.CAPACITY)).isEqualTo(16); assertThat(test.scan(Scannable.Attr.TERMINATED)).isFalse(); assertThat(test.scan(Scannable.Attr.ERROR)).isNull(); test.onError(new IllegalStateException("boom")); assertThat(test.scan(Scannable.Attr.ERROR)).hasMessage("boom"); assertThat(test.scan(Scannable.Attr.TERMINATED)).isTrue(); }