src/test/java/io/reactivex/rxjava4/disposables/SerialDisposableTests.java - 4.x - githubmirror/RxJava

src/test/java/io/reactivex/rxjava4/disposables/SerialDisposableTests.java
218 строк7 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.disposables;

import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.Mockito.*;

import java.util.*;
import java.util.concurrent.CountDownLatch;

import org.junit.jupiter.api.*;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.junit.jupiter.MockitoExtension;

import io.reactivex.rxjava4.core.RxJavaTest;
import io.reactivex.rxjava4.internal.disposables.DisposableHelper;

@ExtendWith(MockitoExtension.class)
public class SerialDisposableTests extends RxJavaTest {
    private SerialDisposable serialDisposable;

    @BeforeEach
    public void setUp() {
        serialDisposable = new SerialDisposable();
    }

    @Test
    public void unsubscribingWithoutUnderlyingDoesNothing() {
        serialDisposable.dispose();
    }

    @Test
    public void getDisposableShouldReturnset() {
        final Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);
        assertSame(underlying, serialDisposable.get());

        final Disposable another = mock(Disposable.class);
        serialDisposable.set(another);
        assertSame(another, serialDisposable.get());
    }

    @Test
    public void notDisposedWhenReplaced() {
        final Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);

        serialDisposable.replace(Disposable.empty());
        serialDisposable.dispose();

        verify(underlying, never()).dispose();
    }

    @Test
    public void unsubscribingTwiceDoesUnsubscribeOnce() {
        Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);

        serialDisposable.dispose();
        verify(underlying).dispose();

        serialDisposable.dispose();
        verifyNoMoreInteractions(underlying);
    }

    @Test
    public void settingSameDisposableTwiceDoesUnsubscribeIt() {
        Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);
        verifyNoInteractions(underlying);
        serialDisposable.set(underlying);
        verify(underlying).dispose();
    }

    @Test
    public void unsubscribingWithSingleUnderlyingUnsubscribes() {
        Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);
        underlying.dispose();
        verify(underlying).dispose();
    }

    @Test
    public void replacingFirstUnderlyingCausesUnsubscription() {
        Disposable first = mock(Disposable.class);
        serialDisposable.set(first);
        Disposable second = mock(Disposable.class);
        serialDisposable.set(second);
        verify(first).dispose();
    }

    @Test
    public void whenUnsubscribingSecondUnderlyingUnsubscribed() {
        Disposable first = mock(Disposable.class);
        serialDisposable.set(first);
        Disposable second = mock(Disposable.class);
        serialDisposable.set(second);
        serialDisposable.dispose();
        verify(second).dispose();
    }

    @Test
    public void settingUnderlyingWhenUnsubscribedCausesImmediateUnsubscription() {
        serialDisposable.dispose();
        Disposable underlying = mock(Disposable.class);
        serialDisposable.set(underlying);
        verify(underlying).dispose();
    }

    @Test
    public void settingUnderlyingWhenUnsubscribedCausesImmediateUnsubscriptionConcurrently()
            throws InterruptedException {
        final Disposable firstSet = mock(Disposable.class);
        serialDisposable.set(firstSet);

        final CountDownLatch start = new CountDownLatch(1);

        final int count = 10;
        final CountDownLatch end = new CountDownLatch(count);

        final List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            final Thread t = new Thread(() -> {
                try {
                    start.await();
                    serialDisposable.dispose();
                } catch (InterruptedException e) {
                    fail(e.getMessage());
                } finally {
                    end.countDown();
                }
            });
            t.start();
            threads.add(t);
        }

        final Disposable underlying = mock(Disposable.class);
        start.countDown();
        serialDisposable.set(underlying);
        end.await();
        verify(firstSet).dispose();
        verify(underlying).dispose();

        for (final Thread t : threads) {
            t.join();
        }
    }

    @Test
    public void concurrentSetDisposableShouldNotInterleave()
            throws InterruptedException {
        final int count = 10;
        final List<Disposable> subscriptions = new ArrayList<>();

        final CountDownLatch start = new CountDownLatch(1);
        final CountDownLatch end = new CountDownLatch(count);

        final List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < count; i++) {
            final Disposable subscription = mock(Disposable.class);
            subscriptions.add(subscription);

            final Thread t = new Thread(() -> {
                try {
                    start.await();
                    serialDisposable.set(subscription);
                } catch (InterruptedException e) {
                    fail(e.getMessage());
                } finally {
                    end.countDown();
                }
            });
            t.start();
            threads.add(t);
        }

        start.countDown();
        end.await();
        serialDisposable.dispose();

        for (final Disposable subscription : subscriptions) {
            verify(subscription).dispose();
        }

        for (final Thread t : threads) {
            t.join();
        }
    }

    @Test
    public void disposeState() {
        Disposable empty = Disposable.empty();
        SerialDisposable d = new SerialDisposable(empty);

        assertFalse(d.isDisposed());

        assertSame(empty, d.get());

        d.dispose();

        assertTrue(d.isDisposed());

        assertNotSame(empty, d.get());

        assertNotSame(DisposableHelper.DISPOSED, d.get());
    }
}