Introduce FluxFlatMapUsageCheck

error-prone-contrib/src/main/java/tech/picnic/errorprone/bugpatterns/FluxFlatMapUsageCheck.java

@@ -0,0 +1,86 @@
+package tech.picnic.errorprone.bugpatterns;
+
+import static com.google.errorprone.matchers.method.MethodMatchers.instanceMethod;
+
+import com.google.auto.service.AutoService;
+import com.google.common.collect.Iterables;
+import com.google.errorprone.BugPattern;
+import com.google.errorprone.BugPattern.LinkType;
+import com.google.errorprone.BugPattern.SeverityLevel;
+import com.google.errorprone.BugPattern.StandardTags;
+import com.google.errorprone.VisitorState;
+import com.google.errorprone.bugpatterns.BugChecker;
+import com.google.errorprone.bugpatterns.BugChecker.MemberReferenceTreeMatcher;
+import com.google.errorprone.bugpatterns.BugChecker.MethodInvocationTreeMatcher;
+import com.google.errorprone.fixes.SuggestedFix;
+import com.google.errorprone.fixes.SuggestedFixes;
+import com.google.errorprone.matchers.Description;
+import com.google.errorprone.matchers.Matcher;
+import com.sun.source.tree.ExpressionTree;
+import com.sun.source.tree.MemberReferenceTree;
+import com.sun.source.tree.MethodInvocationTree;
+import java.util.function.Function;
+import java.util.function.Supplier;
+import reactor.core.publisher.Flux;
+
+/**
+ * A {@link BugChecker} which flags usages of {@link Flux#flatMap(Function)} and {@link
+ * Flux#flatMapSequential(Function)}.
+ *
+ * <p>{@link Flux#flatMap(Function)} and {@link Flux#flatMapSequential(Function)} eagerly perform up
+ * to {@link reactor.util.concurrent.Queues#SMALL_BUFFER_SIZE} subscriptions. Additionally, the
+ * former interleaves values as they are emitted, yielding nondeterministic results. In most cases
+ * {@link Flux#concatMap(Function)} should be preferred, as it produces consistent results and
+ * avoids potentially saturating the thread pool on which subscription happens. If {@code
+ * concatMap}'s single-subscription semantics are undesirable one should invoke a {@code flatMap} or
+ * {@code flatMapSequential} overload with an explicit concurrency level.
+ *
+ * <p>NB: The rarely-used overload {@link Flux#flatMap(Function, Function, Supplier)} is not flagged
+ * by this check because there is no clear alternative to point to.
+ */
+@AutoService(BugChecker.class)
+@BugPattern(
+    name = "FluxFlatMapUsage",
+    summary =
+        "`Flux#flatMap` and `Flux#flatMapSequential` have subtle semantics; "
+            + "please use `Flux#concatMap` or explicitly specify the desired amount of concurrency",
+    linkType = LinkType.NONE,
+    severity = SeverityLevel.ERROR,
+    tags = StandardTags.LIKELY_ERROR)
+public final class FluxFlatMapUsageCheck extends BugChecker
+    implements MethodInvocationTreeMatcher, MemberReferenceTreeMatcher {
+  private static final long serialVersionUID = 1L;
+  private static final String MAX_CONCURRENCY_ARG_NAME = "MAX_CONCURRENCY";
+  private static final Matcher<ExpressionTree> FLUX_FLATMAP =
+      instanceMethod()
+          .onDescendantOf("reactor.core.publisher.Flux")
+          .namedAnyOf("flatMap", "flatMapSequential")
+          .withParameters(Function.class.getName());
+
+  @Override
+  public Description matchMethodInvocation(MethodInvocationTree tree, VisitorState state) {
+    if (!FLUX_FLATMAP.matches(tree, state)) {
+      return Description.NO_MATCH;
+    }
+
+    return buildDescription(tree)
+        .addFix(SuggestedFixes.renameMethodInvocation(tree, "concatMap", state))
+        .addFix(
+            SuggestedFix.builder()
+                .postfixWith(
+                    Iterables.getOnlyElement(tree.getArguments()), ", " + MAX_CONCURRENCY_ARG_NAME)
+                .build())
+        .build();
+  }
+
+  @Override
+  public Description matchMemberReference(MemberReferenceTree tree, VisitorState state) {
+    if (!FLUX_FLATMAP.matches(tree, state)) {
+      return Description.NO_MATCH;
+    }
+
+    // Method references are expected to occur very infrequently; generating both variants of
+    // suggested fixes is not worth the trouble.
+    return describeMatch(tree);
+  }
+}

error-prone-contrib/src/main/java/tech/picnic/errorprone/refastertemplates/ReactorTemplates.java

@@ -142,6 +142,35 @@ Flux<T> after(Flux<T> flux) {
     }
   }
 
+  /** Prefer {@link Flux#concatMap(Function)} over more contrived alternatives. */
+  static final class FluxConcatMap<T, S> {
+    @BeforeTemplate
+    Flux<S> before(Flux<T> flux, Function<? super T, ? extends Publisher<? extends S>> function) {
+      return Refaster.anyOf(flux.flatMap(function, 1), flux.flatMapSequential(function, 1));
+    }
+
+    @AfterTemplate
+    Flux<S> after(Flux<T> flux, Function<? super T, ? extends Publisher<? extends S>> function) {
+      return flux.concatMap(function);
+    }
+  }
+
+  /**
+   * Prefer {@link Flux#concatMapIterable(Function)} over {@link Flux#concatMapIterable(Function)},
+   * as the former has equivalent semantics but a clearer name.
+   */
+  static final class FluxConcatMapIterable<T, S> {
+    @BeforeTemplate
+    Flux<S> before(Flux<T> flux, Function<? super T, ? extends Iterable<? extends S>> function) {
+      return flux.flatMapIterable(function);
+    }
+
+    @AfterTemplate
+    Flux<S> after(Flux<T> flux, Function<? super T, ? extends Iterable<? extends S>> function) {
+      return flux.concatMapIterable(function);
+    }
+  }
+
   /**
    * Don't use {@link Mono#flatMapMany(Function)} to implicitly convert a {@link Mono} to a {@link
    * Flux}.

error-prone-contrib/src/test/java/tech/picnic/errorprone/bugpatterns/FluxFlatMapUsageCheckTest.java

@@ -0,0 +1,122 @@
+package tech.picnic.errorprone.bugpatterns;
+
+import static com.google.errorprone.BugCheckerRefactoringTestHelper.newInstance;
+
+import com.google.errorprone.BugCheckerRefactoringTestHelper;
+import com.google.errorprone.BugCheckerRefactoringTestHelper.FixChoosers;
+import com.google.errorprone.CompilationTestHelper;
+import org.junit.jupiter.api.Test;
+
+final class FluxFlatMapUsageCheckTest {
+  private final CompilationTestHelper compilationTestHelper =
+      CompilationTestHelper.newInstance(FluxFlatMapUsageCheck.class, getClass());
+  private final BugCheckerRefactoringTestHelper refactoringTestHelper =
+      newInstance(FluxFlatMapUsageCheck.class, getClass());
+
+  @Test
+  void identification() {
+    compilationTestHelper
+        .addSourceLines(
+            "A.java",
+            "import java.util.function.BiFunction;",
+            "import java.util.function.Function;",
+            "import reactor.core.publisher.Mono;",
+            "import reactor.core.publisher.Flux;",
+            "",
+            "class A {",
+            "  void m() {",
+            "    // BUG: Diagnostic contains:",
+            "    Flux.just(1).flatMap(Flux::just);",
+            "    // BUG: Diagnostic contains:",
+            "    Flux.just(1).<String>flatMap(i -> Flux.just(String.valueOf(i)));",
+            "    // BUG: Diagnostic contains:",
+            "    Flux.just(1).flatMapSequential(Flux::just);",
+            "    // BUG: Diagnostic contains:",
+            "    Flux.just(1).<String>flatMapSequential(i -> Flux.just(String.valueOf(i)));",
+            "",
+            "    Mono.just(1).flatMap(Mono::just);",
+            "    Flux.just(1).concatMap(Flux::just);",
+            "",
+            "    Flux.just(1).flatMap(Flux::just, 1);",
+            "    Flux.just(1).flatMap(Flux::just, 1, 1);",
+            "    Flux.just(1).flatMap(Flux::just, throwable -> Flux.empty(), Flux::empty);",
+            "",
+            "    Flux.just(1).flatMapSequential(Flux::just, 1);",
+            "    Flux.just(1).flatMapSequential(Flux::just, 1, 1);",
+            "",
+            "    // BUG: Diagnostic contains:",
+            "    this.<String, Flux<String>>sink(Flux::flatMap);",
+            "    // BUG: Diagnostic contains:",
+            "    this.<Integer, Flux<Integer>>sink(Flux::<Integer>flatMap);",
+            "",
+            "    // BUG: Diagnostic contains:",
+            "    this.<String, Flux<String>>sink(Flux::flatMapSequential);",
+            "    // BUG: Diagnostic contains:",
+            "    this.<Integer, Flux<Integer>>sink(Flux::<Integer>flatMapSequential);",
+            "",
+            "    this.<String, Mono<String>>sink(Mono::flatMap);",
+            "  }",
+            "",
+            "  private <T, P> void sink(BiFunction<P, Function<T, P>, P> fun) {}",
+            "}")
+        .doTest();
+  }
+
+  @Test
+  void replacementFirstSuggestedFix() {
+    refactoringTestHelper
+        .setFixChooser(FixChoosers.FIRST)
+        .addInputLines(
+            "in/A.java",
+            "import reactor.core.publisher.Flux;",
+            "",
+            "class A {",
+            "  void m() {",
+            "    Flux.just(1).flatMap(Flux::just);",
+            "    Flux.just(1).flatMapSequential(Flux::just);",
+            "  }",
+            "}")
+        .addOutputLines(
+            "out/A.java",
+            "import reactor.core.publisher.Flux;",
+            "",
+            "class A {",
+            "  void m() {",
+            "    Flux.just(1).concatMap(Flux::just);",
+            "    Flux.just(1).concatMap(Flux::just);",
+            "  }",
+            "}")
+        .doTest();
+  }
+
+  @Test
+  void replacementSecondSuggestedFix() {
+    refactoringTestHelper
+        .setFixChooser(FixChoosers.SECOND)
+        .addInputLines(
+            "in/A.java",
+            "import reactor.core.publisher.Flux;",
+            "",
+            "class A {",
+            "  private static final int MAX_CONCURRENCY = 8;",
+            "",
+            "  void m() {",
+            "    Flux.just(1).flatMap(Flux::just);",
+            "    Flux.just(1).flatMapSequential(Flux::just);",
+            "  }",
+            "}")
+        .addOutputLines(
+            "out/A.java",
+            "import reactor.core.publisher.Flux;",
+            "",
+            "class A {",
+            "  private static final int MAX_CONCURRENCY = 8;",
+            "",
+            "  void m() {",
+            "    Flux.just(1).flatMap(Flux::just, MAX_CONCURRENCY);",
+            "    Flux.just(1).flatMapSequential(Flux::just, MAX_CONCURRENCY);",
+            "  }",
+            "}")
+        .doTest();
+  }
+}

error-prone-contrib/src/test/resources/tech/picnic/errorprone/bugpatterns/ReactorTemplatesTestInput.java

@@ -1,5 +1,6 @@
 package tech.picnic.errorprone.bugpatterns;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import java.time.Duration;
 import java.util.Optional;
@@ -45,6 +46,15 @@ ImmutableSet<Flux<Integer>> testFluxSwitchIfEmptyOfEmptyPublisher() {
         Flux.just(1).switchIfEmpty(Mono.empty()), Flux.just(2).switchIfEmpty(Flux.empty()));
   }
 
+  ImmutableSet<Flux<Integer>> testFluxConcatMap() {
+    return ImmutableSet.of(
+        Flux.just(1).flatMap(Mono::just, 1), Flux.just(2).flatMapSequential(Mono::just, 1));
+  }
+
+  Flux<Integer> testFluxConcatMapIterable() {
+    return Flux.just(1, 2).flatMapIterable(ImmutableList::of);
+  }
+
   Flux<String> testMonoFlatMapToFlux() {
     return Mono.just("foo").flatMapMany(s -> Mono.just(s + s));
   }

error-prone-contrib/src/test/resources/tech/picnic/errorprone/bugpatterns/ReactorTemplatesTestOutput.java

@@ -2,6 +2,7 @@
 
 import static com.google.common.collect.MoreCollectors.toOptional;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import java.time.Duration;
 import java.util.Optional;
@@ -46,6 +47,14 @@ ImmutableSet<Flux<Integer>> testFluxSwitchIfEmptyOfEmptyPublisher() {
     return ImmutableSet.of(Flux.just(1), Flux.just(2));
   }
 
+  ImmutableSet<Flux<Integer>> testFluxConcatMap() {
+    return ImmutableSet.of(Flux.just(1).concatMap(Mono::just), Flux.just(2).concatMap(Mono::just));
+  }
+
+  Flux<Integer> testFluxConcatMapIterable() {
+    return Flux.just(1, 2).concatMapIterable(ImmutableList::of);
+  }
+
   Flux<String> testMonoFlatMapToFlux() {
     return Mono.just("foo").flatMap(s -> Mono.just(s + s)).flux();
   }