diff --git a/build.gradle b/build.gradle index da2ecd3ce4..40a7932571 100644 --- a/build.gradle +++ b/build.gradle @@ -65,6 +65,8 @@ java { tasks.withType(JavaCompile) { options.compilerArgs << "-parameters" + options.compilerArgs += '-Xlint:-module' + options.compilerArgs += '-Xlint:unchecked' } apply from: file("gradle/javadoc_cleanup.gradle") diff --git a/config/checkstyle/checkstyle.xml b/config/checkstyle/checkstyle.xml index 2ef4f9a54c..56eaf251d5 100644 --- a/config/checkstyle/checkstyle.xml +++ b/config/checkstyle/checkstyle.xml @@ -11,7 +11,7 @@ - + @@ -28,6 +28,6 @@ - + diff --git a/config/checkstyle/suppressions.xml b/config/checkstyle/suppressions.xml index 47f6dcef99..f4675fd64c 100644 --- a/config/checkstyle/suppressions.xml +++ b/config/checkstyle/suppressions.xml @@ -6,10 +6,10 @@ - - - - - + + + + + diff --git a/src/jmh/java/io/reactivex/rxjava4/core/BinaryFlatMapPerf.java b/src/jmh/java/io/reactivex/rxjava4/core/BinaryFlatMapPerf.java index 8bb16a2631..2c7480dada 100644 --- a/src/jmh/java/io/reactivex/rxjava4/core/BinaryFlatMapPerf.java +++ b/src/jmh/java/io/reactivex/rxjava4/core/BinaryFlatMapPerf.java @@ -22,7 +22,7 @@ import io.reactivex.rxjava4.functions.Function; -@SuppressWarnings("exports") +@SuppressWarnings("exports") @BenchmarkMode(Mode.Throughput) @Warmup(iterations = 5) @Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS) diff --git a/src/main/java/io/reactivex/rxjava4/core/Flowable.java b/src/main/java/io/reactivex/rxjava4/core/Flowable.java index 4de3f62d64..726dddd5e5 100644 --- a/src/main/java/io/reactivex/rxjava4/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava4/core/Flowable.java @@ -19,6 +19,7 @@ import java.util.stream.*; import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.docs.FlowableDocBasic; import io.reactivex.rxjava4.disposables.*; import io.reactivex.rxjava4.exceptions.*; import io.reactivex.rxjava4.flowables.*; @@ -153,7 +154,9 @@ * @see ParallelFlowable * @see io.reactivex.rxjava4.subscribers.DisposableSubscriber */ -public abstract class Flowable<@NonNull T> implements Publisher { +public abstract non-sealed class Flowable<@NonNull T> implements Publisher, +FlowableDocBasic +{ /** The default buffer size. */ static final int BUFFER_SIZE; static { @@ -10146,29 +10149,12 @@ public final Single elementAtOrError(long index) { return RxJavaPlugins.onAssembly(new FlowableElementAtSingle<>(this, index, null)); } - /** - * Filters items emitted by the current {@code Flowable} by only emitting those that satisfy a specified predicate. - *

- * - *

- *
Backpressure:
- *
The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure - * behavior.
- *
Scheduler:
- *
{@code filter} does not operate by default on a particular {@link Scheduler}.
- *
- * - * @param predicate - * a function that evaluates each item emitted by the current {@code Flowable}, returning {@code true} - * if it passes the filter - * @return the new {@code Flowable} instance - * @throws NullPointerException if {@code predicate} is {@code null} - * @see ReactiveX operators documentation: Filter - */ + /** {@inheritDoc} */ @CheckReturnValue @NonNull @BackpressureSupport(BackpressureKind.PASS_THROUGH) @SchedulerSupport(SchedulerSupport.NONE) + @Override public final Flowable filter(@NonNull Predicate predicate) { Objects.requireNonNull(predicate, "predicate is null"); return RxJavaPlugins.onAssembly(new FlowableFilter<>(this, predicate)); @@ -12038,31 +12024,12 @@ public final Single lastOrError() { return RxJavaPlugins.onAssembly(new FlowableLift<>(this, lifter)); } - /** - * Returns a {@code Flowable} that applies a specified function to each item emitted by the current {@code Flowable} and - * emits the results of these function applications. - *

- * - *

- *
Backpressure:
- *
The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure - * behavior.
- *
Scheduler:
- *
{@code map} does not operate by default on a particular {@link Scheduler}.
- *
- * - * @param the output type - * @param mapper - * a function to apply to each item emitted by the current {@code Flowable} - * @return the new {@code Flowable} instance - * @throws NullPointerException if {@code mapper} is {@code null} - * @see ReactiveX operators documentation: Map - * @see #mapOptional(Function) - */ + /** {@inheritDoc} */ @CheckReturnValue @NonNull @BackpressureSupport(BackpressureKind.PASS_THROUGH) @SchedulerSupport(SchedulerSupport.NONE) + @Override public final <@NonNull R> Flowable map(@NonNull Function mapper) { Objects.requireNonNull(mapper, "mapper is null"); return RxJavaPlugins.onAssembly(new FlowableMap<>(this, mapper)); @@ -20896,7 +20863,7 @@ public final Stream blockingStream(int prefetch) { ObjectHelper.verifyPositive(prefetch, "prefetch"); return RxJavaPlugins.onAssembly(new FlowableFlatMapStream<>(this, mapper, prefetch)); } - + /** * Construct a {@code Flowable} and use the given {@code generator} * to generate items on demand while running on the given {@link ExecutorService}. @@ -20948,7 +20915,7 @@ public final Stream blockingStream(int prefetch) { } /** - * Returns a {@code Flowable} that turns an upstream item an upstream item into + * Returns a {@code Flowable} that turns an upstream item an upstream item into * zero or more downstream values by running on the given {@link ExecutorService}. *

*

diff --git a/src/main/java/io/reactivex/rxjava4/core/docs/FlowableDocBasic.java b/src/main/java/io/reactivex/rxjava4/core/docs/FlowableDocBasic.java new file mode 100644 index 0000000000..357307cddd --- /dev/null +++ b/src/main/java/io/reactivex/rxjava4/core/docs/FlowableDocBasic.java @@ -0,0 +1,76 @@ +/* + * 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.core.docs; + +import io.reactivex.rxjava4.annotations.*; +import io.reactivex.rxjava4.core.*; +import io.reactivex.rxjava4.functions.*; + +/** + * Documents a set of operators so that the main Flowable source file is not cluttered. + * @param the element type of the flow + */ +public sealed interface FlowableDocBasic permits Flowable { + /** + * Returns a {@code Flowable} that applies a specified function to each item emitted by the current {@code Flowable} and + * emits the results of these function applications. + *

+ * + *

+ *
Backpressure:
+ *
The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure + * behavior.
+ *
Scheduler:
+ *
{@code map} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param the output type + * @param mapper + * a function to apply to each item emitted by the current {@code Flowable} + * @return the new {@code Flowable} instance + * @throws NullPointerException if {@code mapper} is {@code null} + * @see ReactiveX operators documentation: Map + * @see #mapOptional(Function) + */ + @CheckReturnValue + @NonNull + @BackpressureSupport(BackpressureKind.PASS_THROUGH) + @SchedulerSupport(SchedulerSupport.NONE) + <@NonNull R> Flowable map(@NonNull Function mapper); + + /** + * Filters items emitted by the current {@code Flowable} by only emitting those that satisfy a specified predicate. + *

+ * + *

+ *
Backpressure:
+ *
The operator doesn't interfere with backpressure which is determined by the current {@code Flowable}'s backpressure + * behavior.
+ *
Scheduler:
+ *
{@code filter} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param predicate + * a function that evaluates each item emitted by the current {@code Flowable}, returning {@code true} + * if it passes the filter + * @return the new {@code Flowable} instance + * @throws NullPointerException if {@code predicate} is {@code null} + * @see ReactiveX operators documentation: Filter + */ + @CheckReturnValue + @NonNull + @BackpressureSupport(BackpressureKind.PASS_THROUGH) + @SchedulerSupport(SchedulerSupport.NONE) + Flowable filter(@NonNull Predicate predicate); + +} diff --git a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java index 4c598957bc..f15b0ba9ad 100644 --- a/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java +++ b/src/main/java/io/reactivex/rxjava4/internal/virtual/FlowableVirtualCreateExecutor.java @@ -48,7 +48,7 @@ protected void subscribeActual(Subscriber s) { executor.submit(parent); } - static final class ExecutorVirtualCreateSubscription extends AtomicLong + static final class ExecutorVirtualCreateSubscription extends AtomicLong implements Subscription, Callable, VirtualEmitter { private static final long serialVersionUID = -6959205135542203083L; diff --git a/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java b/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java index 26bf4ce19d..d20c6fc03f 100644 --- a/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java +++ b/src/main/java/io/reactivex/rxjava4/plugins/RxJavaPlugins.java @@ -16,8 +16,7 @@ import java.lang.Thread.UncaughtExceptionHandler; import java.util.Objects; import java.util.concurrent.*; - -import static java.util.concurrent.Flow.*; +import java.util.concurrent.Flow.Subscriber; import io.reactivex.rxjava4.annotations.*; import io.reactivex.rxjava4.core.*; @@ -115,7 +114,7 @@ public final class RxJavaPlugins { @SuppressWarnings("rawtypes") @Nullable - static volatile BiFunction onParallelSubscribe; + static volatile BiFunction[], @NonNull ? extends Subscriber<@NonNull ?>[]> onParallelSubscribe; @Nullable static volatile BooleanSupplier onBeforeBlocking; @@ -1008,12 +1007,12 @@ public static CompletableObserver onSubscribe(@NonNull Completable source, @NonN * @param subscribers the array of subscribers * @return the value returned by the hook */ - @SuppressWarnings({ "rawtypes" }) + @SuppressWarnings({ "unchecked" }) @NonNull - public static <@NonNull T> Subscriber[] onSubscribe(@NonNull ParallelFlowable source, @NonNull Subscriber[] subscribers) { - BiFunction f = onParallelSubscribe; + public static <@NonNull T> Subscriber[] onSubscribe(@NonNull ParallelFlowable source, @NonNull Subscriber[] subscribers) { + var f = onParallelSubscribe; if (f != null) { - return apply(f, source, subscribers); + return (@NonNull Subscriber<@NonNull ? super @NonNull T>[]) apply(f, source, subscribers); } return subscribers; } @@ -1161,7 +1160,7 @@ public static void setOnParallelAssembly(@Nullable Function handler) { + public static void setOnParallelSubscribe(@Nullable BiFunction[], @NonNull ? extends Subscriber<@NonNull ?>[]> handler) { if (lockdown) { throw new IllegalStateException("Plugins can't be changed anymore"); } @@ -1176,7 +1175,7 @@ public static void setOnParallelSubscribe(@Nullable BiFunction getOnParallelSubscribe() { + public static BiFunction[], @NonNull ? extends Subscriber<@NonNull ?>[]> getOnParallelSubscribe() { return onParallelSubscribe; } @@ -1355,7 +1354,7 @@ public static Scheduler createExecutorScheduler(@NonNull Executor executor, bool * @return the result of the function call */ @NonNull - static <@NonNull T, @NonNull U, @NonNull R> R apply(@NonNull BiFunction f, @NonNull T t, @NonNull U u) { + static <@NonNull T, @NonNull U, @NonNull R> R apply(@NonNull BiFunction f, @NonNull T t, @NonNull U u) { try { return f.apply(t, u); } catch (Throwable ex) { diff --git a/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java b/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java index d36fc9703f..a0464b8817 100644 --- a/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java +++ b/src/test/java/io/reactivex/rxjava4/plugins/RxJavaPluginsTest.java @@ -589,14 +589,12 @@ public void onComplete() { .assertComplete(); } - @SuppressWarnings("rawtypes") @Test public void parallelFlowableStart() { try { - RxJavaPlugins.setOnParallelSubscribe(new BiFunction() { - @Override - public Subscriber[] apply(ParallelFlowable f, final Subscriber[] t) { - return new Subscriber[] { new Subscriber() { + RxJavaPlugins.setOnParallelSubscribe((_, t) -> { + var result = new Subscriber[] { + new Subscriber() { @Override public void onSubscribe(Subscription s) { @@ -606,7 +604,7 @@ public void onSubscribe(Subscription s) { @SuppressWarnings("unchecked") @Override public void onNext(Object value) { - t[0].onNext((Integer)value - 9); + ((Subscriber)t[0]).onNext(((Integer)value - 9)); } @Override @@ -621,8 +619,9 @@ public void onComplete() { } }; + return result; } - }); + ); Flowable.range(10, 3) .parallel(1) diff --git a/src/test/java/io/reactivex/rxjava4/tck/BaseTck.java b/src/test/java/io/reactivex/rxjava4/tck/BaseTck.java index 54663705c7..ec9b64c045 100644 --- a/src/test/java/io/reactivex/rxjava4/tck/BaseTck.java +++ b/src/test/java/io/reactivex/rxjava4/tck/BaseTck.java @@ -62,8 +62,8 @@ public static void before() { @AfterClass public static void after() { service.shutdown(); - } - + } + /** * Creates an Iterable with the specified number of elements or an infinite one if * {@code elements >} {@link Integer#MAX_VALUE}. diff --git a/src/test/java/io/reactivex/rxjava4/validators/BaseTypeParser.java b/src/test/java/io/reactivex/rxjava4/validators/BaseTypeParser.java index e89bce698f..bb1290572c 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/BaseTypeParser.java +++ b/src/test/java/io/reactivex/rxjava4/validators/BaseTypeParser.java @@ -53,6 +53,9 @@ public static List parse(File f, String baseClassName) throws Exceptio StringBuilder b = JavadocForAnnotations.readFile(f); int baseIndex = b.indexOf("public abstract class " + baseClassName); + if (baseIndex < 0) { + baseIndex = b.indexOf("public abstract non-sealed class " + baseClassName); + } if (baseIndex < 0) { throw new AssertionError("Wrong base class file: " + baseClassName); diff --git a/src/test/java/io/reactivex/rxjava4/validators/JavadocForAnnotations.java b/src/test/java/io/reactivex/rxjava4/validators/JavadocForAnnotations.java index 073e1ce77b..e3fa7ea836 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/JavadocForAnnotations.java +++ b/src/test/java/io/reactivex/rxjava4/validators/JavadocForAnnotations.java @@ -90,14 +90,17 @@ static final void scanFor(StringBuilder sourceCode, String annotation, String in int k = sourceCode.indexOf(inDoc, j); if (k < 0 || k > idx) { - // when printed on the console, IDEs will create a clickable link to help navigate to the offending point - e.append("java.lang.RuntimeException: missing ").append(inDoc).append(" section\r\n") - ; - int lc = lineNumber(sourceCode, idx); - - e.append(" at io.reactivex.rxjava4.core.").append(baseClassName) - .append(" (").append(baseClassName).append(".java:") - .append(lc).append(")").append("\r\n\r\n"); + var subSource = sourceCode.substring(j, idx); + if (!subSource.startsWith("/** {@inheritDoc} */")) { + // when printed on the console, IDEs will create a clickable link to help navigate to the offending point + e.append("java.lang.RuntimeException: missing ").append(inDoc).append(" section\r\n") + ; + int lc = lineNumber(sourceCode, idx); + + e.append(" at io.reactivex.rxjava4.core.").append(baseClassName) + .append(" (").append(baseClassName).append(".java:") + .append(lc).append(")").append("\r\n\r\n"); + } } } diff --git a/src/test/java/io/reactivex/rxjava4/validators/JavadocWording.java b/src/test/java/io/reactivex/rxjava4/validators/JavadocWording.java index 3656003904..ca3b96b84d 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/JavadocWording.java +++ b/src/test/java/io/reactivex/rxjava4/validators/JavadocWording.java @@ -1108,7 +1108,14 @@ static void aOrAn(StringBuilder e, RxMethod m, String wrongPre, String word, Str int nn = javadoc2.indexOf(" ", jj + 2); int mm = javadoc2.indexOf("}", jj + 2); - javadoc2 = javadoc2.substring(0, jj) + javadoc2.substring(nn + 1, mm) + javadoc2.substring(mm + 1); + if (nn > mm) { + nn = mm - 1; + } + + var jd2 = javadoc2; + javadoc2 = jd2.substring(0, jj); + javadoc2 += jd2.substring(nn + 1, mm); + javadoc2 += jd2.substring(mm + 1); kk = mm + 1; } diff --git a/src/test/java/io/reactivex/rxjava4/validators/NoAnonymousInnerClassesTest.java b/src/test/java/io/reactivex/rxjava4/validators/NoAnonymousInnerClassesTest.java index be73f5a6bc..394214cbcd 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/NoAnonymousInnerClassesTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/NoAnonymousInnerClassesTest.java @@ -23,7 +23,7 @@ public class NoAnonymousInnerClassesTest { @Test public void verify() throws Exception { - URL u = NoAnonymousInnerClassesTest.class.getResource("/"); + URL u = NoAnonymousInnerClassesTest.class.getResource(""); File f = new File(u.toURI()); String fs = f.toString().toLowerCase().replace("\\", "/"); diff --git a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java index 4fc506eb18..bc4f3fea23 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java +++ b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationCheckerTest.java @@ -605,7 +605,7 @@ public void checkParallelFlowable() { defaultValues.put(Collector.class, Collectors.toList()); defaultValues.put(ExecutorService.class, Executors.newVirtualThreadPerTaskExecutor()); - + VirtualTransformer trs = (_, _) -> { }; defaultValues.put(VirtualTransformer.class, trs); diff --git a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationNaming.java b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationNaming.java index 1abe0beaf1..fc28663e92 100644 --- a/src/test/java/io/reactivex/rxjava4/validators/ParamValidationNaming.java +++ b/src/test/java/io/reactivex/rxjava4/validators/ParamValidationNaming.java @@ -249,6 +249,11 @@ static void processFile(Class clazz) throws Exception { boolean found = false; for (int k = midx - 1; k >= 0; k--) { String linek = lines.get(k).trim(); + if (linek.startsWith("/** {@inheritDoc} */")) { + found = true; + // the docs in in the sealed doc class, skip the check + break; + } if (linek.startsWith("/**")) { break; } @@ -366,6 +371,11 @@ static void processFile(Class clazz) throws Exception { boolean found = false; for (int k = i - 1; k >= 0; k--) { String linek = lines.get(k).trim(); + if (linek.startsWith("/** {@inheritDoc} */")) { + found = true; + break; + // the document is in the sealed doc interface + } if (linek.startsWith("/**")) { break; }