TopicProcessor extends EventLoopProcessor<E> { @Override public Flux<E> drain() { return coldSource(ringBuffer, null, error, minimum); } 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 drainTest() throws Exception { final TopicProcessor<Integer> sink = TopicProcessor.<Integer>builder().name("topic") .build(); sink.onNext(1); sink.onNext(2); sink.onNext(3); StepVerifier.create(sink.forceShutdown()) .expectNext(1, 2, 3) .verifyComplete(); }