src/test/java/io/reactivex/rxjava4/internal/schedulers/ParallelSchedulerTest.java - 4.x - githubmirror/RxJava

src/test/java/io/reactivex/rxjava4/internal/schedulers/ParallelSchedulerTest.java
475 строк15 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.internal.schedulers;

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

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

import org.junit.jupiter.api.*;

import io.reactivex.rxjava4.core.*;
import io.reactivex.rxjava4.core.Scheduler.Worker;
import io.reactivex.rxjava4.core.config.*;
import io.reactivex.rxjava4.disposables.*;
import io.reactivex.rxjava4.internal.functions.Functions;
import io.reactivex.rxjava4.internal.schedulers.ParallelScheduler.TrackingParallelWorker.TrackedAction;
import io.reactivex.rxjava4.schedulers.*;
import io.reactivex.rxjava4.testsupport.TestHelper;

public class ParallelSchedulerTest implements Runnable {

    final AtomicInteger calls = new AtomicInteger();

    @Override
    public void run() {
        calls.getAndIncrement();
    }

    @Test
    public void normalNonTracking() {
        Scheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);

        try {
            for (int i = 0; i < 100; i++) {
                Flowable.range(1, 10).hide()
                .observeOn(s, new StandardBufferedConfig(false, 4))
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void normalNonTrackingVia() {
        Scheduler s = Schedulers.createParallel(new ParallelSchedulerConfig(2, false, ParallelScheduler.DEFAULT_FACTORY));

        try {
            for (int i = 0; i < 100; i++) {
                Flowable.range(1, 10).hide()
                .observeOn(s, new StandardBufferedConfig(false, 4))
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void delayedNonTracking() {
        Scheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);

        try {
            for (int i = 0; i < 100; i++) {
                Flowable.range(1, 10).hide()
                .delay(50, TimeUnit.MILLISECONDS, s)
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void normalTracking() {
        Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);

        try {
            for (int i = 0; i < 100; i++) {
                Flowable.range(1, 10).hide()
                .observeOn(s, new StandardBufferedConfig(false, 4))
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void delayedTracking() {
        Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);

        try {
            for (int i = 0; i < 100; i++) {
                Flowable.range(1, 10).hide()
                .delay(50, TimeUnit.MILLISECONDS, s)
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void shutdownNonTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);

        shutdown(s);
    }

    @Test
    public void shutdownTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);

        shutdown(s);
    }

    protected void shutdown(Scheduler s) throws InterruptedException {
        try {
            Worker w = s.createWorker();

            w.dispose();

            assertSame(Disposable.disposed(), w.schedule(this));

            assertSame(Disposable.disposed(), w.schedule(this, 100, TimeUnit.MILLISECONDS));

            assertSame(Disposable.disposed(), w.schedulePeriodically(this, 100, 100, TimeUnit.MILLISECONDS));

            s.shutdown();

            assertSame(Disposable.disposed(), s.scheduleDirect(this));

            assertSame(Disposable.disposed(), s.scheduleDirect(this, 100, TimeUnit.MILLISECONDS));

            assertSame(Disposable.disposed(), s.schedulePeriodicallyDirect(this, 100, 100, TimeUnit.MILLISECONDS));

            w = s.createWorker();

            assertSame(Disposable.disposed(), w.schedule(this));

            assertSame(Disposable.disposed(), w.schedule(this, 100, TimeUnit.MILLISECONDS));

            assertSame(Disposable.disposed(), w.schedulePeriodically(this, 100, 100, TimeUnit.MILLISECONDS));

            assertEquals(0, calls.get());

            s.start();

            s.scheduleDirect(this);

            s.scheduleDirect(this, 100, TimeUnit.MILLISECONDS);

            s.schedulePeriodicallyDirect(this, 100, 100, TimeUnit.MILLISECONDS);

            w = s.createWorker();

            w.schedule(this);

            w.schedule(this, 100, TimeUnit.MILLISECONDS);

            w.schedulePeriodically(this, 100, 100, TimeUnit.MILLISECONDS);

            Thread.sleep(1000);

            int c = calls.get();
            assertTrue(c > 6, "" + c);
        } finally {
            s.shutdown();
        }
    }

    @Test
    @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
    public void taskThrowsNonTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);
        taskThrows(s);
    }

    @Test
    @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
    public void taskThrowsTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
        taskThrows(s);
    }

    protected void taskThrows(Scheduler s) throws InterruptedException {
        try {
            List<Throwable> errors = TestHelper.trackPluginErrors();

            Worker w = s.createWorker();

            w.schedule(() -> {
                calls.getAndIncrement();
                throw new IllegalStateException();
            });

            while (errors.isEmpty()) {
                Thread.sleep(20);
            }

            TestHelper.assertError(errors, 0, IllegalStateException.class);
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void cancelledTaskNotTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);
        cancelledTask(s);
    }

    @Test
    public void cancelledTaskTracking() throws Exception {
        Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
        cancelledTask(s);
    }

    void cancelledTask(Scheduler s) throws InterruptedException {
        try {
            Worker w = s.createWorker();

            try {
                assertFalse(w.isDisposed());

                Disposable d = w.schedule(this, 200, TimeUnit.MILLISECONDS);

                assertFalse(d.isDisposed());

                d.dispose();

                assertTrue(d.isDisposed());

                Thread.sleep(300);

                assertEquals(0, calls.get());
                w.dispose();

                assertTrue(w.isDisposed());
            } finally {
                w.dispose();
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void constructors() {
        startStop(new ParallelScheduler(1, true, new RxThreadFactory("Test")));
    }

    private void startStop(Scheduler s) {
        s.start();
        s.shutdown();
        s.shutdown();
    }

    @Test
    public void shutdownBackingTracking() {
        ParallelScheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);

        shutdownBacking(s);
    }

    @Test
    public void shutdownBackingNonTracking() {
        ParallelScheduler s = new ParallelScheduler(2, false, ParallelScheduler.DEFAULT_FACTORY);

        shutdownBacking(s);
    }

    private void shutdownBacking(ParallelScheduler s) {
        for (ScheduledExecutorService exec : s.pool.get()) {
            exec.shutdown();
        }

        assertSame(Disposable.disposed(), s.scheduleDirect(this));

        assertSame(Disposable.disposed(), s.scheduleDirect(this, 100, TimeUnit.MILLISECONDS));

        assertSame(Disposable.disposed(), s.schedulePeriodicallyDirect(this, 100, 100, TimeUnit.MILLISECONDS));

        Worker w = s.createWorker();

        assertSame(Disposable.disposed(), w.schedule(this));

        assertSame(Disposable.disposed(), w.schedule(this, 100, TimeUnit.MILLISECONDS));

        assertSame(Disposable.disposed(), w.schedulePeriodically(this, 100, 100, TimeUnit.MILLISECONDS));

        assertEquals(0, calls.get());
    }

    @Test
    public void startRace() {
        for (int i = 0; i < 1000; i++) {
            final Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
            s.shutdown();

            Runnable r = s::start;

            TestHelper.race(r, r, Schedulers.single());
        }
    }

    @Test
    public void setFutureRace() {
        final Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
        try {
            for (int i = 0; i < 1000; i++) {
                final Worker w = s.createWorker();

                Runnable r1 = () -> w.schedule(ParallelSchedulerTest.this);

                Runnable r2 = w::dispose;
                TestHelper.race(r1, r2, Schedulers.single());
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void setFutureRace2() {
        final Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
        try {
            for (int i = 0; i < 1000; i++) {
                final CompositeDisposable cd = new CompositeDisposable();
                var tt = new TrackedAction(this, cd);
                final FutureTask<Object> ft = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);

                Runnable r1 = () -> tt.setFuture(ft);

                Runnable r2 = () -> tt.future.set(TrackedAction.FINISHED);
                TestHelper.race(r1, r2, Schedulers.single());
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void setFutureRace3() {
        final Scheduler s = new ParallelScheduler(2, true, ParallelScheduler.DEFAULT_FACTORY);
        try {
            for (int i = 0; i < 1000; i++) {
                final CompositeDisposable cd = new CompositeDisposable();

                var tt = new TrackedAction(this, cd);
                final FutureTask<Object> ft = new FutureTask<>(Functions.EMPTY_RUNNABLE, null);

                Runnable r1 = () -> tt.setFuture(ft);

                Runnable r2 = () -> tt.future.set(TrackedAction.DISPOSED);
                TestHelper.race(r1, r2, Schedulers.single());
            }
        } finally {
            s.shutdown();
        }
    }

    @Test
    public void illegalParallelism() {
        assertThrows(IllegalArgumentException.class, () -> {
            new ParallelScheduler(0, true, ParallelScheduler.DEFAULT_FACTORY);
        });
    }

    @Test
    public void parallelSchedulerConfig() {
        {
        var psc1 = ParallelSchedulerConfig.DEFAULT;
        assertEquals(psc1.parallelism(), Runtime.getRuntime().availableProcessors(), "Parallelism");
        assertTrue(psc1.tracking(), "Tracking");
        assertEquals(psc1.threadNamePrefix(), "RxParallelScheduler", "threadNamePrefix");
        }

        {
        var psc2 = new ParallelSchedulerConfig(1);
        assertEquals(psc2.parallelism(), 1, "Parallelism");
        assertTrue(psc2.tracking(), "Tracking");
        assertEquals(psc2.threadNamePrefix(), "RxParallelScheduler", "threadNamePrefix");
        }

        {
        var psc3 = new ParallelSchedulerConfig(1, false);
        assertEquals(psc3.parallelism(), 1, "Parallelism");
        assertFalse(psc3.tracking(), "Tracking");
        assertEquals(psc3.threadNamePrefix(), "RxParallelScheduler", "threadNamePrefix");
        }

        {
        var psc4 = new ParallelSchedulerConfig(1, "Test");
        assertEquals(psc4.parallelism(), 1, "Parallelism");
        assertTrue(psc4.tracking(), "Tracking");
        assertEquals(psc4.threadNamePrefix(), "Test", "threadNamePrefix");
        }

        {
        var psc5 = new ParallelSchedulerConfig(1, false, "Test");
        assertEquals(psc5.parallelism(), 1, "Parallelism");
        assertFalse(psc5.tracking(), "Tracking");
        assertEquals(psc5.threadNamePrefix(), "Test", "threadNamePrefix");
        }
    }

    @Test
    public void introspectionNonTracking() {
        Runnable run = () -> { };
        var scheduler = Schedulers.createParallel(new ParallelSchedulerConfig(2, false));
        try {
            var worker = scheduler.createWorker();
            try {
                var task = worker.schedule(run);

                if (task instanceof SchedulerRunnableIntrospection intro) {
                    assertSame(run, intro.getWrappedRunnable());
                } else {
                    fail(task.getClass() + " doesn't implement SchedulerRunnableIntrospection");
                }
            } finally {
                worker.dispose();
            }
        } finally {
            scheduler.shutdown();
        }
    }

    @Test
    public void introspectionTracking() {
        Runnable run = () -> { };
        var scheduler = Schedulers.createParallel(new ParallelSchedulerConfig(2, true));
        try {
            var worker = scheduler.createWorker();
            try {
                var task = worker.schedule(run);

                if (task instanceof SchedulerRunnableIntrospection intro) {
                    assertSame(run, intro.getWrappedRunnable());
                } else {
                    fail(task.getClass() + " doesn't implement SchedulerRunnableIntrospection");
                }
            } finally {
                worker.dispose();
            }
        } finally {
            scheduler.shutdown();
        }
    }
}