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.internal.operators.flowable;
  
  import static io.reactivex.rxjava3.core.BackpressureOverflowStrategy.*;
  import static io.reactivex.rxjava3.internal.functions.Functions.EMPTY_ACTION;
  import static org.junit.Assert.*;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.Mockito.*;
  
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.junit.Test;
  import org.reactivestreams.*;
  
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.exceptions.*;
  import io.reactivex.rxjava3.functions.*;
  import io.reactivex.rxjava3.internal.functions.Functions;
  import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription;
  import io.reactivex.rxjava3.processors.PublishProcessor;
  import io.reactivex.rxjava3.subscribers.*;
  import io.reactivex.rxjava3.testsupport.*;
  
  public class FlowableOnBackpressureBufferStrategyTest extends RxJavaTest {
  
      @Test
      public void backpressureWithBufferDropOldest() throws InterruptedException {
          int bufferSize = 3;
          final AtomicInteger droppedCount = new AtomicInteger(0);
          Action incrementOnDrop = new Action() {
              @Override
              public void run() throws Exception {
                  droppedCount.incrementAndGet();
              }
          };
          TestSubscriber<Long> ts = createTestSubscriber();
          Flowable.fromPublisher(send500ValuesAndComplete.onBackpressureBuffer(bufferSize, incrementOnDrop, DROP_OLDEST))
                  .subscribe(ts);
          // we request 10 but only 3 should come from the buffer
          ts.request(10);
          ts.awaitDone(5, TimeUnit.SECONDS);
          assertEquals(bufferSize, ts.values().size());
          ts.assertNoErrors();
          assertEquals(497, ts.values().get(0).intValue());
          assertEquals(498, ts.values().get(1).intValue());
          assertEquals(499, ts.values().get(2).intValue());
          assertEquals(droppedCount.get(), 500 - bufferSize);
      }
  
      private TestSubscriber<Long> createTestSubscriber() {
          return new TestSubscriber<>(new DefaultSubscriber<Long>() {
  
              @Override
              protected void onStart() {
              }
  
              @Override
              public void onComplete() {
              }
  
              @Override
              public void onError(Throwable e) {
              }
  
              @Override
              public void onNext(Long t) {
              }
  
          }, 0L);
      }
  
      @Test
      public void backpressureWithBufferDropLatest() throws InterruptedException {
          int bufferSize = 3;
          final AtomicInteger droppedCount = new AtomicInteger(0);
          Action incrementOnDrop = new Action() {
              @Override
              public void run() throws Exception {
                  droppedCount.incrementAndGet();
              }
          };
          TestSubscriber<Long> ts = createTestSubscriber();
          Flowable.fromPublisher(send500ValuesAndComplete.onBackpressureBuffer(bufferSize, incrementOnDrop, DROP_LATEST))
                  .subscribe(ts);
          // we request 10 but only 3 should come from the buffer
          ts.request(10);
          ts.awaitDone(5, TimeUnit.SECONDS);
          assertEquals(bufferSize, ts.values().size());
          ts.assertNoErrors();
          assertEquals(0, ts.values().get(0).intValue());
          assertEquals(1, ts.values().get(1).intValue());
          assertEquals(499, ts.values().get(2).intValue());
          assertEquals(droppedCount.get(), 500 - bufferSize);
      }
  
      private static final Flowable<Long> send500ValuesAndComplete = Flowable.unsafeCreate(new Publisher<Long>() {
          @Override
          public void subscribe(Subscriber<? super Long> s) {
              BooleanSubscription bs = new BooleanSubscription();
              s.onSubscribe(bs);
              long i = 0;
              while (!bs.isCancelled() && i < 500) {
                  s.onNext(i++);
              }
              if (!bs.isCancelled()) {
                  s.onComplete();
              }
          }
      });
  
      @Test(expected = IllegalArgumentException.class)
      public void backpressureBufferNegativeCapacity() throws InterruptedException {
          Flowable.empty().onBackpressureBuffer(-1, EMPTY_ACTION , DROP_OLDEST);
      }
  
      @Test(expected = IllegalArgumentException.class)
      public void backpressureBufferZeroCapacity() throws InterruptedException {
          Flowable.empty().onBackpressureBuffer(0, EMPTY_ACTION , DROP_OLDEST);
      }
  
      @Test
      public void dispose() {
          TestHelper.checkDisposed(Flowable.just(1)
                  .onBackpressureBuffer(16, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR));
      }
  
      @Test
      public void error() {
          Flowable
          .error(new TestException())
          .onBackpressureBuffer(16, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR)
          .test()
          .assertFailure(TestException.class);
      }
  
      @Test
      public void overflowError() {
          Flowable.range(1, 20)
          .onBackpressureBuffer(8, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR)
          .test(0L)
          .assertFailure(MissingBackpressureException.class);
      }
  
      @Test
      public void badSource() {
          TestHelper.checkBadSourceFlowable(new Function<Flowable<Object>, Object>() {
              @Override
              public Object apply(Flowable<Object> f) throws Exception {
                  return f.onBackpressureBuffer(8, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR);
              }
          }, false, 1, 1, 1);
      }
  
      @Test
      public void doubleOnSubscribe() {
          TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Object>>() {
              @Override
              public Flowable<Object> apply(Flowable<Object> f) throws Exception {
                  return f.onBackpressureBuffer(8, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR);
              }
          });
      }
  
      @Test
      public void overflowCrashes() {
          Flowable.range(1, 20)
          .onBackpressureBuffer(8, new Action() {
              @Override
              public void run() throws Exception {
                  throw new TestException();
              }
          }, BackpressureOverflowStrategy.DROP_OLDEST)
          .test(0L)
          .assertFailure(TestException.class);
      }
  
      @Test
      public void badRequest() {
          TestHelper.assertBadRequestReported(Flowable.just(1)
                  .onBackpressureBuffer(16, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR));
      }
  
      @Test
      public void empty() {
          Flowable.empty()
          .onBackpressureBuffer(16, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR)
          .test(0L)
          .assertResult();
      }
  
      @Test
      public void justTake() {
          Flowable.just(1)
          .onBackpressureBuffer(16, Functions.EMPTY_ACTION, BackpressureOverflowStrategy.ERROR)
          .take(1)
          .test()
          .assertResult(1);
      }
  
      @Test
      public void overflowNullAction() {
          Flowable.range(1, 5)
          .onBackpressureBuffer(1, null, BackpressureOverflowStrategy.DROP_OLDEST)
          .test(0L)
          .assertEmpty();
      }
  
      @Test
      public void cancelOnDrain() {
          Flowable.range(1, 5)
          .onBackpressureBuffer(10, null, BackpressureOverflowStrategy.DROP_OLDEST)
          .takeUntil(v -> true)
          .test(0L)
          .assertEmpty()
          .requestMore(10)
          .assertResult(1);
      }
  
      @Test
      public void onDroppedNormalDropOldest() throws Throwable {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          @SuppressWarnings("unchecked")
          Consumer<Integer> onDropped = mock(Consumer.class);
  
          TestSubscriber<Integer> ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.DROP_OLDEST, onDropped)
          .test(0L);
  
          ts.assertEmpty();
  
          pp.onNext(1);
  
          ts.assertEmpty();
          verify(onDropped, never()).accept(any());
  
          pp.onNext(2);
  
          ts.assertEmpty();
  
          verify(onDropped).accept(1);
      }
  
      @Test
      public void onDroppedNormalDropLatest() throws Throwable {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          @SuppressWarnings("unchecked")
          Consumer<Integer> onDropped = mock(Consumer.class);
  
          TestSubscriber<Integer> ts = pp.onBackpressureBuffer(2, null, BackpressureOverflowStrategy.DROP_LATEST, onDropped)
          .test(0L);
  
          ts.assertEmpty();
  
          pp.onNext(1);
  
          pp.onNext(2);
  
          ts.assertEmpty();
          verify(onDropped, never()).accept(any());
  
          pp.onNext(3);
  
          ts.assertEmpty();
  
          verify(onDropped).accept(2);
      }
  
      @Test
      public void onDroppedNormalError() throws Throwable {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          @SuppressWarnings("unchecked")
          Consumer<Integer> onDropped = mock(Consumer.class);
  
          TestSubscriber<Integer> ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.ERROR, onDropped)
          .test(0L);
  
          ts.assertEmpty();
  
          pp.onNext(1);
  
          ts.assertEmpty();
          verify(onDropped, never()).accept(any());
  
          pp.onNext(2);
  
          ts.assertFailure(MissingBackpressureException.class);
  
          verify(onDropped).accept(2);
      }
  
      @Test
      public void onDroppedCrash() throws Throwable {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          Consumer<Integer> onDropped = v -> { throw new TestException(); };
  
          TestSubscriberEx<Integer> ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.DROP_OLDEST, onDropped)
          .subscribeWith(new TestSubscriberEx<Integer>(0L));
  
          ts.assertEmpty();
  
          pp.onNext(1);
  
          ts.assertEmpty();
  
          pp.onNext(2);
  
          ts.assertFailure(TestException.class);
      }
  }
  