src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java - 4.x - githubmirror/RxJava

src/test/java/io/reactivex/rxjava4/observable/ObservableGroupByTests.java

60 строк2 KB

David Karnok

4.x: Streamable + takeUntil + groupBy + refactor + helpers (#8215)

05 июл 2026, 01:45

05 июл 2026, 01:45580b39c

/*
 * 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.observable;

import org.junit.jupiter.api.Test;

import io.reactivex.rxjava4.core.*;
import io.reactivex.rxjava4.functions.Function;
import io.reactivex.rxjava4.observable.ObservableEventStream.Event;

public class ObservableGroupByTests extends RxJavaTest {

    @Test
    public void takeUnsubscribesOnGroupBy() throws Exception {
        Observable.mergeArray(
            ObservableEventStream.getEventStream("HTTP-ClusterA", 50),
            ObservableEventStream.getEventStream("HTTP-ClusterB", 20)
        )
        // group by type (2 clusters)
        .groupBy(Event::type)
        .take(1)
        .blockingForEach(v -> {
            System.out.println(v);
            v.take(1).subscribe();  // FIXME groups need consumption to a certain degree to cancel upstream
        });

        System.out.println("**** finished");

        Thread.sleep(200); // make sure the event streams receive their interrupt
    }

    @Test
    public void takeUnsubscribesOnFlatMapOfGroupBy() throws Exception {
        Observable.mergeArray(
            ObservableEventStream.getEventStream("HTTP-ClusterA", 50),
            ObservableEventStream.getEventStream("HTTP-ClusterB", 20)
        )
        // group by type (2 clusters)
        .groupBy(Event::type)
        .flatMap((Function<GroupedObservable<String, Event>, Observable<Object>>) g ->
            g.map((Function<Event, Object>) event -> event.instanceId() + " - " + event.values().get("count200")))
        .take(20)
        .blockingForEach(System.out::println);

        System.out.println("**** finished");

        Thread.sleep(200); // make sure the event streams receive their interrupt
    }
}