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 org.junit.Assert.*;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.Mockito.*;
  
  import java.util.*;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.*;
  
  import org.junit.Test;
  import org.reactivestreams.*;
  
  import io.reactivex.rxjava3.core.*;
  import io.reactivex.rxjava3.core.Flowable;
  import io.reactivex.rxjava3.exceptions.TestException;
  import io.reactivex.rxjava3.functions.*;
  import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription;
  import io.reactivex.rxjava3.processors.PublishProcessor;
  import io.reactivex.rxjava3.schedulers.Schedulers;
  import io.reactivex.rxjava3.subscribers.*;
  import io.reactivex.rxjava3.testsupport.*;
  
  public class FlowableWindowWithSizeTest extends RxJavaTest {
  
      private static <T> List<List<T>> toLists(Flowable<Flowable<T>> observables) {
  
          return observables.flatMapSingle(new Function<Flowable<T>, SingleSource<List<T>>>() {
              @Override
              public SingleSource<List<T>> apply(Flowable<T> w) throws Throwable {
                  return w.toList();
              }
          }).toList().blockingGet();
      }
  
      @Test
      public void nonOverlappingWindows() {
          Flowable<String> subject = Flowable.just("one", "two", "three", "four", "five");
          Flowable<Flowable<String>> windowed = subject.window(3);
  
          List<List<String>> windows = toLists(windowed);
  
          assertEquals(2, windows.size());
          assertEquals(list("one", "two", "three"), windows.get(0));
          assertEquals(list("four", "five"), windows.get(1));
      }
  
      @Test
      public void skipAndCountGaplessWindows() {
          Flowable<String> subject = Flowable.just("one", "two", "three", "four", "five");
          Flowable<Flowable<String>> windowed = subject.window(3, 3);
  
          List<List<String>> windows = toLists(windowed);
  
          assertEquals(2, windows.size());
          assertEquals(list("one", "two", "three"), windows.get(0));
          assertEquals(list("four", "five"), windows.get(1));
      }
  
      @Test
      public void overlappingWindows() {
          Flowable<String> subject = Flowable.fromArray(new String[] { "zero", "one", "two", "three", "four", "five" });
          Flowable<Flowable<String>> windowed = subject.window(3, 1);
  
          List<List<String>> windows = toLists(windowed);
  
          assertEquals(6, windows.size());
          assertEquals(list("zero", "one", "two"), windows.get(0));
          assertEquals(list("one", "two", "three"), windows.get(1));
          assertEquals(list("two", "three", "four"), windows.get(2));
          assertEquals(list("three", "four", "five"), windows.get(3));
          assertEquals(list("four", "five"), windows.get(4));
          assertEquals(list("five"), windows.get(5));
      }
  
      @Test
      public void skipAndCountWindowsWithGaps() {
          Flowable<String> subject = Flowable.just("one", "two", "three", "four", "five");
          Flowable<Flowable<String>> windowed = subject.window(2, 3);
  
          List<List<String>> windows = toLists(windowed);
  
          assertEquals(2, windows.size());
          assertEquals(list("one", "two"), windows.get(0));
          assertEquals(list("four", "five"), windows.get(1));
      }
  
      @Test
      public void windowUnsubscribeNonOverlapping() {
          TestSubscriberEx<Integer> ts = new TestSubscriberEx<>();
  
          final AtomicInteger count = new AtomicInteger();
          Flowable.merge(Flowable.range(1, 10000).doOnNext(new Consumer<Integer>() {
  
              @Override
              public void accept(Integer t1) {
                  count.incrementAndGet();
              }
  
          }).window(5).take(2))
          .subscribe(ts);
  
          ts.awaitDone(500, TimeUnit.MILLISECONDS);
          ts.assertTerminated();
          ts.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
          //        System.out.println(ts.getOnNextEvents());
          assertEquals(10, count.get());
      }
  
      @Test
      public void windowUnsubscribeNonOverlappingAsyncSource() {
          TestSubscriberEx<Integer> ts = new TestSubscriberEx<>();
          final AtomicInteger count = new AtomicInteger();
          Flowable.merge(Flowable.range(1, 100000)
                  .doOnNext(new Consumer<Integer>() {
  
                      @Override
                      public void accept(Integer t1) {
                          count.incrementAndGet();
                      }
  
                  })
                  .observeOn(Schedulers.computation())
                  .window(5)
                  .take(2))
                  .subscribe(ts);
          ts.awaitDone(500, TimeUnit.MILLISECONDS);
          ts.assertTerminated();
          ts.assertValues(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
          // make sure we don't emit all values ... the unsubscribe should propagate
          assertTrue(count.get() < 100000);
      }
  
      @Test
      public void windowUnsubscribeOverlapping() {
          TestSubscriberEx<Integer> ts = new TestSubscriberEx<>();
          final AtomicInteger count = new AtomicInteger();
          Flowable.merge(Flowable.range(1, 10000).doOnNext(new Consumer<Integer>() {
  
              @Override
              public void accept(Integer t1) {
                  count.incrementAndGet();
              }
  
          }).window(5, 4).take(2)).subscribe(ts);
          ts.awaitDone(500, TimeUnit.MILLISECONDS);
          ts.assertTerminated();
          //        System.out.println(ts.getOnNextEvents());
          ts.assertValues(1, 2, 3, 4, 5, 5, 6, 7, 8, 9);
          assertEquals(9, count.get());
      }
  
      @Test
      public void windowUnsubscribeOverlappingAsyncSource() {
          TestSubscriberEx<Integer> ts = new TestSubscriberEx<>();
          final AtomicInteger count = new AtomicInteger();
          Flowable.merge(Flowable.range(1, 100000)
                  .doOnNext(new Consumer<Integer>() {
  
                      @Override
                      public void accept(Integer t1) {
                          count.incrementAndGet();
                      }
  
                  })
                  .observeOn(Schedulers.computation())
                  .window(5, 4)
                  .take(2), 128)
                  .subscribe(ts);
          ts.awaitDone(500, TimeUnit.MILLISECONDS);
          ts.assertTerminated();
          ts.assertValues(1, 2, 3, 4, 5, 5, 6, 7, 8, 9);
          // make sure we don't emit all values ... the unsubscribe should propagate
          // assertTrue(count.get() < 100000); // disabled: a small hiccup in the consumption may allow the source to run to completion
      }
  
      private List<String> list(String... args) {
          List<String> list = new ArrayList<>();
          for (String arg : args) {
              list.add(arg);
          }
          return list;
      }
  
      @Test
      public void backpressureOuter() {
          Flowable<Flowable<Integer>> source = Flowable.range(1, 10).window(3);
  
          final List<Integer> list = new ArrayList<>();
  
          final Subscriber<Integer> subscriber = TestHelper.mockSubscriber();
  
          source.subscribe(new DefaultSubscriber<Flowable<Integer>>() {
              @Override
              public void onStart() {
                  request(1);
              }
  
              @Override
              public void onNext(Flowable<Integer> t) {
                  t.subscribe(new DefaultSubscriber<Integer>() {
                      @Override
                      public void onNext(Integer t) {
                          list.add(t);
                      }
  
                      @Override
                      public void onError(Throwable e) {
                          subscriber.onError(e);
                      }
  
                      @Override
                      public void onComplete() {
                          subscriber.onComplete();
                      }
                  });
              }
  
              @Override
              public void onError(Throwable e) {
                  subscriber.onError(e);
              }
  
              @Override
              public void onComplete() {
                  subscriber.onComplete();
              }
          });
  
          assertEquals(Arrays.asList(1, 2, 3), list);
  
          verify(subscriber, never()).onError(any(Throwable.class));
          verify(subscriber, times(1)).onComplete(); // 1 inner
      }
  
      public static Flowable<Integer> hotStream() {
          return Flowable.unsafeCreate(new Publisher<Integer>() {
              @Override
              public void subscribe(Subscriber<? super Integer> s) {
                  BooleanSubscription bs = new BooleanSubscription();
                  s.onSubscribe(bs);
                  while (!bs.isCancelled()) {
                      // burst some number of items
                      for (int i = 0; i < Math.random() * 20; i++) {
                          s.onNext(i);
                      }
                      try {
                          // sleep for a random amount of time
                          // NOTE: Only using Thread.sleep here as an artificial demo.
                          Thread.sleep((long) (Math.random() * 200));
                      } catch (Exception e) {
                          // do nothing
                      }
                  }
                  System.out.println("Hot done.");
              }
          }).subscribeOn(Schedulers.newThread()); // use newThread since we are using sleep to block
      }
  
      @Test
      public void takeFlatMapCompletes() {
          TestSubscriber<Integer> ts = new TestSubscriber<>();
  
          final int indicator = 999999999;
  
          hotStream()
          .window(10)
          .take(2)
          .flatMap(new Function<Flowable<Integer>, Flowable<Integer>>() {
              @Override
              public Flowable<Integer> apply(Flowable<Integer> w) {
                  return w.startWithItem(indicator);
              }
          }).subscribe(ts);
  
          ts.awaitDone(2, TimeUnit.SECONDS);
          ts.assertComplete();
          ts.assertValueCount(22);
      }
  
      @Test
      public void backpressureOuterInexact() {
          TestSubscriber<List<Integer>> ts = new TestSubscriber<>(0L);
  
          Flowable.range(1, 5)
          .window(2, 1)
          .map(new Function<Flowable<Integer>, Flowable<List<Integer>>>() {
              @Override
              public Flowable<List<Integer>> apply(Flowable<Integer> t) {
                  return t.toList().toFlowable();
              }
          })
          .concatMapEager(new Function<Flowable<List<Integer>>, Publisher<List<Integer>>>() {
              @Override
              public Publisher<List<Integer>> apply(Flowable<List<Integer>> v) {
                  return v;
              }
          })
          .subscribe(ts);
  
          ts.assertNoErrors();
          ts.assertNoValues();
          ts.assertNotComplete();
  
          ts.request(2);
  
          ts.assertValues(Arrays.asList(1, 2), Arrays.asList(2, 3));
          ts.assertNoErrors();
          ts.assertNotComplete();
  
          ts.request(5);
  
          System.out.println(ts.values());
  
          ts.assertValues(Arrays.asList(1, 2), Arrays.asList(2, 3),
                  Arrays.asList(3, 4), Arrays.asList(4, 5), Arrays.asList(5));
          ts.assertNoErrors();
          ts.assertComplete();
      }
  
      @Test
      public void dispose() {
          TestHelper.checkDisposed(PublishProcessor.create().window(1));
  
          TestHelper.checkDisposed(PublishProcessor.create().window(2, 1));
  
          TestHelper.checkDisposed(PublishProcessor.create().window(1, 2));
      }
  
      @Test
      public void doubleOnSubscribe() {
          TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Flowable<Object>>>() {
              @Override
              public Flowable<Flowable<Object>> apply(Flowable<Object> f) throws Exception {
                  return f.window(1);
              }
          });
  
          TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Flowable<Object>>>() {
              @Override
              public Flowable<Flowable<Object>> apply(Flowable<Object> f) throws Exception {
                  return f.window(2, 1);
              }
          });
  
          TestHelper.checkDoubleOnSubscribeFlowable(new Function<Flowable<Object>, Flowable<Flowable<Object>>>() {
              @Override
              public Flowable<Flowable<Object>> apply(Flowable<Object> f) throws Exception {
                  return f.window(1, 2);
              }
          });
      }
  
      @Test
      public void errorExact() {
          Flowable.error(new TestException())
          .window(1)
          .test()
          .assertFailure(TestException.class);
      }
  
      @Test
      public void errorSkip() {
          Flowable.error(new TestException())
          .window(1, 2)
          .test()
          .assertFailure(TestException.class);
      }
  
      @Test
      public void errorOverlap() {
          Flowable.error(new TestException())
          .window(2, 1)
          .test()
          .assertFailure(TestException.class);
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void errorExactInner() {
          @SuppressWarnings("rawtypes")
          final TestSubscriber[] to = { null };
          Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
          .window(2)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> w) throws Exception {
                  to[0] = w.test();
              }
          })
          .test()
          .assertError(TestException.class);
  
          to[0].assertFailure(TestException.class, 1);
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void errorSkipInner() {
          @SuppressWarnings("rawtypes")
          final TestSubscriber[] to = { null };
          Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
          .window(2, 3)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> w) throws Exception {
                  to[0] = w.test();
              }
          })
          .test()
          .assertError(TestException.class);
  
          to[0].assertFailure(TestException.class, 1);
      }
  
      @SuppressWarnings("unchecked")
      @Test
      public void errorOverlapInner() {
          @SuppressWarnings("rawtypes")
          final TestSubscriber[] to = { null };
          Flowable.just(1).concatWith(Flowable.<Integer>error(new TestException()))
          .window(3, 2)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> w) throws Exception {
                  to[0] = w.test();
              }
          })
          .test()
          .assertError(TestException.class);
  
          to[0].assertFailure(TestException.class, 1);
      }
  
      @Test
      public void cancellingWindowCancelsUpstreamSize() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Integer> ts = pp.window(10)
          .take(1)
          .flatMap(new Function<Flowable<Integer>, Publisher<Integer>>() {
              @Override
              public Publisher<Integer> apply(Flowable<Integer> w) throws Throwable {
                  return w.take(1);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertResult(1);
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void windowAbandonmentCancelsUpstreamSize() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          final AtomicReference<Flowable<Integer>> inner = new AtomicReference<>();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10)
          .take(1)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> v) throws Throwable {
                  inner.set(v);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertValueCount(1)
          .assertNoErrors()
          .assertComplete();
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
  
          inner.get().test().assertResult(1);
      }
  
      @Test
      public void cancellingWindowCancelsUpstreamSkip() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Integer> ts = pp.window(5, 10)
          .take(1)
          .flatMap(new Function<Flowable<Integer>, Publisher<Integer>>() {
              @Override
              public Publisher<Integer> apply(Flowable<Integer> w) throws Throwable {
                  return w.take(1);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertResult(1);
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void windowAbandonmentCancelsUpstreamSkip() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          final AtomicReference<Flowable<Integer>> inner = new AtomicReference<>();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(5, 10)
          .take(1)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> v) throws Throwable {
                  inner.set(v);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertValueCount(1)
          .assertNoErrors()
          .assertComplete();
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
  
          inner.get().test().assertResult(1);
      }
  
      @Test
      public void cancellingWindowCancelsUpstreamOverlap() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Integer> ts = pp.window(5, 3)
          .take(1)
          .flatMap(new Function<Flowable<Integer>, Publisher<Integer>>() {
              @Override
              public Publisher<Integer> apply(Flowable<Integer> w) throws Throwable {
                  return w.take(1);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertResult(1);
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void windowAbandonmentCancelsUpstreamOverlap() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          final AtomicReference<Flowable<Integer>> inner = new AtomicReference<>();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(5, 3)
          .take(1)
          .doOnNext(new Consumer<Flowable<Integer>>() {
              @Override
              public void accept(Flowable<Integer> v) throws Throwable {
                  inner.set(v);
              }
          })
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts
          .assertValueCount(1)
          .assertNoErrors()
          .assertComplete();
  
          assertFalse("Processor still has subscribers!", pp.hasSubscribers());
  
          inner.get().test().assertResult(1);
      }
  
      @Test
      public void badRequestExact() {
          TestHelper.assertBadRequestReported(Flowable.never().window(1));
      }
  
      @Test
      public void badRequestSkip() {
          TestHelper.assertBadRequestReported(Flowable.never().window(1, 2));
      }
  
      @Test
      public void badRequestOverlap() {
          TestHelper.assertBadRequestReported(Flowable.never().window(2, 1));
      }
  
      @Test
      public void skipEmpty() {
          Flowable.empty()
          .window(1, 2)
          .test()
          .assertResult();
      }
  
      @Test
      public void exactEmpty() {
          Flowable.empty()
          .window(2)
          .test()
          .assertResult();
      }
  
      @Test
      public void skipMultipleRequests() {
          Flowable.range(1, 10)
          .window(1, 2)
          .doOnNext(w -> w.test())
          .rebatchRequests(1)
          .test()
          .assertComplete();
      }
  
      @Test
      public void skipOne() {
          Flowable.just(1)
          .window(2, 3)
          .flatMap(v -> v)
          .test()
          .assertResult(1);
      }
  
      @Test
      public void overlapMultipleRequests() {
          Flowable.range(1, 10)
          .window(2, 1)
          .doOnNext(w -> w.test())
          .rebatchRequests(1)
          .test()
          .assertComplete();
      }
  
      @Test
      public void overlapCancelAfterWindow() {
          Flowable.range(1, 10)
          .window(2, 1)
          .takeUntil(v -> true)
          .doOnNext(w -> w.test())
          .test(0L)
          .requestMore(10)
          .assertComplete();
      }
  
      @Test
      public void overlapEmpty() {
          Flowable.empty()
          .window(2, 1)
          .test()
          .assertResult();
      }
  
      @Test
      public void overlapEmptyNoRequest() {
          Flowable.empty()
          .window(2, 1)
          .test(0L)
          .assertResult();
      }
  
      @Test
      public void overlapMoreWorkAfterOnNext() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
          AtomicBoolean once = new AtomicBoolean();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(2, 1)
          .doOnNext(v -> {
              v.test();
              if (once.compareAndSet(false, true)) {
                  pp.onNext(2);
                  pp.onComplete();
              }
          })
          .test();
  
          pp.onNext(1);
  
          ts.assertComplete();
      }
  
      @Test
      public void moreQueuedClean() {
          Flowable.range(1, 10)
          .window(5, 1)
          .doOnNext(w -> w.test())
          .test(3)
          .cancel();
      }
  
      @Test
      public void cancelWithoutWindowSize() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void cancelAfterAbandonmentSize() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void cancelWithoutWindowSkip() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10, 15)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void cancelAfterAbandonmentSkip() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10, 15)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void cancelWithoutWindowOverlap() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10, 5)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  
      @Test
      public void cancelAfterAbandonmentOverlap() {
          PublishProcessor<Integer> pp = PublishProcessor.create();
  
          TestSubscriber<Flowable<Integer>> ts = pp.window(10, 5)
          .test();
  
          assertTrue(pp.hasSubscribers());
  
          pp.onNext(1);
  
          ts.cancel();
  
          assertFalse("Subject still has subscribers!", pp.hasSubscribers());
      }
  }
  