Reactor

Recipes to perform Reactor migration tasks.

Composite Recipes

Recipes that include further recipes, often including the individual recipes below.

• Migrate to Reactor 3.5
• Replace various Processor.cache calls with their Sinks equivalent
• Replace various Processor.create calls with their Sinks equivalent

Recipes

• Reactor Best Practices
• Replace DirectProcessor.create() with Sinks.many().multicast().directBestEffort()
• Replace EmitterProcessor.create() with Sinks.many().multicast().onBackpressureBuffer()
• Replace EmitterProcessor.create(Boolean) with Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, Boolean)
• Replace EmitterProcessor.create(int) with Sinks.many().multicast().onBackpressureBuffer(int)
• Replace EmitterProcessor.create(int, Boolean) with Sinks.many().multicast().onBackpressureBuffer(int, Boolean)
• Replace MonoProcessor.create() with Sinks.one()
• Replace ReplayProcessor.cacheLast() with Sinks.many().replay().latest()
• Replace ReplayProcessor.cacheLast() with Sinks.many().replay().latest()
• Replace ReplayProcessor.create() with Sinks.many().replay().all()
• Replace ReplayProcessor.create(int) with Sinks.many().replay().limit(int)
• Replace ReplayProcessor.create(int, false) with Sinks.many().replay().limit(int)
• Replace ReplayProcessor.create(int, true) with Sinks.many().replay().all(int)
• Replace ReplayProcessor.createSizeAndTimeout(int, Duration) with Sinks.many().replay().limit(int, duration)
• Replace ReplayProcessor.createSizeAndTimeout(int, Duration, Scheduler) with Sinks.many().replay().limit(int, Duration, Scheduler)
• Replace ReplayProcessor.createTimeout(Duration) with Sinks.many().replay().limit(duration)
• Replace ReplayProcessor.createTimeout(Duration, Scheduler) with Sinks.many().replay().limit(Duration, Scheduler)
• Replace UnicastProcessor.create() with Sinks.many().unicast().onBackpressureBuffer()
• Replace UnicastProcessor.create(Queue) with Sinks.many().unicast().onBackpressureBuffer(Queue)
• Replace UnicastProcessor.create(Queue, Consumer, Disposable) with Sinks.many().unicast().onBackpressureBuffer(Queue, Disposable)
• Replace UnicastProcessor.create(Queue, Disposable) with Sinks.many().unicast().onBackpressureBuffer(Queue, Disposable)
• Replace doAfterSuccessOrError calls with tap operator