diff --git a/src/main/java/io/reactivex/rxjava4/core/Flowable.java b/src/main/java/io/reactivex/rxjava4/core/Flowable.java index 7b65f90cdf..99322df4e2 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Flowable.java @@ -12629,7 +12629,7 @@ public final Flowable 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 skip(long time, @NonNull TimeUnit unit) { return skipUntil(timer(time, unit)); @@ -12730,7 +12730,7 @@ public final Flowable 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 skipLast(long time, @NonNull TimeUnit unit) { return skipLast(time, unit, Schedulers.computation(), StandardBufferedConfig.DEFAULT); @@ -14035,7 +14035,7 @@ public final Flowable 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 takeLast(long count, long time, @NonNull TimeUnit unit) { return takeLast(count, time, unit, Schedulers.computation()); @@ -14860,7 +14860,7 @@ public final Flowable> 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> timeInterval(@NonNull TimeUnit unit) { return timeInterval(unit, Schedulers.computation()); @@ -14889,7 +14889,7 @@ public final Flowable> 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> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); @@ -15284,7 +15284,7 @@ public final Flowable> 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> timestamp(@NonNull TimeUnit unit) { return timestamp(unit, Schedulers.computation()); @@ -15314,7 +15314,7 @@ public final Flowable> 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> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); Objects.requireNonNull(scheduler, "scheduler is null"); diff --git a/src/main/java/io/reactivex/rxjava4/core/Observable.java b/src/main/java/io/reactivex/rxjava4/core/Observable.java index fabcda82bb..947a54696a 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Observable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Observable.java @@ -12304,7 +12304,7 @@ public final Observable> timeInterval(@NonNull Scheduler scheduler) { * @see ReactiveX operators documentation: TimeInterval */ @CheckReturnValue - @SchedulerSupport(SchedulerSupport.NONE) + @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps. @NonNull public final Observable> timeInterval(@NonNull TimeUnit unit) { return timeInterval(unit, Schedulers.computation()); @@ -12329,7 +12329,7 @@ public final Observable> timeInterval(@NonNull TimeUnit unit) { * @see ReactiveX operators documentation: TimeInterval */ @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> timeInterval(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); @@ -12674,7 +12674,7 @@ public final Observable> timestamp(@NonNull Scheduler scheduler) { * @see ReactiveX operators documentation: Timestamp */ @CheckReturnValue - @SchedulerSupport(SchedulerSupport.NONE) + @SchedulerSupport(SchedulerSupport.COMPUTATION) // Computation scheduler is only used for creating timestamps. @NonNull public final Observable> timestamp(@NonNull TimeUnit unit) { return timestamp(unit, Schedulers.computation()); @@ -12699,7 +12699,7 @@ public final Observable> timestamp(@NonNull TimeUnit unit) { * @see ReactiveX operators documentation: Timestamp */ @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> timestamp(@NonNull TimeUnit unit, @NonNull Scheduler scheduler) { Objects.requireNonNull(unit, "unit is null"); diff --git a/src/main/java/io/reactivex/rxjava4/core/Streamable.java b/src/main/java/io/reactivex/rxjava4/core/Streamable.java index 8b7073602a..89955ae678 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Streamable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Streamable.java @@ -397,6 +397,9 @@ static Streamable fromPublisher(@NonNull Flow.Publisher 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 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 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 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 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 timer(long delay, TimeUnit unit, Scheduler scheduler) { @@ -556,6 +563,7 @@ static Streamable 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 timer(long delay, TimeUnit unit, ExecutorService executor) { @@ -696,6 +704,7 @@ default Streamable collect(@NonNull Collector 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 delay(long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler) { @@ -1082,6 +1091,7 @@ default Streamable takeWhile(@NonNull Predicate 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 timeout(long timeout, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, @NonNull Streamable fallback) { diff --git a/src/test/java/io/reactivex/rxjava4/validators/CheckSchedulerAnnotationsTest.java b/src/test/java/io/reactivex/rxjava4/validators/CheckSchedulerAnnotationsTest.java new file mode 100644 index 0000000000..0216064ce5 --- /dev/null +++ b/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 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); + } +}