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

src/test/java/io/reactivex/rxjava4/observable/ObservableSubscriberTest.java
213 строк6 KB
David Karnok
4.x: Convert from JUnit 4 to JUnit 6 (#8202)
30 июн 2026, 12:01
30 июн 2026, 12:01531388f
/*
 * 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 static org.junit.jupiter.api.Assertions.*;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.jupiter.api.Test;

import io.reactivex.rxjava4.core.*;
import io.reactivex.rxjava4.core.Observable;
import io.reactivex.rxjava4.core.Observer;
import io.reactivex.rxjava4.exceptions.TestException;
import io.reactivex.rxjava4.observers.*;
import io.reactivex.rxjava4.plugins.RxJavaPlugins;
import io.reactivex.rxjava4.subjects.PublishSubject;
import io.reactivex.rxjava4.testsupport.TestHelper;

public class ObservableSubscriberTest extends RxJavaTest {
    @Test
    public void onStartCalledOnceViaSubscribe() {
        final AtomicInteger c = new AtomicInteger();
        Observable.just(1, 2, 3, 4).take(2)
        .subscribe(new DefaultObserver<>() /* NFI */ {

            @Override
            public void onStart() {
                c.incrementAndGet();
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer t) {
            }

        });

        assertEquals(1, c.get());
    }

    @Test
    public void onStartCalledOnceViaUnsafeSubscribe() {
        final AtomicInteger c = new AtomicInteger();
        Observable.just(1, 2, 3, 4).take(2)
        .subscribe(new DefaultObserver<>() /* NFI */ {

            @Override
            public void onStart() {
                c.incrementAndGet();
            }

            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer t) {
            }

        });

        assertEquals(1, c.get());
    }

    @Test
    public void onStartCalledOnceViaLift() {
        final AtomicInteger c = new AtomicInteger();
        Observable.just(1, 2, 3, 4)
        .lift((ObservableOperator<Integer, Integer>) child -> new DefaultObserver<>() /* NFI */ {

            @Override
            public void onStart() {
                c.incrementAndGet();
            }

            @Override
            public void onComplete() {
                child.onComplete();
            }

            @Override
            public void onError(Throwable e) {
                child.onError(e);
            }

            @Override
            public void onNext(Integer t) {
                child.onNext(t);
            }

        }).subscribe();

        assertEquals(1, c.get());
    }

    @Test
    public void subscribeConsumerConsumer() {
        final List<Integer> list = new ArrayList<>();

        Observable.just(1).subscribe(list::add, _ -> list.add(100));

        assertEquals(List.of(1), list);
    }

    @Test
    public void subscribeConsumerConsumerWithError() {
        final List<Integer> list = new ArrayList<>();

        Observable.<Integer>error(new TestException()).subscribe(list::add, _ -> list.add(100));

        assertEquals(List.of(100), list);
    }

    @Test
    public void methodTestCancelled() {
        PublishSubject<Integer> ps = PublishSubject.create();

        ps.test(true);

        assertFalse(ps.hasObservers());
    }

    @Test
    public void safeSubscriberAlreadySafe() {
        TestObserver<Integer> to = new TestObserver<>();
        Observable.just(1).safeSubscribe(new SafeObserver<>(to));

        to.assertResult(1);
    }

    @Test
    public void methodTestNoCancel() {
        PublishSubject<Integer> ps = PublishSubject.create();

        ps.test(false);

        assertTrue(ps.hasObservers());
    }

    @Test
    public void pluginNull() {
        RxJavaPlugins.setOnObservableSubscribe((_, _) -> null);

        try {
            try {

                Observable.just(1).test();
                fail("Should have thrown");
            } catch (NullPointerException ex) {
                assertEquals("The RxJavaPlugins.onSubscribe hook returned a null Observer. "
                        + "Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. "
                        + "Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins", ex.getMessage());
            }
        } finally {
            RxJavaPlugins.reset();
        }
    }

    static final class BadObservable extends Observable<Integer> {
        @Override
        protected void subscribeActual(Observer<? super Integer> observer) {
            throw new IllegalArgumentException();
        }
    }

    @Test
    public void subscribeActualThrows() {
        List<Throwable> list = TestHelper.trackPluginErrors();
        try {
            try {
                new BadObservable().test();
                fail("Should have thrown!");
            } catch (NullPointerException ex) {
                if (!(ex.getCause() instanceof IllegalArgumentException)) {
                    fail(ex.toString() + ": Should be NPE(IAE)");
                }
            }

            TestHelper.assertError(list, 0, IllegalArgumentException.class);
        } finally {
            RxJavaPlugins.reset();
        }
    }

}