Skip to content

Commit 580b39c

Browse files
akarnokdakarnokd
authored
4.x: Streamable + takeUntil + groupBy + refactor + helpers (#8215)
* 4.x: Streamable ops + takeUntil + groupBy + move types * Complete groupBy, improve API overall * Fix Javadoc errors
1 parent 31cddef commit 580b39c

71 files changed

Lines changed: 1304 additions & 154 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎gradle/javadoc_cleanup.gradle‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,8 @@ task javadocCleanup(dependsOn: "javadoc") doLast {
66
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Maybe.html'))
77
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/Completable.html'))
88

9-
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/flowables/ConnectableFlowable.html'))
10-
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/observables/ConnectableObservable.html'))
9+
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableFlowable.html'))
10+
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/core/ConnectableObservable.html'))
1111

1212
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/subjects/ReplaySubject.html'))
1313
fixJavadocFile(rootProject.file('build/docs/javadoc/io/reactivex/rxjava4/processors/ReplayProcessor.html'))

‎src/jmh/java/io/reactivex/rxjava4/parallel/ParallelPerf.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222

2323
import io.reactivex.rxjava4.core.*;
2424
import io.reactivex.rxjava4.core.config.StandardConcurrentBufferedConfig;
25-
import io.reactivex.rxjava4.flowables.GroupedFlowable;
2625
import io.reactivex.rxjava4.functions.Function;
2726
import io.reactivex.rxjava4.schedulers.Schedulers;
2827

src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java renamed to src/main/java/io/reactivex/rxjava4/core/ConnectableFlowable.java

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -11,15 +11,13 @@
1111
* the License for the specific language governing permissions and limitations under the License.
1212
*/
1313

14-
package io.reactivex.rxjava4.flowables;
14+
package io.reactivex.rxjava4.core;
1515

1616
import java.util.Objects;
17+
import java.util.concurrent.Flow.*;
1718
import java.util.concurrent.TimeUnit;
1819

19-
import static java.util.concurrent.Flow.*;
20-
2120
import io.reactivex.rxjava4.annotations.*;
22-
import io.reactivex.rxjava4.core.*;
2321
import io.reactivex.rxjava4.disposables.Disposable;
2422
import io.reactivex.rxjava4.functions.Consumer;
2523
import io.reactivex.rxjava4.internal.functions.*;

src/main/java/io/reactivex/rxjava4/observables/ConnectableObservable.java renamed to src/main/java/io/reactivex/rxjava4/core/ConnectableObservable.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,13 +11,12 @@
1111
* the License for the specific language governing permissions and limitations under the License.
1212
*/
1313

14-
package io.reactivex.rxjava4.observables;
14+
package io.reactivex.rxjava4.core;
1515

1616
import java.util.Objects;
1717
import java.util.concurrent.TimeUnit;
1818

1919
import io.reactivex.rxjava4.annotations.*;
20-
import io.reactivex.rxjava4.core.*;
2120
import io.reactivex.rxjava4.disposables.Disposable;
2221
import io.reactivex.rxjava4.functions.Consumer;
2322
import io.reactivex.rxjava4.internal.functions.*;

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@
2222
import io.reactivex.rxjava4.core.config.*;
2323
import io.reactivex.rxjava4.disposables.*;
2424
import io.reactivex.rxjava4.exceptions.*;
25-
import io.reactivex.rxjava4.flowables.*;
2625
import io.reactivex.rxjava4.functions.*;
2726
import io.reactivex.rxjava4.internal.functions.*;
2827
import io.reactivex.rxjava4.internal.jdk8.*;

src/main/java/io/reactivex/rxjava4/flowables/GroupedFlowable.java renamed to src/main/java/io/reactivex/rxjava4/core/GroupedFlowable.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,9 @@
1111
* the License for the specific language governing permissions and limitations under the License.
1212
*/
1313

14-
package io.reactivex.rxjava4.flowables;
14+
package io.reactivex.rxjava4.core;
1515

1616
import io.reactivex.rxjava4.annotations.Nullable;
17-
import io.reactivex.rxjava4.core.Flowable;
1817

1918
/**
2019
* A {@link Flowable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.

src/main/java/io/reactivex/rxjava4/observables/GroupedObservable.java renamed to src/main/java/io/reactivex/rxjava4/core/GroupedObservable.java

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,9 @@
1111
* the License for the specific language governing permissions and limitations under the License.
1212
*/
1313

14-
package io.reactivex.rxjava4.observables;
14+
package io.reactivex.rxjava4.core;
1515

1616
import io.reactivex.rxjava4.annotations.Nullable;
17-
import io.reactivex.rxjava4.core.Observable;
1817

1918
/**
2019
* An {@link Observable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.

‎src/main/java/io/reactivex/rxjava4/core/GroupedStreamable.java‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/*
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava4.core;
15+
16+
import io.reactivex.rxjava4.annotations.*;
17+
18+
/**
19+
* An {@link Streamable} that has been grouped by key, the value of which can be obtained with {@link #getKey()}.
20+
* @param <K>
21+
* the type of the key, can be null
22+
* @param <T>
23+
* the type of the items emitted by the {@code GroupedStreamable}
24+
* @see Streamable#groupBy(io.reactivex.rxjava4.functions.Function)
25+
* @see <a href="http://reactivex.io/documentation/operators/groupby.html">ReactiveX documentation: GroupBy</a>
26+
* @since 4.0.0
27+
*/
28+
public abstract class GroupedStreamable<@Nullable K, @NonNull T> implements Streamable<T> {
29+
30+
final K key;
31+
32+
/**
33+
* Constructs a GroupedStreamable with the given key.
34+
* @param key the key
35+
*/
36+
protected GroupedStreamable(@Nullable K key) {
37+
this.key = key;
38+
}
39+
40+
/**
41+
* Returns the key that identifies the group of items emitted by this {@code GroupedStreamable}.
42+
*
43+
* @return the key that the items emitted by this {@code GroupedStreamable} were grouped by
44+
*/
45+
@Nullable
46+
public K getKey() {
47+
return key;
48+
}
49+
50+
}

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

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import io.reactivex.rxjava4.internal.operators.observable.*;
3333
import io.reactivex.rxjava4.internal.operators.single.SingleToObservable;
3434
import io.reactivex.rxjava4.internal.util.*;
35-
import io.reactivex.rxjava4.observables.*;
3635
import io.reactivex.rxjava4.observers.*;
3736
import io.reactivex.rxjava4.operators.ScalarSupplier;
3837
import io.reactivex.rxjava4.plugins.RxJavaPlugins;

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -572,6 +572,20 @@ default <R> Streamable<R> flatMap(@NonNull Function<? super T, ? extends Streama
572572
return RxJavaPlugins.onAssembly(new StreamableFlatMap<>(this, mapper, config.maxConcurrency()));
573573
}
574574

575+
/**
576+
* Maps each upstream item into a {@code GroupedStreamable} group, emits those groups and keeps
577+
* relaying the upstream items into those groups.
578+
* @param <K> the key type, {@code null}s allowed
579+
* @param keySelector the function that receives the upstream item and returns a key that determines
580+
* which group the item will go into
581+
* @return the new {@code Streamable} instance
582+
* @throws NullPointerException if {@code keySelector} is {@code null}
583+
*/
584+
default <@Nullable K> Streamable<GroupedStreamable<K, T>> groupBy(Function<? super T, ? extends K> keySelector) {
585+
Objects.requireNonNull(keySelector, "keySelector is null");
586+
return RxJavaPlugins.onAssembly(new StreamableGroupBy<>(this, keySelector));
587+
}
588+
575589
/**
576590
* Hides the identity of this {@code Streamable} and its {@link Streamer}.
577591
* <p>
@@ -688,6 +702,21 @@ default Streamable<T> takeWhile(@NonNull Predicate<? super T> predicate) {
688702
return RxJavaPlugins.onAssembly(new StreamableTakeWhile<>(this, predicate));
689703
}
690704

705+
/**
706+
* Relays items from this {@code Streamable} until the other {@code Streamable} signals
707+
* an item or completes.
708+
* @param <U> the element type of the other {@code Streamable}
709+
* @param other the {@code Streamable} expected to signal when to stop taking items from this {@code Streamable}
710+
* @return the new {@code Streamable} instance
711+
* @throws NullPointerException if {@code other} is {@code null}
712+
*/
713+
@CheckReturnValue
714+
@NonNull
715+
default <U> Streamable<T> takeUntil(@NonNull Streamable<U> other) {
716+
Objects.requireNonNull(other, "other is null");
717+
return RxJavaPlugins.onAssembly(new StreamableTakeUntil<>(this, other));
718+
}
719+
691720
/**
692721
* Calls the specified converter function during assembly time and returns its resulting value.
693722
* <p>

0 commit comments

Comments
 (0)