Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

3.x: Add missing annotations, fix many diamonds #6795

Merged
merged 3 commits into from
Dec 25, 2019
Merged

3.x: Add missing annotations, fix many diamonds #6795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6909f86
3.x: Add missing annotations, fix many diamonds
akarnokd Dec 24, 2019
8dee5c3
Fix accidental JDK 9 API usage
akarnokd Dec 24, 2019
995d6a6
Fix NPE in unicast due to null-cleanup
akarnokd Dec 24, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/main/java/io/reactivex/rxjava3/core/Completable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2335,7 +2335,7 @@ public final void subscribe(@NonNull CompletableObserver observer) {
* applied by {@link #subscribe(CompletableObserver)} before this method gets called.
* @param observer the CompletableObserver instance, never null
*/
protected abstract void subscribeActual(CompletableObserver observer);
protected abstract void subscribeActual(@NonNull CompletableObserver observer);

/**
* Subscribes a given CompletableObserver (subclass) to this Completable and returns the given
Expand Down
36 changes: 29 additions & 7 deletions src/main/java/io/reactivex/rxjava3/core/Flowable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -183,7 +183,7 @@ public abstract class Flowable<T> implements Publisher<T> {
@NonNull
@BackpressureSupport(BackpressureKind.PASS_THROUGH)
@SchedulerSupport(SchedulerSupport.NONE)
public static <T> Flowable<T> amb(Iterable<? extends Publisher<? extends T>> sources) {
public static <T> Flowable<T> amb(@NonNull Iterable<? extends Publisher<? extends T>> sources) {
Objects.requireNonNull(sources, "sources is null");
return RxJavaPlugins.onAssembly(new FlowableAmb<>(null, sources));
}
Expand Down Expand Up @@ -232,6 +232,7 @@ public static <T> Flowable<T> ambArray(Publisher<? extends T>... sources) {
* <em>before</em> the Flowable class is loaded.
* @return the default internal buffer size.
*/
@CheckReturnValue
public static int bufferSize() {
return BUFFER_SIZE;
}
Expand Down Expand Up @@ -673,6 +674,7 @@ public static <T1, T2, R> Flowable<R> combineLatest(
@NonNull BiFunction<? super T1, ? super T2, ? extends R> combiner) {
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -723,6 +725,7 @@ public static <T1, T2, T3, R> Flowable<R> combineLatest(
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -777,6 +780,7 @@ public static <T1, T2, T3, T4, R> Flowable<R> combineLatest(
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -836,6 +840,7 @@ public static <T1, T2, T3, T4, T5, R> Flowable<R> combineLatest(
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4, source5 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -899,6 +904,7 @@ public static <T1, T2, T3, T4, T5, T6, R> Flowable<R> combineLatest(
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4, source5, source6 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -967,6 +973,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, R> Flowable<R> combineLatest(
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4, source5, source6, source7 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -1039,6 +1046,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Flowable<R> combineLatest(
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(source8, "source8 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4, source5, source6, source7, source8 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -1116,6 +1124,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Flowable<R> combineLatest(
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(source8, "source8 is null");
Objects.requireNonNull(source9, "source9 is null");
Objects.requireNonNull(combiner, "combiner is null");
return combineLatestArray(new Publisher[] { source1, source2, source3, source4, source5, source6, source7, source8, source9 }, Functions.toFunction(combiner), bufferSize());
}

Expand Down Expand Up @@ -4735,6 +4744,7 @@ public static <T1, T2, R> Flowable<R> zip(
@NonNull BiFunction<? super T1, ? super T2, ? extends R> zipper) {
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2);
}

Expand Down Expand Up @@ -4796,6 +4806,7 @@ public static <T1, T2, R> Flowable<R> zip(
@NonNull BiFunction<? super T1, ? super T2, ? extends R> zipper, boolean delayError) {
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), delayError, bufferSize(), source1, source2);
}

Expand Down Expand Up @@ -4858,6 +4869,7 @@ public static <T1, T2, R> Flowable<R> zip(
@NonNull BiFunction<? super T1, ? super T2, ? extends R> zipper, boolean delayError, int bufferSize) {
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), delayError, bufferSize, source1, source2);
}

Expand Down Expand Up @@ -4923,6 +4935,7 @@ public static <T1, T2, T3, R> Flowable<R> zip(
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3);
}

Expand Down Expand Up @@ -4993,6 +5006,7 @@ public static <T1, T2, T3, T4, R> Flowable<R> zip(
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4);
}

Expand Down Expand Up @@ -5067,6 +5081,7 @@ public static <T1, T2, T3, T4, T5, R> Flowable<R> zip(
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4, source5);
}

Expand Down Expand Up @@ -5144,6 +5159,7 @@ public static <T1, T2, T3, T4, T5, T6, R> Flowable<R> zip(
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4, source5, source6);
}

Expand Down Expand Up @@ -5226,6 +5242,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, R> Flowable<R> zip(
Objects.requireNonNull(source5, "source5 is null");
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4, source5, source6, source7);
}

Expand Down Expand Up @@ -5312,6 +5329,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, R> Flowable<R> zip(
Objects.requireNonNull(source6, "source6 is null");
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(source8, "source8 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4, source5, source6, source7, source8);
}

Expand Down Expand Up @@ -5403,6 +5421,7 @@ public static <T1, T2, T3, T4, T5, T6, T7, T8, T9, R> Flowable<R> zip(
Objects.requireNonNull(source7, "source7 is null");
Objects.requireNonNull(source8, "source8 is null");
Objects.requireNonNull(source9, "source9 is null");
Objects.requireNonNull(zipper, "zipper is null");
return zipArray(Functions.toFunction(zipper), false, bufferSize(), source1, source2, source3, source4, source5, source6, source7, source8, source9);
}

Expand Down Expand Up @@ -6791,7 +6810,7 @@ public final <B> Flowable<List<T>> buffer(@NonNull Publisher<B> boundaryIndicato
@BackpressureSupport(BackpressureKind.ERROR)
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
public final <B> Flowable<List<T>> buffer(@NonNull Publisher<B> boundaryIndicator, final int initialCapacity) {
public final <B> Flowable<List<T>> buffer(@NonNull Publisher<B> boundaryIndicator, int initialCapacity) {
ObjectHelper.verifyPositive(initialCapacity, "initialCapacity");
return buffer(boundaryIndicator, Functions.<T>createArrayList(initialCapacity));
}
Expand Down Expand Up @@ -12899,7 +12918,7 @@ public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? ext
@NonNull
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? extends Publisher<R>> selector, final int bufferSize) {
public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? extends Publisher<R>> selector, int bufferSize) {
Objects.requireNonNull(selector, "selector is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return FlowableReplay.multicastSelector(FlowableInternalHelper.replaySupplier(this, bufferSize, false), selector);
Expand Down Expand Up @@ -12942,7 +12961,7 @@ public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? ext
@NonNull
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.NONE)
public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? extends Publisher<R>> selector, final int bufferSize, boolean eagerTruncate) {
public final <R> Flowable<R> replay(@NonNull Function<? super Flowable<T>, ? extends Publisher<R>> selector, int bufferSize, boolean eagerTruncate) {
Objects.requireNonNull(selector, "selector is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return FlowableReplay.multicastSelector(FlowableInternalHelper.replaySupplier(this, bufferSize, eagerTruncate), selector);
Expand Down Expand Up @@ -13551,7 +13570,7 @@ public final ConnectableFlowable<T> replay(long time, @NonNull TimeUnit unit, @N
@BackpressureSupport(BackpressureKind.FULL)
@SchedulerSupport(SchedulerSupport.CUSTOM)
@NonNull
public final ConnectableFlowable<T> replay(final long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean eagerTruncate) {
public final ConnectableFlowable<T> replay(long time, @NonNull TimeUnit unit, @NonNull Scheduler scheduler, boolean eagerTruncate) {
Objects.requireNonNull(unit, "unit is null");
Objects.requireNonNull(scheduler, "scheduler is null");
return FlowableReplay.create(this, time, unit, scheduler, eagerTruncate);
Expand Down Expand Up @@ -16930,7 +16949,7 @@ public final <U, V> Flowable<T> timeout(
@NonNull Publisher<U> firstTimeoutIndicator,
@NonNull Function<? super T, ? extends Publisher<V>> itemTimeoutIndicator,
@NonNull Publisher<? extends T> other) {
Objects.requireNonNull(firstTimeoutIndicator, "firstTimeoutSelector is null");
Objects.requireNonNull(firstTimeoutIndicator, "firstTimeoutIndicator is null");
Objects.requireNonNull(other, "other is null");
return timeout0(firstTimeoutIndicator, itemTimeoutIndicator, other);
}
Expand Down Expand Up @@ -17152,7 +17171,7 @@ public final Single<List<T>> toList() {
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
@SchedulerSupport(SchedulerSupport.NONE)
@NonNull
public final Single<List<T>> toList(final int capacityHint) {
public final Single<List<T>> toList(int capacityHint) {
ObjectHelper.verifyPositive(capacityHint, "capacityHint");
return RxJavaPlugins.onAssembly(new FlowableToListSingle<>(this, Functions.<T>createArrayList(capacityHint)));
}
Expand Down Expand Up @@ -18470,6 +18489,7 @@ public final <U, V> Flowable<Flowable<T>> window(
@NonNull Function3<? super T, ? super T1, ? super T2, R> combiner) {
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(combiner, "combiner is null");
Function<Object[], R> f = Functions.toFunction(combiner);
return withLatestFrom(new Publisher[] { source1, source2 }, f);
}
Expand Down Expand Up @@ -18513,6 +18533,7 @@ public final <U, V> Flowable<Flowable<T>> window(
Objects.requireNonNull(source1, "source1 is null");
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(combiner, "combiner is null");
Function<Object[], R> f = Functions.toFunction(combiner);
return withLatestFrom(new Publisher[] { source1, source2, source3 }, f);
}
Expand Down Expand Up @@ -18559,6 +18580,7 @@ public final <U, V> Flowable<Flowable<T>> window(
Objects.requireNonNull(source2, "source2 is null");
Objects.requireNonNull(source3, "source3 is null");
Objects.requireNonNull(source4, "source4 is null");
Objects.requireNonNull(combiner, "combiner is null");
Function<Object[], R> f = Functions.toFunction(combiner);
return withLatestFrom(new Publisher[] { source1, source2, source3, source4 }, f);
}
Expand Down
Loading