4.x: Add SchedulerSupport annotation validator

src/main/java/io/reactivex/rxjava4/core/Flowable.java

@@ -12629,7 +12629,7 @@ public final Flowable<T> skip(long count) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<T> skip(long time, @NonNull TimeUnit unit) {
         return skipUntil(timer(time, unit));
@@ -12730,7 +12730,7 @@ public final Flowable<T> skipLast(int count) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<T> skipLast(long time, @NonNull TimeUnit unit) {
         return skipLast(time, unit, Schedulers.computation(), StandardBufferedConfig.DEFAULT);
@@ -14035,7 +14035,7 @@ public final Flowable<T> takeLast(int count) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.FULL)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<T> takeLast(long count, long time, @NonNull TimeUnit unit) {
         return takeLast(count, time, unit, Schedulers.computation());
@@ -14860,7 +14860,7 @@ public final Flowable<Timed<T>> timeInterval(@NonNull Scheduler scheduler) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
         return timeInterval(unit, Schedulers.computation());
@@ -14889,7 +14889,7 @@ public final Flowable<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
-    @SchedulerSupport(SchedulerSupport.NONE) // Supplied scheduler is only used for creating timestamps.
+    @SchedulerSupport(SchedulerSupport.CUSTOM) // Supplied scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<Timed<T>> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");
@@ -15284,7 +15284,7 @@ public final Flowable<Timed<T>> timestamp(@NonNull Scheduler scheduler) {
      */
     @CheckReturnValue
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Flowable<Timed<T>> timestamp(@NonNull TimeUnit unit) {
         return timestamp(unit, Schedulers.computation());
@@ -15314,7 +15314,7 @@ public final Flowable<Timed<T>> timestamp(@NonNull TimeUnit unit) {
     @CheckReturnValue
     @NonNull
     @BackpressureSupport(BackpressureKind.PASS_THROUGH)
-    @SchedulerSupport(SchedulerSupport.NONE) // Supplied scheduler is only used for creating timestamps.
+    @SchedulerSupport(SchedulerSupport.CUSTOM) // Supplied scheduler is only used for creating timestamps.
     public final Flowable<Timed<T>> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");
         Objects.requireNonNull(scheduler, "scheduler is null");

src/main/java/io/reactivex/rxjava4/core/Observable.java

@@ -12304,7 +12304,7 @@ public final Observable<Timed<T>> timeInterval(@NonNull Scheduler scheduler) {
      * @see <a href="http://reactivex.io/documentation/operators/timeinterval.html">ReactiveX operators documentation: TimeInterval</a>
      */
     @CheckReturnValue
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Observable<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
         return timeInterval(unit, Schedulers.computation());
@@ -12329,7 +12329,7 @@ public final Observable<Timed<T>> timeInterval(@NonNull TimeUnit unit) {
      * @see <a href="http://reactivex.io/documentation/operators/timeinterval.html">ReactiveX operators documentation: TimeInterval</a>
      */
     @CheckReturnValue
-    @SchedulerSupport(SchedulerSupport.NONE) // Supplied scheduler is only used for creating timestamps.
+    @SchedulerSupport(SchedulerSupport.CUSTOM) // Supplied scheduler is only used for creating timestamps.
     @NonNull
     public final Observable<Timed<T>> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");
@@ -12674,7 +12674,7 @@ public final Observable<Timed<T>> timestamp(@NonNull Scheduler scheduler) {
      * @see <a href="http://reactivex.io/documentation/operators/timestamp.html">ReactiveX operators documentation: Timestamp</a>
      */
     @CheckReturnValue
-    @SchedulerSupport(SchedulerSupport.NONE)
+    @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps.
     @NonNull
     public final Observable<Timed<T>> timestamp(@NonNull TimeUnit unit) {
         return timestamp(unit, Schedulers.computation());
@@ -12699,7 +12699,7 @@ public final Observable<Timed<T>> timestamp(@NonNull TimeUnit unit) {
      * @see <a href="http://reactivex.io/documentation/operators/timestamp.html">ReactiveX operators documentation: Timestamp</a>
      */
     @CheckReturnValue
-    @SchedulerSupport(SchedulerSupport.NONE) // Supplied scheduler is only used for creating timestamps.
+    @SchedulerSupport(SchedulerSupport.CUSTOM) // Supplied scheduler is only used for creating timestamps.
     @NonNull
     public final Observable<Timed<T>> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");

src/main/java/io/reactivex/rxjava4/core/Streamable.java

@@ -397,6 +397,9 @@ static <T> Streamable<T> fromPublisher(@NonNull Flow.Publisher<T> source, @NonNu
      * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
      * @throws IllegalArgumentException if {@code count} is negative
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
+    @CheckReturnValue
+    @NonNull
     static Streamable<Long> intervalRange(long start, long count,
             long initialDelay, long period, TimeUnit unit, Scheduler scheduler) {
         Objects.requireNonNull(unit, "unit is null");
@@ -432,6 +435,9 @@ static Streamable<Long> intervalRange(long start, long count,
      * @throws NullPointerException if {@code unit} or {@code executor} is {@code null}
      * @throws IllegalArgumentException if {@code count} is negative
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
+    @CheckReturnValue
+    @NonNull
     static Streamable<Long> intervalRange(long start, long count,
             long initialDelay, long period, TimeUnit unit, ExecutorService executor) {
         Objects.requireNonNull(unit, "unit is null");
@@ -535,6 +541,7 @@ static Streamable<Long> rangeLong(long start, long count) {
      * @return the new {@code Streamable} instance
      * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
     @CheckReturnValue
     @NonNull
     static Streamable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
@@ -556,6 +563,7 @@ static Streamable<Long> timer(long delay, TimeUnit unit, Scheduler scheduler) {
      * @return the new {@code Streamable} instance
      * @throws NullPointerException if {@code unit} or {@code executor} is {@code null}
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
     @CheckReturnValue
     @NonNull
     static Streamable<Long> timer(long delay, TimeUnit unit, ExecutorService executor) {
@@ -696,6 +704,7 @@ default <A, R> Streamable<R> collect(@NonNull Collector<T, A, R> collector) {
      * @return the new {@code Streamable} instance
      * @throws NullPointerException if {@code unit} or {@code scheduler} is {@code null}
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
     @CheckReturnValue
     @NonNull
     default Streamable<T> delay(long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) {
@@ -1082,6 +1091,7 @@ default Streamable<T> takeWhile(@NonNull Predicate<? super T> predicate) {
      * @return the new {@code Streamable} instance
      * @throws NullPointerException if {@code unit} or {@code scheduler} or {@code fallback} is {@code null}
      */
+    @SchedulerSupport(SchedulerSupport.CUSTOM)
     @CheckReturnValue
     @NonNull
     default Streamable<T> timeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Streamable<T> fallback) {

src/test/java/io/reactivex/rxjava4/validators/CheckSchedulerAnnotationsTest.java

@@ -0,0 +1,193 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
+ * compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is
+ * distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
+ * the License for the specific language governing permissions and limitations under the License.
+ */
+
+package io.reactivex.rxjava4.validators;
+
+import java.lang.classfile.*;
+import java.lang.constant.MethodTypeDesc;
+import java.lang.reflect.Method;
+import java.util.*;
+import java.util.concurrent.*;
+
+import org.junit.jupiter.api.Test;
+
+import io.reactivex.rxjava4.annotations.SchedulerSupport;
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.core.Observable;
+
+public class CheckSchedulerAnnotationsTest extends RxJavaTest {
+
+    // Via Grok
+    static int firstLine(Method m) {
+        Class<?> c = m.getDeclaringClass();
+        byte[] bytes;
+        try (var in = c.getResourceAsStream("/" + c.getName().replace('.', '/') + ".class")) {
+            if (in == null) {
+                return -1;
+            }
+            bytes = in.readAllBytes();
+        } catch (Exception ex) {
+            return 0;
+        }
+
+        var cm = ClassFile.of().parse(bytes);
+        var desc = MethodTypeDesc.ofDescriptor(
+                java.lang.invoke.MethodType.methodType(m.getReturnType(), m.getParameterTypes())
+                        .descriptorString()).descriptorString(); // <- had to fix this because Grok made a type error
+
+        for (var mm : cm.methods()) {
+            if (mm.methodName().equalsString(m.getName())
+                    && mm.methodType().toString().equals(desc)) { // <- Grok mistake, need to compare via canonical strings
+                var mc = mm.code();
+                if (!mc.isEmpty()) {
+                    var lnc = mc.get().findAttribute(Attributes.lineNumberTable());
+                    if (!lnc.isEmpty()) {
+                        var lnn = lnc.get().lineNumbers();
+                        var x = lnn.stream().mapToInt(i -> i.lineNumber()).min().orElse(-2);
+                        return x;
+                    }
+                }
+            }
+        }
+        return -3;
+    }
+
+    /// Methods do not need a Scheduler because they effectively block on the current thread
+    static final Set<String> skipMethods = new HashSet<>(List.of(
+            "fromFuture",
+            "blockingAwait"
+    ));
+
+    void processClass(Class<?> theClass) {
+        var sb = new StringBuilder();
+
+        for (var method : theClass.getMethods()) {
+            if (skipMethods.contains(method.getName())) {
+                continue;
+            }
+
+            var ann = method.getAnnotation(SchedulerSupport.class);
+
+            var hasTimeUnit = Flowable.fromArray(method.getParameters()).any(p -> p.getType() == TimeUnit.class).blockingGet();
+            var hasScheduler = Flowable.fromArray(method.getParameters()).any(p -> p.getType() == Scheduler.class).blockingGet();
+            var hasExecutor = Flowable.fromArray(method.getParameters()).any(p ->
+                p.getType() == Scheduler.class
+                || p.getType() == Executor.class
+                || p.getType() == ExecutorService.class
+                || p.getType() == ScheduledExecutorService.class
+            ).blockingGet();
+
+            var lineNum = firstLine(method);
+
+            if (hasTimeUnit) {
+                if (hasScheduler || hasExecutor) {
+                    if (ann == null) {
+                        sb.append("java.lang.AssertionError: missing SchedulerSupport annotation: ")
+                        .append(method)
+                        .append("\r\n")
+                        .append(" at ")
+                        .append(theClass.getCanonicalName())
+                        .append(".")
+                        .append(method.getName())
+                        .append("(")
+                        .append(theClass.getSimpleName())
+                        .append(".java:")
+                        .append(lineNum)
+                        .append(")\r\n");
+                        ;
+                    } else {
+                        if (!ann.value().equals(SchedulerSupport.CUSTOM)) {
+                            sb.append("java.lang.AssertionError: SchedulerSupport annotation is not CUSTOM: ")
+                            .append(method)
+                            .append("\r\n")
+                            .append(" at ")
+                            .append(theClass.getCanonicalName())
+                            .append(".")
+                            .append(method.getName())
+                            .append("(")
+                            .append(theClass.getSimpleName())
+                            .append(".java:")
+                            .append(lineNum)
+                            .append(")\r\n");
+                        }
+                    }
+                } else {
+                    if (ann == null) {
+                        sb.append("java.lang.AssertionError: missing SchedulerSupport annotation: ")
+                        .append(method)
+                        .append("\r\n")
+                        .append(" at ")
+                        .append(theClass.getCanonicalName())
+                        .append(".")
+                        .append(method.getName())
+                        .append("(")
+                        .append(theClass.getSimpleName())
+                        .append(".java:")
+                        .append(lineNum)
+                        .append(")\r\n");
+                    } else {
+                        if (ann.value().equals(SchedulerSupport.NONE)) {
+                            sb.append("java.lang.AssertionError: SchedulerSupport annotation is NONE: ")
+                            .append(method)
+                            .append("\r\n")
+                            .append(" at ")
+                            .append(theClass.getCanonicalName())
+                            .append(".")
+                            .append(method.getName())
+                            .append("(")
+                            .append(theClass.getSimpleName())
+                            .append(".java:")
+                            .append(lineNum)
+                            .append(")\r\n");
+                        }
+                    }
+                }
+            }
+        }
+
+        if (sb.length() != 0) {
+            sb.insert(0, "\r\n");
+            throw new AssertionError(sb.toString());
+        }
+    }
+
+    @Test
+    public void checkObservable() {
+        processClass(Observable.class);
+    }
+
+    @Test
+    public void checkFlowable() {
+        processClass(Flowable.class);
+    }
+
+    @Test
+    public void checkSingle() {
+        processClass(Single.class);
+    }
+
+    @Test
+    public void checkMaybe() {
+        processClass(Maybe.class);
+    }
+
+    @Test
+    public void checkCompletable() {
+        processClass(Completable.class);
+    }
+
+    @Test
+    public void checkStreamable() {
+        processClass(Streamable.class);
+    }
+}