TopicProcessor extends EventLoopProcessor<E> { public static <E> Builder<E> builder() { return new Builder<>(); } 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 createDefaultExecutorOverrideAll() { String name = "nameOverride"; int bufferSize = 1024; WaitStrategy waitStrategy = WaitStrategy.busySpin(); boolean autoCancel = false; TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<Integer>builder().autoCancel(autoCancel) .build(); assertProcessor(processor, false, null, null, null, autoCancel, null, null); }
@Test public void createOverrideExecutor() { ExecutorService executor = Executors.newSingleThreadExecutor(); TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<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(); TopicProcessor<Integer> processor = TopicProcessor.<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; TopicProcessor<Integer> processor = TopicProcessor.<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"; TopicProcessor<Integer> processor = TopicProcessor.<Integer>builder().name(name) .build(); assertProcessor(processor, false, name, null, null, null, null, null); }
@Test public void createOverrideNameBufferSize() { String name = "nameOverride"; int bufferSize = 1024; TopicProcessor<Integer> processor = TopicProcessor.<Integer>builder().name(name) .bufferSize(bufferSize) .build(); assertProcessor(processor, false, name, bufferSize, null, null, null, null); }
@Test public void createOverrideNameBufferSizeAutoCancel() { String name = "nameOverride"; int bufferSize = 1024; boolean autoCancel = false; TopicProcessor<Integer> processor = TopicProcessor.<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(); TopicProcessor<Integer> processor = TopicProcessor.<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() { TopicProcessor.builder() .name("test") .bufferSize(-1); }
@Test(expected = IllegalArgumentException.class) public void failNonPowerOfTwo() { TopicProcessor.builder() .name("test") .bufferSize(3); }
@Test(expected = IllegalArgumentException.class) public void failNullBufferSize() { TopicProcessor.builder() .name("test") .bufferSize(0); }
@Test public void indexBugTest() throws InterruptedException { int numOfItems = 20; TopicProcessor<String> ring = TopicProcessor.<String>builder().name("test") .bufferSize(1024) .build(); Flux<String> stream2 = ring.zipWith(Mono.fromCallable(System::currentTimeMillis) .repeat(), (t1, t2) -> String.format("%s : %s", t1, t2)); Mono<List<String>> p = stream2.doOnNext(System.out::println) .buffer(numOfItems) .next(); for (int curr = 0; curr < numOfItems; curr++) { if (curr % 5 == 0 && curr % 3 == 0) { ring.onNext("FizBuz" + curr); } else if (curr % 3 == 0) { ring.onNext("Fiz" + curr); } else if (curr % 5 == 0) { ring.onNext("Buz" + curr); } else { ring.onNext(String.valueOf(curr)); } } Assert.assertTrue("Has not returned list", p.block(Duration.ofSeconds(5)) != null); }
@Test public void testShutdown() { for (int i = 0; i < 1000; i++) { TopicProcessor<?> dispatcher = TopicProcessor.<String>builder().name("rb-test-dispose") .bufferSize(16) .build(); dispatcher.awaitAndShutdown(); } }
@Test public void testTopicProcessorGetters() { final int TEST_BUFFER_SIZE = 16; TopicProcessor<Object> processor = TopicProcessor.builder() .name("testProcessor") .bufferSize(TEST_BUFFER_SIZE) .build(); assertEquals(TEST_BUFFER_SIZE, processor.getAvailableCapacity()); processor.awaitAndShutdown(); }