code: /*
   * 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.rxjava3.processors;
  
  import static org.junit.Assert.*;
  
  import java.io.IOException;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  
  import org.junit.Test;
  import org.reactivestreams.Subscription;
  
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.exceptions.*;
  import io.reactivex.rxjava3.functions.Function;
  import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription;
  import io.reactivex.rxjava3.plugins.RxJavaPlugins;
  import io.reactivex.rxjava3.subscribers.TestSubscriber;
  import io.reactivex.rxjava3.testsupport.TestHelper;
  
  public class MulticastProcessorTest extends RxJavaTest {
  
      @Test
      public void complete() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
          mp.start();
  
          assertFalse(mp.hasSubscribers());
          assertFalse(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          TestSubscriber<Integer> ts = mp.test();
  
          assertTrue(mp.hasSubscribers());
          assertFalse(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          mp.onNext(1);
          mp.onComplete();
  
          ts.assertResult(1);
  
          assertFalse(mp.hasSubscribers());
          assertTrue(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          mp.test().assertResult();
      }
  
      @Test
      public void error() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
          mp.start();
  
          assertFalse(mp.hasSubscribers());
          assertFalse(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          TestSubscriber<Integer> ts = mp.test();
  
          assertTrue(mp.hasSubscribers());
          assertFalse(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          mp.onNext(1);
          mp.onError(new IOException());
  
          ts.assertFailure(IOException.class, 1);
  
          assertFalse(mp.hasSubscribers());
          assertFalse(mp.hasComplete());
          assertTrue(mp.hasThrowable());
          assertNotNull(mp.getThrowable());
          assertTrue("" + mp.getThrowable(), mp.getThrowable() instanceof IOException);
  
          mp.test().assertFailure(IOException.class);
      }
  
      @Test
      public void overflow() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(1);
          mp.start();
  
          TestSubscriber<Integer> ts = mp.test(0);
  
          assertTrue(mp.offer(1));
          assertFalse(mp.offer(2));
  
          mp.onNext(3);
  
          ts.assertEmpty();
  
          ts.request(1);
  
          ts.assertFailure(MissingBackpressureException.class, 1);
  
          mp.test().assertFailure(MissingBackpressureException.class);
      }
  
      @Test
      public void backpressure() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16, false);
          mp.start();
  
          for (int i = 0; i < 10; i++) {
              mp.onNext(i);
          }
          mp.onComplete();
  
          mp.test(0)
          .assertEmpty()
          .requestMore(1)
          .assertValuesOnly(0)
          .requestMore(2)
          .assertValuesOnly(0, 1, 2)
          .requestMore(3)
          .assertValuesOnly(0, 1, 2, 3, 4, 5)
          .requestMore(4)
          .assertResult(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
      }
  
      @Test
      public void refCounted() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
          BooleanSubscription bs = new BooleanSubscription();
  
          mp.onSubscribe(bs);
  
          assertFalse(bs.isCancelled());
  
          mp.test().cancel();
  
          assertTrue(bs.isCancelled());
  
          assertFalse(mp.hasSubscribers());
          assertTrue(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          mp.test().assertResult();
      }
  
      @Test
      public void refCounted2() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16, true);
          BooleanSubscription bs = new BooleanSubscription();
  
          mp.onSubscribe(bs);
  
          assertFalse(bs.isCancelled());
  
          mp.test(1, true);
  
          assertTrue(bs.isCancelled());
  
          assertFalse(mp.hasSubscribers());
          assertTrue(mp.hasComplete());
          assertFalse(mp.hasThrowable());
          assertNull(mp.getThrowable());
  
          mp.test().assertResult();
      }
  
      @Test
      public void longRunning() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16);
          Flowable.range(1, 1000).subscribe(mp);
  
          mp.test().assertValueCount(1000).assertNoErrors().assertComplete();
      }
  
      @Test
      public void oneByOne() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16);
          Flowable.range(1, 1000).subscribe(mp);
  
          mp
          .rebatchRequests(1)
          .test()
          .assertValueCount(1000)
          .assertNoErrors()
          .assertComplete();
      }
  
      @Test
      public void take() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16);
          Flowable.range(1, 1000).subscribe(mp);
  
          mp.take(10).test().assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
      }
  
      @Test
      public void takeRefCount() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16, true);
          Flowable.range(1, 1000).subscribe(mp);
  
          mp.take(10).test().assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
      }
  
      @Test
      public void takeRefCountExact() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(16, true);
          Flowable.range(1, 10).subscribe(mp);
  
          mp
          .rebatchRequests(10)
          .take(10)
          .test().assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
      }
  
      @Test
      public void crossCancel() {
  
          final TestSubscriber<Integer> ts1 = new TestSubscriber<>();
  
          TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>() {
              @Override
              public void onNext(Integer t) {
                  super.onNext(t);
                  ts1.cancel();
                  ts1.onComplete();
              }
          };
  
          MulticastProcessor<Integer> mp = MulticastProcessor.create(false);
  
          mp.subscribe(ts2);
          mp.subscribe(ts1);
  
          mp.start();
  
          mp.onNext(1);
          mp.onComplete();
  
          ts1.assertResult();
          ts2.assertResult(1);
      }
  
      @Test
      public void crossCancelError() {
  
          final TestSubscriber<Integer> ts1 = new TestSubscriber<>();
  
          TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>() {
              @Override
              public void onError(Throwable t) {
                  super.onError(t);
                  ts1.cancel();
                  ts1.onComplete();
              }
          };
  
          MulticastProcessor<Integer> mp = MulticastProcessor.create(false);
  
          mp.subscribe(ts2);
          mp.subscribe(ts1);
  
          mp.start();
  
          mp.onNext(1);
          mp.onError(new IOException());
  
          ts1.assertResult(1);
          ts2.assertFailure(IOException.class, 1);
      }
  
      @Test
      public void crossCancelComplete() {
  
          final TestSubscriber<Integer> ts1 = new TestSubscriber<>();
  
          TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>() {
              @Override
              public void onComplete() {
                  super.onComplete();
                  ts1.cancel();
                  ts1.onNext(2);
                  ts1.onComplete();
              }
          };
  
          MulticastProcessor<Integer> mp = MulticastProcessor.create(false);
  
          mp.subscribe(ts2);
          mp.subscribe(ts1);
  
          mp.start();
  
          mp.onNext(1);
          mp.onComplete();
  
          ts1.assertResult(1, 2);
          ts2.assertResult(1);
      }
  
      @Test
      public void crossCancel1() {
  
          final TestSubscriber<Integer> ts1 = new TestSubscriber<>(1);
  
          TestSubscriber<Integer> ts2 = new TestSubscriber<Integer>(1) {
              @Override
              public void onNext(Integer t) {
                  super.onNext(t);
                  ts1.cancel();
                  ts1.onComplete();
              }
          };
  
          MulticastProcessor<Integer> mp = MulticastProcessor.create(false);
  
          mp.subscribe(ts2);
          mp.subscribe(ts1);
  
          mp.start();
  
          mp.onNext(1);
          mp.onComplete();
  
          ts1.assertResult();
          ts2.assertResult(1);
      }
  
      @Test
      public void requestCancel() {
          List<Throwable> errors = TestHelper.trackPluginErrors();
          try {
              MulticastProcessor<Integer> mp = MulticastProcessor.create(false);
  
              mp.subscribe(new FlowableSubscriber<Integer>() {
  
                  @Override
                  public void onNext(Integer t) {
                  }
  
                  @Override
                  public void onError(Throwable t) {
                  }
  
                  @Override
                  public void onComplete() {
                  }
  
                  @Override
                  public void onSubscribe(Subscription t) {
                      t.request(-1);
                      t.request(1);
                      t.request(Long.MAX_VALUE);
                      t.request(Long.MAX_VALUE);
                      t.cancel();
                      t.cancel();
                      t.request(2);
                  }
              });
  
              TestHelper.assertError(errors, 0, IllegalArgumentException.class);
          } finally {
              RxJavaPlugins.reset();
          }
      }
  
      @Test
      public void unbounded() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(4, false);
          mp.startUnbounded();
  
          for (int i = 0; i < 10; i++) {
              assertTrue(mp.offer(i));
          }
          mp.onComplete();
  
          mp.test().assertResult(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
      }
  
      @Test
      public void multiStart() {
          List<Throwable> errors = TestHelper.trackPluginErrors();
          try {
              MulticastProcessor<Integer> mp = MulticastProcessor.create(4, false);
  
              mp.start();
              mp.start();
              mp.startUnbounded();
              BooleanSubscription bs = new BooleanSubscription();
              mp.onSubscribe(bs);
  
              assertTrue(bs.isCancelled());
  
              TestHelper.assertError(errors, 0, ProtocolViolationException.class);
              TestHelper.assertError(errors, 1, ProtocolViolationException.class);
              TestHelper.assertError(errors, 2, ProtocolViolationException.class);
          } finally {
              RxJavaPlugins.reset();
          }
      }
  
      @Test(expected = NullPointerException.class)
      public void onNextNull() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(4, false);
          mp.start();
          mp.onNext(null);
      }
  
      @Test(expected = NullPointerException.class)
      public void onOfferNull() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(4, false);
          mp.start();
          mp.offer(null);
      }
  
      @Test(expected = NullPointerException.class)
      public void onErrorNull() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create(4, false);
          mp.start();
          mp.onError(null);
      }
  
      @Test
      public void afterTerminated() {
          List<Throwable> errors = TestHelper.trackPluginErrors();
          try {
              MulticastProcessor<Integer> mp = MulticastProcessor.create();
              mp.start();
              mp.onComplete();
              mp.onComplete();
              mp.onError(new IOException());
              mp.onNext(1);
              mp.offer(1);
  
              mp.test().assertResult();
  
              TestHelper.assertUndeliverable(errors, 0, IOException.class);
          } finally {
              RxJavaPlugins.reset();
          }
      }
  
      @Test
      public void asyncFused() {
          UnicastProcessor<Integer> up = UnicastProcessor.create();
          MulticastProcessor<Integer> mp = MulticastProcessor.create(4);
  
          up.subscribe(mp);
  
          TestSubscriber<Integer> ts = mp.test();
  
          for (int i = 0; i < 10; i++) {
              up.onNext(i);
          }
  
          try {
              mp.offer(10);
              fail("Should have thrown IllegalStateException");
          } catch (IllegalStateException expected) {
              // expected
          }
  
          up.onComplete();
  
          ts.assertResult(0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
      }
  
      @Test
      public void fusionCrash() {
          MulticastProcessor<Integer> mp = Flowable.range(1, 5)
          .map(new Function<Integer, Integer>() {
              @Override
              public Integer apply(Integer v) throws Exception {
                  throw new IOException();
              }
          })
          .subscribeWith(MulticastProcessor.<Integer>create());
  
          mp.test().assertFailure(IOException.class);
      }
  
      @Test
      public void lockstep() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
          TestSubscriber<Integer> ts1 = mp.test();
          mp.start();
  
          mp.onNext(1);
          mp.onNext(2);
  
          ts1.assertValues(1, 2);
  
          TestSubscriber<Integer> ts2 = mp.test(0);
  
          ts2.assertEmpty();
  
          mp.onNext(3);
  
          ts1.assertValues(1, 2);
          ts2.assertEmpty();
  
          mp.onComplete();
  
          ts1.assertValues(1, 2);
          ts2.assertEmpty();
  
          ts2.request(1);
  
          ts1.assertResult(1, 2, 3);
          ts2.assertResult(3);
      }
  
      @Test
      public void rejectedFusion() {
  
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
          TestHelper.<Integer>rejectFlowableFusion()
          .subscribe(mp);
  
          mp.test().assertEmpty();
      }
  
      @Test
      public void addRemoveRaceNoRefCount() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
              final TestSubscriber<Integer> ts = mp.test();
              final TestSubscriber<Integer> ts2 = new TestSubscriber<>();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      mp.subscribe(ts2);
                  }
              };
  
              TestHelper.race(r1, r2);
  
              assertTrue(mp.hasSubscribers());
          }
      }
  
      @Test
      public void addRemoveRaceNoRefCountNonEmpty() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
              mp.test();
              final TestSubscriber<Integer> ts = mp.test();
              final TestSubscriber<Integer> ts2 = new TestSubscriber<>();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      mp.subscribe(ts2);
                  }
              };
  
              TestHelper.race(r1, r2);
  
              assertTrue(mp.hasSubscribers());
          }
      }
  
      @Test
      public void addRemoveRaceWitRefCount() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
  
              final TestSubscriber<Integer> ts = mp.test();
              final TestSubscriber<Integer> ts2 = new TestSubscriber<>();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      mp.subscribe(ts2);
                  }
              };
  
              TestHelper.race(r1, r2);
          }
      }
  
      @Test
      public void cancelUpfront() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
          mp.test(0, true).assertEmpty();
  
          assertFalse(mp.hasSubscribers());
      }
  
      @Test
      public void cancelUpfrontOtherConsumersPresent() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
          mp.test();
  
          mp.test(0, true).assertEmpty();
  
          assertTrue(mp.hasSubscribers());
      }
  
      @Test
      public void consumerRequestRace() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
              mp.startUnbounded();
              mp.onNext(1);
              mp.onNext(2);
  
              final TestSubscriber<Integer> ts = mp.test();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.request(1);
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      ts.request(1);
                  }
              };
  
              TestHelper.race(r1, r2);
  
              ts.assertValuesOnly(1, 2);
          }
      }
  
      @Test
      public void consumerUpstreamRace() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
  
              final Flowable<Integer> source = Flowable.range(1, 5);
  
              final TestSubscriber<Integer> ts = mp.test();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.request(5);
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      source.subscribe(mp);
                  }
              };
  
              TestHelper.race(r1, r2);
  
              ts
              .awaitDone(5, TimeUnit.SECONDS)
              .assertResult(1, 2, 3, 4, 5);
          }
      }
  
      @Test
      public void emitCancelRace() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
              mp.startUnbounded();
  
              final TestSubscriber<Integer> ts = mp.test();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      mp.onNext(1);
                  }
              };
  
              TestHelper.race(r1, r2);
          }
      }
  
      @Test
      public void cancelCancelDrain() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
  
              final TestSubscriber<Integer> ts1 = mp.test();
              final TestSubscriber<Integer> ts2 = mp.test();
  
              mp.test();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts1.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      ts2.cancel();
                  }
              };
  
              TestHelper.race(r1, r2);
          }
      }
  
      @Test
      public void requestCancelRace() {
          for (int i = 0; i < TestHelper.RACE_LONG_LOOPS; i++) {
              final MulticastProcessor<Integer> mp = MulticastProcessor.create(true);
  
              final TestSubscriber<Integer> ts1 = mp.test();
  
              Runnable r1 = new Runnable() {
                  @Override
                  public void run() {
                      ts1.cancel();
                  }
              };
  
              Runnable r2 = new Runnable() {
                  @Override
                  public void run() {
                      ts1.request(1);
                  }
              };
  
              TestHelper.race(r1, r2);
          }
      }
  
      @Test
      public void noUpstream() {
          MulticastProcessor<Integer> mp = MulticastProcessor.create();
  
          TestSubscriber<Integer> ts = mp.test(0);
  
          ts.request(1);
  
          assertTrue(mp.hasSubscribers());
      }
  
      @Test
      public void requestUpstreamPrefetchNonFused() {
          for (int j = 1; j < 12; j++) {
              MulticastProcessor<Integer> mp = MulticastProcessor.create(j, true);
  
              TestSubscriber<Integer> ts = mp.test(0).withTag("Prefetch: " + j);
  
              Flowable.range(1, 10).hide().subscribe(mp);
  
              ts.assertEmpty()
              .requestMore(3)
              .assertValuesOnly(1, 2, 3)
              .requestMore(3)
              .assertValuesOnly(1, 2, 3, 4, 5, 6)
              .requestMore(4)
              .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
          }
      }
  
      @Test
      public void requestUpstreamPrefetchNonFused2() {
          for (int j = 1; j < 12; j++) {
              MulticastProcessor<Integer> mp = MulticastProcessor.create(j, true);
  
              TestSubscriber<Integer> ts = mp.test(0).withTag("Prefetch: " + j);
  
              Flowable.range(1, 10).hide().subscribe(mp);
  
              ts.assertEmpty()
              .requestMore(2)
              .assertValuesOnly(1, 2)
              .requestMore(2)
              .assertValuesOnly(1, 2, 3, 4)
              .requestMore(6)
              .assertResult(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
          }
      }
  }
  