Replace various Processor.create calls with their Sinks equivalent

Usage

This recipe has no required configuration options. You’ll need the Moderne CLI configured before running the command below.

mod run . --recipe org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes

If the recipe isn’t available locally, install it with:

mod config recipes jar install org.openrewrite.recipe:rewrite-reactive-streams:0.20.4

Definition

This recipe runs the following recipes in order.

• Replace MonoProcessor.create() with Sinks.one()org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$MonoProcessorCreateToSinkRecipe
• Replace ReplayProcessor.create() with Sinks.many().replay().all()org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateToSinkRecipe
• Replace ReplayProcessor.create(int) with Sinks.many().replay().limit(int)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateIntToSinkRecipe
• Replace ReplayProcessor.create(int, false) with Sinks.many().replay().limit(int)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateIntLiteralFalseToSinkRecipe
• Replace ReplayProcessor.create(int, true) with Sinks.many().replay().all(int)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateIntLiteralTrueToSinkRecipe
• Replace ReplayProcessor.createSizeAndTimeout(int, Duration) with Sinks.many().replay().limit(int, duration)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateSizeAndTimeoutToSinkRecipe
• Replace ReplayProcessor.createSizeAndTimeout(int, Duration, Scheduler) with Sinks.many().replay().limit(int, Duration, Scheduler)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateSizeAndTimeoutSchedulerToSinkRecipe
• Replace ReplayProcessor.createTimeout(Duration) with Sinks.many().replay().limit(duration)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateTimeoutToSinkRecipe
• Replace ReplayProcessor.createTimeout(Duration, Scheduler) with Sinks.many().replay().limit(Duration, Scheduler)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$ReplayProcessorCreateTimeoutSchedulerToSinkRecipe
• Replace DirectProcessor.create() with Sinks.many().multicast().directBestEffort()org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$DirectProcessorCreateToSinkRecipe
• Replace EmitterProcessor.create() with Sinks.many().multicast().onBackpressureBuffer()org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$EmitterProcessorCreateToSinkRecipe
• Replace EmitterProcessor.create(Boolean) with Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, Boolean)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$EmitterProcessorCreateBooleanToSinkRecipe
• Replace EmitterProcessor.create(int) with Sinks.many().multicast().onBackpressureBuffer(int)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$EmitterProcessorCreateIntToSinkRecipe
• Replace EmitterProcessor.create(int, Boolean) with Sinks.many().multicast().onBackpressureBuffer(int, Boolean)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$EmitterProcessorCreateIntBooleanToSinkRecipe
• Replace UnicastProcessor.create() with Sinks.many().unicast().onBackpressureBuffer()org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$UnicastProcessorCreateToSinkRecipe
• Replace UnicastProcessor.create(Queue) with Sinks.many().unicast().onBackpressureBuffer(Queue)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$UnicastProcessorCreateQueueToSinkRecipe
• Replace UnicastProcessor.create(Queue, Disposable) with Sinks.many().unicast().onBackpressureBuffer(Queue, Disposable)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$UnicastProcessorCreateQueueDisposableToSinkRecipe
• Replace UnicastProcessor.create(Queue, Consumer, Disposable) with Sinks.many().unicast().onBackpressureBuffer(Queue, Disposable)org.openrewrite.reactive.reactor.ReactorProcessorCreateToSinkRecipes$UnicastProcessorCreateQueueConsumerDisposableToSinkRecipe

Data tables

Structured output this recipe can produce.

• Source files that had resultsSource files that were modified by the recipe run.org.openrewrite.table.SourcesFileResults
• Source files that had search resultsSearch results that were found during the recipe run.org.openrewrite.table.SearchResults
• Source files that errored on a recipeThe details of all errors produced by a recipe run.org.openrewrite.table.SourcesFileErrors
• Recipe performanceStatistics used in analyzing the performance of recipes.org.openrewrite.table.RecipeRunStats