content: /**
   * 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.flowable.internal.operators;
  
  import static org.junit.Assert.*;
  import static org.mockito.ArgumentMatchers.*;
  import static org.mockito.Mockito.*;
  
  import java.io.IOException;
  import java.util.List;
  import java.util.concurrent.*;
  
  import org.junit.*;
  import org.mockito.InOrder;
  import org.reactivestreams.*;
  
  import io.reactivex.common.*;
  import io.reactivex.common.exceptions.TestException;
  import io.reactivex.flowable.*;
  import io.reactivex.flowable.internal.subscriptions.BooleanSubscription;
  import io.reactivex.flowable.processors.PublishProcessor;
  import io.reactivex.flowable.subscribers.TestSubscriber;
  
  public class FlowableTimeoutTests {
      private PublishProcessor<String> underlyingSubject;
      private TestScheduler testScheduler;
      private Flowable<String> withTimeout;
      private static final long TIMEOUT = 3;
      private static final TimeUnit TIME_UNIT = TimeUnit.SECONDS;
  
      @Before
      public void setUp() {
  
          underlyingSubject = PublishProcessor.create();
          testScheduler = new TestScheduler();
          withTimeout = underlyingSubject.timeout(TIMEOUT, TIME_UNIT, testScheduler);
      }
  
      @Test
      public void shouldNotTimeoutIfOnNextWithinTimeout() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
  
          withTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          verify(observer).onNext("One");
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          verify(observer, never()).onError(any(Throwable.class));
          ts.cancel();
      }
  
      @Test
      public void shouldNotTimeoutIfSecondOnNextWithinTimeout() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
  
          withTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("Two");
          verify(observer).onNext("Two");
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          verify(observer, never()).onError(any(Throwable.class));
          ts.dispose();
      }
  
      @Test
      public void shouldTimeoutIfOnNextNotWithinTimeout() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
  
          withTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(TIMEOUT + 1, TimeUnit.SECONDS);
          verify(observer).onError(any(TimeoutException.class));
          ts.dispose();
      }
  
      @Test
      public void shouldTimeoutIfSecondOnNextNotWithinTimeout() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          withTimeout.subscribe(observer);
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          verify(observer).onNext("One");
          testScheduler.advanceTimeBy(TIMEOUT + 1, TimeUnit.SECONDS);
          verify(observer).onError(any(TimeoutException.class));
          ts.dispose();
      }
  
      @Test
      public void shouldCompleteIfUnderlyingComletes() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          withTimeout.subscribe(observer);
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onComplete();
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          verify(observer).onComplete();
          verify(observer, never()).onError(any(Throwable.class));
          ts.dispose();
      }
  
      @Test
      public void shouldErrorIfUnderlyingErrors() {
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          withTimeout.subscribe(observer);
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onError(new UnsupportedOperationException());
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          verify(observer).onError(any(UnsupportedOperationException.class));
          ts.dispose();
      }
  
      @Test
      public void shouldSwitchToOtherIfOnNextNotWithinTimeout() {
          Flowable<String> other = Flowable.just("a", "b", "c");
          Flowable<String> source = underlyingSubject.timeout(TIMEOUT, TIME_UNIT, testScheduler, other);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          source.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          testScheduler.advanceTimeBy(4, TimeUnit.SECONDS);
          underlyingSubject.onNext("Two");
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onNext("One");
          inOrder.verify(observer, times(1)).onNext("a");
          inOrder.verify(observer, times(1)).onNext("b");
          inOrder.verify(observer, times(1)).onNext("c");
          inOrder.verify(observer, times(1)).onComplete();
          inOrder.verifyNoMoreInteractions();
          ts.dispose();
      }
  
      @Test
      public void shouldSwitchToOtherIfOnErrorNotWithinTimeout() {
          Flowable<String> other = Flowable.just("a", "b", "c");
          Flowable<String> source = underlyingSubject.timeout(TIMEOUT, TIME_UNIT, testScheduler, other);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          source.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          testScheduler.advanceTimeBy(4, TimeUnit.SECONDS);
          underlyingSubject.onError(new UnsupportedOperationException());
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onNext("One");
          inOrder.verify(observer, times(1)).onNext("a");
          inOrder.verify(observer, times(1)).onNext("b");
          inOrder.verify(observer, times(1)).onNext("c");
          inOrder.verify(observer, times(1)).onComplete();
          inOrder.verifyNoMoreInteractions();
          ts.dispose();
      }
  
      @Test
      public void shouldSwitchToOtherIfOnCompletedNotWithinTimeout() {
          Flowable<String> other = Flowable.just("a", "b", "c");
          Flowable<String> source = underlyingSubject.timeout(TIMEOUT, TIME_UNIT, testScheduler, other);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          source.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          testScheduler.advanceTimeBy(4, TimeUnit.SECONDS);
          underlyingSubject.onComplete();
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onNext("One");
          inOrder.verify(observer, times(1)).onNext("a");
          inOrder.verify(observer, times(1)).onNext("b");
          inOrder.verify(observer, times(1)).onNext("c");
          inOrder.verify(observer, times(1)).onComplete();
          inOrder.verifyNoMoreInteractions();
          ts.dispose();
      }
  
      @Test
      public void shouldSwitchToOtherAndCanBeUnsubscribedIfOnNextNotWithinTimeout() {
          PublishProcessor<String> other = PublishProcessor.create();
          Flowable<String> source = underlyingSubject.timeout(TIMEOUT, TIME_UNIT, testScheduler, other);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          source.subscribe(ts);
  
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
          underlyingSubject.onNext("One");
          testScheduler.advanceTimeBy(4, TimeUnit.SECONDS);
          underlyingSubject.onNext("Two");
  
          other.onNext("a");
          other.onNext("b");
          ts.dispose();
  
          // The following messages should not be delivered.
          other.onNext("c");
          other.onNext("d");
          other.onComplete();
  
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onNext("One");
          inOrder.verify(observer, times(1)).onNext("a");
          inOrder.verify(observer, times(1)).onNext("b");
          inOrder.verifyNoMoreInteractions();
      }
  
      @Test
      public void shouldTimeoutIfSynchronizedFlowableEmitFirstOnNextNotWithinTimeout()
              throws InterruptedException {
          final CountDownLatch exit = new CountDownLatch(1);
          final CountDownLatch timeoutSetuped = new CountDownLatch(1);
  
          final Subscriber<String> observer = TestHelper.mockSubscriber();
          final TestSubscriber<String> ts = new TestSubscriber<String>(observer);
  
          new Thread(new Runnable() {
  
              @Override
              public void run() {
                  Flowable.unsafeCreate(new Publisher<String>() {
  
                      @Override
                      public void subscribe(Subscriber<? super String> subscriber) {
                          subscriber.onSubscribe(new BooleanSubscription());
                          try {
                              timeoutSetuped.countDown();
                              exit.await();
                          } catch (InterruptedException e) {
                              e.printStackTrace();
                          }
                          subscriber.onNext("a");
                          subscriber.onComplete();
                      }
  
                  }).timeout(1, TimeUnit.SECONDS, testScheduler)
                          .subscribe(ts);
              }
          }).start();
  
          timeoutSetuped.await();
          testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);
  
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onError(isA(TimeoutException.class));
          inOrder.verifyNoMoreInteractions();
  
          exit.countDown(); // exit the thread
      }
  
      @Test
      public void shouldUnsubscribeFromUnderlyingSubscriptionOnTimeout() throws InterruptedException {
          // From https://github.com/ReactiveX/RxJava/pull/951
          final Subscription s = mock(Subscription.class);
  
          Flowable<String> never = Flowable.unsafeCreate(new Publisher<String>() {
              @Override
              public void subscribe(Subscriber<? super String> subscriber) {
                  subscriber.onSubscribe(s);
              }
          });
  
          TestScheduler testScheduler = new TestScheduler();
          Flowable<String> observableWithTimeout = never.timeout(1000, TimeUnit.MILLISECONDS, testScheduler);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          observableWithTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
  
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer).onError(isA(TimeoutException.class));
          inOrder.verifyNoMoreInteractions();
  
          verify(s, times(1)).cancel();
      }
  
      @Test
      @Ignore("s should be considered cancelled upon executing onComplete and not expect downstream to call cancel")
      public void shouldUnsubscribeFromUnderlyingSubscriptionOnImmediatelyComplete() {
          // From https://github.com/ReactiveX/RxJava/pull/951
          final Subscription s = mock(Subscription.class);
  
          Flowable<String> immediatelyComplete = Flowable.unsafeCreate(new Publisher<String>() {
              @Override
              public void subscribe(Subscriber<? super String> subscriber) {
                  subscriber.onSubscribe(s);
                  subscriber.onComplete();
              }
          });
  
          TestScheduler testScheduler = new TestScheduler();
          Flowable<String> observableWithTimeout = immediatelyComplete.timeout(1000, TimeUnit.MILLISECONDS,
                  testScheduler);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          observableWithTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
  
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer).onComplete();
          inOrder.verifyNoMoreInteractions();
  
          verify(s, times(1)).cancel();
      }
  
      @Test
      @Ignore("s should be considered cancelled upon executing onError and not expect downstream to call cancel")
      public void shouldUnsubscribeFromUnderlyingSubscriptionOnImmediatelyErrored() throws InterruptedException {
          // From https://github.com/ReactiveX/RxJava/pull/951
          final Subscription s = mock(Subscription.class);
  
          Flowable<String> immediatelyError = Flowable.unsafeCreate(new Publisher<String>() {
              @Override
              public void subscribe(Subscriber<? super String> subscriber) {
                  subscriber.onSubscribe(s);
                  subscriber.onError(new IOException("Error"));
              }
          });
  
          TestScheduler testScheduler = new TestScheduler();
          Flowable<String> observableWithTimeout = immediatelyError.timeout(1000, TimeUnit.MILLISECONDS,
                  testScheduler);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          TestSubscriber<String> ts = new TestSubscriber<String>(observer);
          observableWithTimeout.subscribe(ts);
  
          testScheduler.advanceTimeBy(2000, TimeUnit.MILLISECONDS);
  
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer).onError(isA(IOException.class));
          inOrder.verifyNoMoreInteractions();
  
          verify(s, times(1)).cancel();
      }
  
      @Test
      public void shouldUnsubscribeFromUnderlyingSubscriptionOnDispose() {
          final PublishProcessor<String> subject = PublishProcessor.create();
          final TestScheduler scheduler = new TestScheduler();
  
          final TestSubscriber<String> observer = subject
                  .timeout(100, TimeUnit.MILLISECONDS, scheduler)
                  .test();
  
          assertTrue(subject.hasSubscribers());
  
          observer.dispose();
  
          assertFalse(subject.hasSubscribers());
      }
  
      @Test
      public void timedAndOther() {
          Flowable.never().timeout(100, TimeUnit.MILLISECONDS, Flowable.just(1))
          .test()
          .awaitDone(5, TimeUnit.SECONDS)
          .assertResult(1);
      }
  
      @Test
      public void disposed() {
          TestHelper.checkDisposed(PublishProcessor.create().timeout(1, TimeUnit.DAYS));
  
          TestHelper.checkDisposed(PublishProcessor.create().timeout(1, TimeUnit.DAYS, Flowable.just(1)));
      }
  
      @Test
      public void timedErrorOther() {
          Flowable.error(new TestException())
          .timeout(1, TimeUnit.DAYS, Flowable.just(1))
          .test()
          .assertFailure(TestException.class);
      }
  
      @Test
      public void timedError() {
          Flowable.error(new TestException())
          .timeout(1, TimeUnit.DAYS)
          .test()
          .assertFailure(TestException.class);
      }
  
      @Test
      public void timedEmptyOther() {
          Flowable.empty()
          .timeout(1, TimeUnit.DAYS, Flowable.just(1))
          .test()
          .assertResult();
      }
  
      @Test
      public void timedEmpty() {
          Flowable.empty()
          .timeout(1, TimeUnit.DAYS)
          .test()
          .assertResult();
      }
  
      @Test
      public void newTimer() {
          FlowableTimeoutTimed.NEW_TIMER.dispose();
          assertTrue(FlowableTimeoutTimed.NEW_TIMER.isDisposed());
      }
  
      @Test
      public void badSource() {
          List<Throwable> errors = TestCommonHelper.trackPluginErrors();
          try {
              new Flowable<Integer>() {
                  @Override
                  protected void subscribeActual(Subscriber<? super Integer> observer) {
                      observer.onSubscribe(new BooleanSubscription());
  
                      observer.onNext(1);
                      observer.onComplete();
                      observer.onNext(2);
                      observer.onError(new TestException());
                      observer.onComplete();
                  }
              }
              .timeout(1, TimeUnit.DAYS)
              .test()
              .assertResult(1);
  
              TestCommonHelper.assertUndeliverable(errors, 0, TestException.class);
          } finally {
              RxJavaCommonPlugins.reset();
          }
      }
  
      @Test
      public void badSourceOther() {
          List<Throwable> errors = TestCommonHelper.trackPluginErrors();
          try {
              new Flowable<Integer>() {
                  @Override
                  protected void subscribeActual(Subscriber<? super Integer> observer) {
                      observer.onSubscribe(new BooleanSubscription());
  
                      observer.onNext(1);
                      observer.onComplete();
                      observer.onNext(2);
                      observer.onError(new TestException());
                      observer.onComplete();
                  }
              }
              .timeout(1, TimeUnit.DAYS, Flowable.just(3))
              .test()
              .assertResult(1);
  
              TestCommonHelper.assertUndeliverable(errors, 0, TestException.class);
          } finally {
              RxJavaCommonPlugins.reset();
          }
      }
  
  
      @Test
      public void timedTake() {
          PublishProcessor<Integer> ps = PublishProcessor.create();
  
          TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS)
          .take(1)
          .test();
  
          assertTrue(ps.hasSubscribers());
  
          ps.onNext(1);
  
          assertFalse(ps.hasSubscribers());
  
          to.assertResult(1);
      }
  
      @Test
      public void timedFallbackTake() {
          PublishProcessor<Integer> ps = PublishProcessor.create();
  
          TestSubscriber<Integer> to = ps.timeout(1, TimeUnit.DAYS, Flowable.just(2))
          .take(1)
          .test();
  
          assertTrue(ps.hasSubscribers());
  
          ps.onNext(1);
  
          assertFalse(ps.hasSubscribers());
  
          to.assertResult(1);
      }
  
  }
  