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.util.Arrays;
  import java.util.concurrent.*;
  import java.util.concurrent.atomic.*;
  
  import org.junit.*;
  import org.mockito.InOrder;
  import org.reactivestreams.*;
  
  import io.reactivex.common.Schedulers;
  import io.reactivex.common.exceptions.TestException;
  import io.reactivex.common.functions.*;
  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 FlowableTakeTest {
  
      @Test
      public void testTake1() {
          Flowable<String> w = Flowable.fromIterable(Arrays.asList("one", "two", "three"));
          Flowable<String> take = w.take(2);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          take.subscribe(observer);
          verify(observer, times(1)).onNext("one");
          verify(observer, times(1)).onNext("two");
          verify(observer, never()).onNext("three");
          verify(observer, never()).onError(any(Throwable.class));
          verify(observer, times(1)).onComplete();
      }
  
      @Test
      public void testTake2() {
          Flowable<String> w = Flowable.fromIterable(Arrays.asList("one", "two", "three"));
          Flowable<String> take = w.take(1);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
          take.subscribe(observer);
          verify(observer, times(1)).onNext("one");
          verify(observer, never()).onNext("two");
          verify(observer, never()).onNext("three");
          verify(observer, never()).onError(any(Throwable.class));
          verify(observer, times(1)).onComplete();
      }
  
      @Test(expected = IllegalArgumentException.class)
      public void testTakeWithError() {
          Flowable.fromIterable(Arrays.asList(1, 2, 3)).take(1)
          .map(new Function<Integer, Integer>() {
              @Override
              public Integer apply(Integer t1) {
                  throw new IllegalArgumentException("some error");
              }
          }).blockingSingle();
      }
  
      @Test
      public void testTakeWithErrorHappeningInOnNext() {
          Flowable<Integer> w = Flowable.fromIterable(Arrays.asList(1, 2, 3))
                  .take(2).map(new Function<Integer, Integer>() {
              @Override
              public Integer apply(Integer t1) {
                  throw new IllegalArgumentException("some error");
              }
          });
  
          Subscriber<Integer> observer = TestHelper.mockSubscriber();
          w.subscribe(observer);
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onError(any(IllegalArgumentException.class));
          inOrder.verifyNoMoreInteractions();
      }
  
      @Test
      public void testTakeWithErrorHappeningInTheLastOnNext() {
          Flowable<Integer> w = Flowable.fromIterable(Arrays.asList(1, 2, 3)).take(1).map(new Function<Integer, Integer>() {
              @Override
              public Integer apply(Integer t1) {
                  throw new IllegalArgumentException("some error");
              }
          });
  
          Subscriber<Integer> observer = TestHelper.mockSubscriber();
          w.subscribe(observer);
          InOrder inOrder = inOrder(observer);
          inOrder.verify(observer, times(1)).onError(any(IllegalArgumentException.class));
          inOrder.verifyNoMoreInteractions();
      }
  
      @Test
      public void testTakeDoesntLeakErrors() {
          Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
              @Override
              public void subscribe(Subscriber<? super String> observer) {
                  observer.onSubscribe(new BooleanSubscription());
                  observer.onNext("one");
                  observer.onError(new Throwable("test failed"));
              }
          });
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
  
          source.take(1).subscribe(observer);
  
          verify(observer).onSubscribe((Subscription)notNull());
          verify(observer, times(1)).onNext("one");
          // even though onError is called we take(1) so shouldn't see it
          verify(observer, never()).onError(any(Throwable.class));
          verify(observer, times(1)).onComplete();
          verifyNoMoreInteractions(observer);
      }
  
      @Test
      @Ignore("take(0) is now empty() and doesn't even subscribe to the original source")
      public void testTakeZeroDoesntLeakError() {
          final AtomicBoolean subscribed = new AtomicBoolean(false);
          final BooleanSubscription bs = new BooleanSubscription();
          Flowable<String> source = Flowable.unsafeCreate(new Publisher<String>() {
              @Override
              public void subscribe(Subscriber<? super String> observer) {
                  subscribed.set(true);
                  observer.onSubscribe(bs);
                  observer.onError(new Throwable("test failed"));
              }
          });
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
  
          source.take(0).subscribe(observer);
          assertTrue("source subscribed", subscribed.get());
          assertTrue("source unsubscribed", bs.isCancelled());
  
          verify(observer, never()).onNext(anyString());
          // even though onError is called we take(0) so shouldn't see it
          verify(observer, never()).onError(any(Throwable.class));
          verify(observer, times(1)).onComplete();
          verifyNoMoreInteractions(observer);
      }
  
      @Test
      public void testUnsubscribeAfterTake() {
          TestFlowableFunc f = new TestFlowableFunc("one", "two", "three");
          Flowable<String> w = Flowable.unsafeCreate(f);
  
          Subscriber<String> observer = TestHelper.mockSubscriber();
  
          Flowable<String> take = w.take(1);
          take.subscribe(observer);
  
          // wait for the Flowable to complete
          try {
              f.t.join();
          } catch (Throwable e) {
              e.printStackTrace();
              fail(e.getMessage());
          }
  
          System.out.println("TestFlowable thread finished");
          verify(observer).onSubscribe((Subscription)notNull());
          verify(observer, times(1)).onNext("one");
          verify(observer, never()).onNext("two");
          verify(observer, never()).onNext("three");
          verify(observer, times(1)).onComplete();
          // FIXME no longer assertable
  //        verify(s, times(1)).unsubscribe();
          verifyNoMoreInteractions(observer);
      }
  
      @Test(timeout = 2000)
      public void testUnsubscribeFromSynchronousInfiniteFlowable() {
          final AtomicLong count = new AtomicLong();
          INFINITE_OBSERVABLE.take(10).subscribe(new Consumer<Long>() {
  
              @Override
              public void accept(Long l) {
                  count.set(l);
              }
  
          });
          assertEquals(10, count.get());
      }
  
      @Test(timeout = 2000)
      public void testMultiTake() {
          final AtomicInteger count = new AtomicInteger();
          Flowable.unsafeCreate(new Publisher<Integer>() {
  
              @Override
              public void subscribe(Subscriber<? super Integer> s) {
                  BooleanSubscription bs = new BooleanSubscription();
                  s.onSubscribe(bs);
                  for (int i = 0; !bs.isCancelled(); i++) {
                      System.out.println("Emit: " + i);
                      count.incrementAndGet();
                      s.onNext(i);
                  }
              }
  
          }).take(100).take(1).blockingForEach(new Consumer<Integer>() {
  
              @Override
              public void accept(Integer t1) {
                  System.out.println("Receive: " + t1);
  
              }
  
          });
  
          assertEquals(1, count.get());
      }
  
      static class TestFlowableFunc implements Publisher<String> {
  
          final String[] values;
          Thread t;
  
          TestFlowableFunc(String... values) {
              this.values = values;
          }
  
          @Override
          public void subscribe(final Subscriber<? super String> observer) {
              observer.onSubscribe(new BooleanSubscription());
              System.out.println("TestFlowable subscribed to ...");
              t = new Thread(new Runnable() {
  
                  @Override
                  public void run() {
                      try {
                          System.out.println("running TestFlowable thread");
                          for (String s : values) {
                              System.out.println("TestFlowable onNext: " + s);
                              observer.onNext(s);
                          }
                          observer.onComplete();
                      } catch (Throwable e) {
                          throw new RuntimeException(e);
                      }
                  }
  
              });
              System.out.println("starting TestFlowable thread");
              t.start();
              System.out.println("done starting TestFlowable thread");
          }
      }
  
      private static Flowable<Long> INFINITE_OBSERVABLE = Flowable.unsafeCreate(new Publisher<Long>() {
  
          @Override
          public void subscribe(Subscriber<? super Long> op) {
              BooleanSubscription bs = new BooleanSubscription();
              op.onSubscribe(bs);
              long l = 1;
              while (!bs.isCancelled()) {
                  op.onNext(l++);
              }
              op.onComplete();
          }
  
      });
  
      @Test(timeout = 2000)
      public void testTakeObserveOn() {
          Subscriber<Object> o = TestHelper.mockSubscriber();
          TestSubscriber<Object> ts = new TestSubscriber<Object>(o);
  
          INFINITE_OBSERVABLE.onBackpressureDrop()
          .observeOn(Schedulers.newThread()).take(1).subscribe(ts);
          ts.awaitTerminalEvent();
          ts.assertNoErrors();
  
          verify(o).onNext(1L);
          verify(o, never()).onNext(2L);
          verify(o).onComplete();
          verify(o, never()).onError(any(Throwable.class));
      }
  
      @Test
      public void testProducerRequestThroughTake() {
          TestSubscriber<Integer> ts = new TestSubscriber<Integer>(3);
          final AtomicLong requested = new AtomicLong();
          Flowable.unsafeCreate(new Publisher<Integer>() {
  
              @Override
              public void subscribe(Subscriber<? super Integer> s) {
                  s.onSubscribe(new Subscription() {
  
                      @Override
                      public void request(long n) {
                          requested.set(n);
                      }
  
                      @Override
                      public void cancel() {
  
                      }
                  });
              }
  
          }).take(3).subscribe(ts);
          assertEquals(Long.MAX_VALUE, requested.get());
      }
  
      @Test
      public void testProducerRequestThroughTakeIsModified() {
          TestSubscriber<Integer> ts = new TestSubscriber<Integer>(3);
          final AtomicLong requested = new AtomicLong();
          Flowable.unsafeCreate(new Publisher<Integer>() {
  
              @Override
              public void subscribe(Subscriber<? super Integer> s) {
                  s.onSubscribe(new Subscription() {
  
                      @Override
                      public void request(long n) {
                          requested.set(n);
                      }
  
                      @Override
                      public void cancel() {
  
                      }
                  });
              }
  
          }).take(1).subscribe(ts);
          //FIXME take triggers fast path if downstream requests more than the limit
          assertEquals(Long.MAX_VALUE, requested.get());
      }
  
      @Test
      public void testInterrupt() throws InterruptedException {
          final AtomicReference<Object> exception = new AtomicReference<Object>();
          final CountDownLatch latch = new CountDownLatch(1);
          Flowable.just(1).subscribeOn(Schedulers.computation()).take(1)
          .subscribe(new Consumer<Integer>() {
  
              @Override
              public void accept(Integer t1) {
                  try {
                      Thread.sleep(100);
                  } catch (Exception e) {
                      exception.set(e);
                      e.printStackTrace();
                  } finally {
                      latch.countDown();
                  }
              }
  
          });
  
          latch.await();
          assertNull(exception.get());
      }
  
      @Test
      public void testDoesntRequestMoreThanNeededFromUpstream() throws InterruptedException {
          final AtomicLong requests = new AtomicLong();
          TestSubscriber<Long> ts = new TestSubscriber<Long>(0L);
          Flowable.interval(100, TimeUnit.MILLISECONDS)
              //
              .doOnRequest(new LongConsumer() {
                  @Override
                  public void accept(long n) {
                      System.out.println(n);
                      requests.addAndGet(n);
              }})
              //
              .take(2)
              //
              .subscribe(ts);
          Thread.sleep(50);
          ts.request(1);
          ts.request(1);
          ts.request(1);
          ts.awaitTerminalEvent();
          ts.assertComplete();
          ts.assertNoErrors();
          assertEquals(3, requests.get());
      }
  
      @Test
      public void takeFinalValueThrows() {
          Flowable<Integer> source = Flowable.just(1).take(1);
  
          TestSubscriber<Integer> ts = new TestSubscriber<Integer>() {
              @Override
              public void onNext(Integer t) {
                  throw new TestException();
              }
          };
  
          source.safeSubscribe(ts);
  
          ts.assertNoValues();
          ts.assertError(TestException.class);
          ts.assertNotComplete();
      }
  
      @Test
      public void testReentrantTake() {
          final PublishProcessor<Integer> source = PublishProcessor.create();
  
          TestSubscriber<Integer> ts = new TestSubscriber<Integer>();
  
          source.take(1).doOnNext(new Consumer<Integer>() {
              @Override
              public void accept(Integer v) {
                  source.onNext(2);
              }
          }).subscribe(ts);
  
          source.onNext(1);
  
          ts.assertValue(1);
          ts.assertNoErrors();
          ts.assertComplete();
      }
  
  
      @Test
      public void takeNegative() {
          try {
              Flowable.just(1).take(-99);
              fail("Should have thrown");
          } catch (IllegalArgumentException ex) {
              assertEquals("count >= 0 required but it was -99", ex.getMessage());
          }
      }
  
      @Test
      public void takeZero() {
          Flowable.just(1)
          .take(0)
          .test()
          .assertResult();
      }
  
      @Test
      public void dispose() {
          TestHelper.checkDisposed(PublishProcessor.create().take(2));
      }
  
      @Test
      public void doubleOnSubscribe() {
          TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Object>>() {
              @Override
              public Flowable<Object> apply(Flowable<Object> o) throws Exception {
                  return o.take(2);
              }
          });
      }
  
  }
  