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.common;
  
  import java.util.*;
  import java.util.concurrent.*;
  
  import io.reactivex.common.exceptions.CompositeException;
  import io.reactivex.common.functions.Predicate;
  import io.reactivex.common.internal.functions.*;
  import io.reactivex.common.internal.utils.*;
  
  /**
   * Base class with shared infrastructure to support TestObserver and TestObserver.
   * @param <T> the value type consumed
   * @param <U> the subclass of this BaseTestConsumer
   */
  public abstract class TestConsumer<T, U extends TestConsumer<T, U>> implements Disposable {
      /** The latch that indicates an onError or onComplete has been called. */
      protected final CountDownLatch done;
      /** The list of values received. */
      protected final List<T> values;
      /** The list of errors received. */
      protected final List<Throwable> errors;
      /** The number of completions. */
      protected long completions;
      /** The last thread seen by the observer. */
      protected Thread lastThread;
  
      protected boolean checkSubscriptionOnce;
  
      protected int initialFusionMode;
  
      protected int establishedFusionMode;
  
      /**
       * The optional tag associated with this test consumer.
       * @since 2.0.7
       */
      protected CharSequence tag;
  
      /**
       * Indicates that one of the awaitX method has timed out.
       * @since 2.0.7
       */
      protected boolean timeout;
  
      public TestConsumer() {
          this.values = new VolatileSizeArrayList<T>();
          this.errors = new VolatileSizeArrayList<Throwable>();
          this.done = new CountDownLatch(1);
      }
  
      /**
       * Returns the last thread which called the onXXX methods of this TestObserver/TestObserver.
       * @return the last thread which called the onXXX methods
       */
      public final Thread lastThread() {
          return lastThread;
      }
  
      /**
       * Returns a shared list of received onNext values.
       * @return a list of received onNext values
       */
      public final List<T> values() {
          return values;
      }
  
      /**
       * Returns a shared list of received onError exceptions.
       * @return a list of received events onError exceptions
       */
      public final List<Throwable> errors() {
          return errors;
      }
  
      /**
       * Returns the number of times onComplete was called.
       * @return the number of times onComplete was called
       */
      public final long completions() {
          return completions;
      }
  
      /**
       * Returns true if TestObserver/TestObserver received any onError or onComplete events.
       * @return true if TestObserver/TestObserver received any onError or onComplete events
       */
      public final boolean isTerminated() {
          return done.getCount() == 0;
      }
  
      /**
       * Returns the number of onNext values received.
       * @return the number of onNext values received
       */
      public final int valueCount() {
          return values.size();
      }
  
      /**
       * Returns the number of onError exceptions received.
       * @return the number of onError exceptions received
       */
      public final int errorCount() {
          return errors.size();
      }
  
  
      /**
       * Fail with the given message and add the sequence of errors as suppressed ones.
       * <p>Note this is deliberately the only fail method. Most of the times an assertion
       * would fail but it is possible it was due to an exception somewhere. This construct
       * will capture those potential errors and report it along with the original failure.
       *
       * @param message the message to use
       * @return AssertionError the prepared AssertionError instance
       */
      protected final AssertionError fail(String message) {
          StringBuilder b = new StringBuilder(64 + message.length());
          b.append(message);
  
          b.append(" (")
          .append("latch = ").append(done.getCount()).append(", ")
          .append("values = ").append(values.size()).append(", ")
          .append("errors = ").append(errors.size()).append(", ")
          .append("completions = ").append(completions)
          ;
  
          if (timeout) {
              b.append(", timeout!");
          }
  
          if (isDisposed()) {
              b.append(", disposed!");
          }
  
          CharSequence tag = this.tag;
          if (tag != null) {
              b.append(", tag = ")
              .append(tag);
          }
  
          b
          .append(')')
          ;
  
          AssertionError ae = new AssertionError(b.toString());
          if (!errors.isEmpty()) {
              if (errors.size() == 1) {
                  ae.initCause(errors.get(0));
              } else {
                  CompositeException ce = new CompositeException(errors);
                  ae.initCause(ce);
              }
          }
          return ae;
      }
  
      /**
       * Awaits until this TestObserver/TestObserver receives an onError or onComplete events.
       * @return this
       * @throws InterruptedException if the current thread is interrupted while waiting
       * @see #awaitTerminalEvent()
       */
      @SuppressWarnings("unchecked")
      public final U await() throws InterruptedException {
          if (done.getCount() == 0) {
              return (U)this;
          }
  
          done.await();
          return (U)this;
      }
  
      /**
       * Awaits the specified amount of time or until this TestObserver/TestObserver
       * receives an onError or onComplete events, whichever happens first.
       * @param time the waiting time
       * @param unit the time unit of the waiting time
       * @return true if the TestObserver/TestObserver terminated, false if timeout happened
       * @throws InterruptedException if the current thread is interrupted while waiting
       * @see #awaitTerminalEvent(long, TimeUnit)
       */
      public final boolean await(long time, TimeUnit unit) throws InterruptedException {
          boolean d = done.getCount() == 0 || (done.await(time, unit));
          timeout = !d;
          return d;
      }
  
      // assertion methods
  
      /**
       * Assert that this TestObserver/TestObserver received exactly one onComplete event.
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertComplete() {
          long c = completions;
          if (c == 0) {
              throw fail("Not completed");
          } else
          if (c > 1) {
              throw fail("Multiple completions: " + c);
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver has not received any onComplete event.
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertNotComplete() {
          long c = completions;
          if (c == 1) {
              throw fail("Completed!");
          } else
          if (c > 1) {
              throw fail("Multiple completions: " + c);
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver has not received any onError event.
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertNoErrors() {
          int s = errors.size();
          if (s != 0) {
              throw fail("Error(s) present: " + errors);
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver received exactly the specified onError event value.
       *
       * <p>The comparison is performed via Objects.equals(); since most exceptions don't
       * implement equals(), this assertion may fail. Use the {@link #assertError(Class)}
       * overload to test against the class of an error instead of an instance of an error
       * or {@link #assertError(Predicate)} to test with different condition.
       * @param error the error to check
       * @return this;
       * @see #assertError(Class)
       * @see #assertError(Predicate)
       */
      public final U assertError(Throwable error) {
          return assertError(Functions.equalsWith(error));
      }
  
      /**
       * Asserts that this TestObserver/TestObserver received exactly one onError event which is an
       * instance of the specified errorClass class.
       * @param errorClass the error class to expect
       * @return this;
       */
      @SuppressWarnings({ "unchecked", "rawtypes", "cast" })
      public final U assertError(Class<? extends Throwable> errorClass) {
          return (U)assertError((Predicate)Functions.isInstanceOf(errorClass));
      }
  
      /**
       * Asserts that this TestObserver/TestObserver received exactly one onError event for which
       * the provided predicate returns true.
       * @param errorPredicate
       *            the predicate that receives the error Throwable
       *            and should return true for expected errors.
       * @return this
       */
      @SuppressWarnings("unchecked")
      public final U assertError(Predicate<Throwable> errorPredicate) {
          int s = errors.size();
          if (s == 0) {
              throw fail("No errors");
          }
  
          boolean found = false;
  
          for (Throwable e : errors) {
              try {
                  if (errorPredicate.test(e)) {
                      found = true;
                      break;
                  }
              } catch (Exception ex) {
                  throw ExceptionHelper.wrapOrThrow(ex);
              }
          }
  
          if (found) {
              if (s != 1) {
                  throw fail("Error present but other errors as well");
              }
          } else {
              throw fail("Error not present");
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver received exactly one onNext value which is equal to
       * the given value with respect to Objects.equals.
       * @param value the value to expect
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertValue(T value) {
          int s = values.size();
          if (s != 1) {
              throw fail("Expected: " + valueAndClass(value) + ", Actual: " + values);
          }
          T v = values.get(0);
          if (!ObjectHelper.equals(value, v)) {
              throw fail("Expected: " + valueAndClass(value) + ", Actual: " + valueAndClass(v));
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver did not receive an onNext value which is equal to
       * the given value with respect to Objects.equals.
       *
       * <p>History: 2.0.5 - experimental
       * @since 2.1
       * @param value the value to expect not being received
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertNever(T value) {
          int s = values.size();
  
          for (int i = 0; i < s; i++) {
              T v = this.values.get(i);
              if (ObjectHelper.equals(v, value)) {
                  throw fail("Value at position " + i + " is equal to " + valueAndClass(value) + "; Expected them to be different");
              }
          }
          return (U) this;
      }
  
      /**
       * Asserts that this TestObserver/TestObserver received exactly one onNext value for which
       * the provided predicate returns true.
       * @param valuePredicate
       *            the predicate that receives the onNext value
       *            and should return true for the expected value.
       * @return this
       */
      @SuppressWarnings("unchecked")
      public final U assertValue(Predicate<T> valuePredicate) {
          assertValueAt(0, valuePredicate);
  
          if (values.size() > 1) {
              throw fail("Value present but other values as well");
          }
  
          return (U)this;
      }
  
      /**
       * Asserts that this TestObserver/TestObserver did not receive any onNext value for which
       * the provided predicate returns true.
       *
       * <p>History: 2.0.5 - experimental
       * @since 2.1
       * @param valuePredicate the predicate that receives the onNext value
       *                       and should return true for the expected value.
       * @return this
       */
      @SuppressWarnings("unchecked")
      public final U assertNever(Predicate<? super T> valuePredicate) {
          int s = values.size();
  
          for (int i = 0; i < s; i++) {
              T v = this.values.get(i);
              try {
                  if (valuePredicate.test(v)) {
                      throw fail("Value at position " + i + " matches predicate " + valuePredicate.toString() + ", which was not expected.");
                  }
              } catch (Exception ex) {
                  throw ExceptionHelper.wrapOrThrow(ex);
              }
          }
          return (U)this;
      }
  
      /**
       * Asserts that this TestObserver/TestObserver received an onNext value at the given index
       * for the provided predicate returns true.
       * @param index the position to assert on
       * @param valuePredicate
       *            the predicate that receives the onNext value
       *            and should return true for the expected value.
       * @return this
       */
      @SuppressWarnings("unchecked")
      public final U assertValueAt(int index, Predicate<T> valuePredicate) {
          int s = values.size();
          if (s == 0) {
              throw fail("No values");
          }
  
          if (index >= values.size()) {
              throw fail("Invalid index: " + index);
          }
  
          boolean found = false;
  
          try {
              if (valuePredicate.test(values.get(index))) {
                  found = true;
              }
          } catch (Exception ex) {
              throw ExceptionHelper.wrapOrThrow(ex);
          }
  
          if (!found) {
              throw fail("Value not present");
          }
          return (U)this;
      }
  
      /**
       * Appends the class name to a non-null value.
       * @param o the object
       * @return the string representation
       */
      public static String valueAndClass(Object o) {
          if (o != null) {
              return o + " (class: " + o.getClass().getSimpleName() + ")";
          }
          return "null";
      }
  
      /**
       * Assert that this TestObserver/TestObserver received the specified number onNext events.
       * @param count the expected number of onNext events
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertValueCount(int count) {
          int s = values.size();
          if (s != count) {
              throw fail("Value counts differ; Expected: " + count + ", Actual: " + s);
          }
          return (U)this;
      }
  
      /**
       * Assert that this TestObserver/TestObserver has not received any onNext events.
       * @return this;
       */
      public final U assertNoValues() {
          return assertValueCount(0);
      }
  
      /**
       * Assert that the TestObserver/TestObserver received only the specified values in the specified order.
       * @param values the values expected
       * @return this;
       * @see #assertValueSet(Collection)
       */
      @SuppressWarnings("unchecked")
      public final U assertValues(T... values) {
          int s = this.values.size();
          if (s != values.length) {
              throw fail("Value count differs; Expected: " + values.length + " " + Arrays.toString(values)
              + ", Actual: " + s + " " + this.values);
          }
          for (int i = 0; i < s; i++) {
              T v = this.values.get(i);
              T u = values[i];
              if (!ObjectHelper.equals(u, v)) {
                  throw fail("Values at position " + i + " differ; Expected: " + valueAndClass(u) + ", Actual: " + valueAndClass(v));
              }
          }
          return (U)this;
      }
  
      /**
       * Assert that the TestObserver/TestObserver received only the specified values in any order.
       * <p>This helps asserting when the order of the values is not guaranteed, i.e., when merging
       * asynchronous streams.
       *
       * @param expected the collection of values expected in any order
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertValueSet(Collection<? extends T> expected) {
          if (expected.isEmpty()) {
              assertNoValues();
              return (U)this;
          }
          for (T v : this.values) {
              if (!expected.contains(v)) {
                  throw fail("Value not in the expected collection: " + valueAndClass(v));
              }
          }
          return (U)this;
      }
  
      /**
       * Assert that the TestObserver/TestObserver received only the specified sequence of values in the same order.
       * @param sequence the sequence of expected values in order
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertValueSequence(Iterable<? extends T> sequence) {
          int i = 0;
          Iterator<T> vit = values.iterator();
          Iterator<? extends T> it = sequence.iterator();
          boolean actualNext;
          boolean expectedNext;
          for (;;) {
              actualNext = it.hasNext();
              expectedNext = vit.hasNext();
  
              if (!actualNext || !expectedNext) {
                  break;
              }
  
              T v = it.next();
              T u = vit.next();
  
              if (!ObjectHelper.equals(u, v)) {
                  throw fail("Values at position " + i + " differ; Expected: " + valueAndClass(u) + ", Actual: " + valueAndClass(v));
              }
              i++;
          }
  
          if (actualNext) {
              throw fail("More values received than expected (" + i + ")");
          }
          if (expectedNext) {
              throw fail("Fewer values received than expected (" + i + ")");
          }
          return (U)this;
      }
  
      /**
       * Assert that the TestObserver/TestObserver terminated (i.e., the terminal latch reached zero).
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertTerminated() {
          if (done.getCount() != 0) {
              throw fail("Subscriber still running!");
          }
          long c = completions;
          if (c > 1) {
              throw fail("Terminated with multiple completions: " + c);
          }
          int s = errors.size();
          if (s > 1) {
              throw fail("Terminated with multiple errors: " + s);
          }
  
          if (c != 0 && s != 0) {
              throw fail("Terminated with multiple completions and errors: " + c);
          }
          return (U)this;
      }
  
      /**
       * Assert that the TestObserver/TestObserver has not terminated (i.e., the terminal latch is still non-zero).
       * @return this;
       */
      @SuppressWarnings("unchecked")
      public final U assertNotTerminated() {
          if (done.getCount() == 0) {
              throw fail("Subscriber terminated!");
          }
          return (U)this;
      }
  
      /**
       * Waits until the any terminal event has been received by this TestObserver/TestObserver
       * or returns false if the wait has been interrupted.
       * @return true if the TestObserver/TestObserver terminated, false if the wait has been interrupted
       */
      public final boolean awaitTerminalEvent() {
          try {
              await();
              return true;
          } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
              return false;
          }
      }
  
      /**
       * Awaits the specified amount of time or until this TestObserver/TestObserver
       * receives an onError or onComplete events, whichever happens first.
       * @param duration the waiting time
       * @param unit the time unit of the waiting time
       * @return true if the TestObserver/TestObserver terminated, false if timeout or interrupt happened
       */
      public final boolean awaitTerminalEvent(long duration, TimeUnit unit) {
          try {
              return await(duration, unit);
          } catch (InterruptedException ex) {
              Thread.currentThread().interrupt();
              return false;
          }
      }
  
      /**
       * Assert that there is a single error and it has the given message.
       * @param message the message expected
       * @return this
       */
      @SuppressWarnings("unchecked")
      public final U assertErrorMessage(String message) {
          int s = errors.size();
          if (s == 0) {
              throw fail("No errors");
          } else
          if (s == 1) {
              Throwable e = errors.get(0);
              String errorMessage = e.getMessage();
              if (!ObjectHelper.equals(message, errorMessage)) {
                  throw fail("Error message differs; Expected: " + message + ", Actual: " + errorMessage);
              }
          } else {
              throw fail("Multiple errors");
          }
          return (U)this;
      }
  
      /**
       * Returns a list of 3 other lists: the first inner list contains the plain
       * values received; the second list contains the potential errors
       * and the final list contains the potential completions as Notifications.
       *
       * @return a list of (values, errors, completion-notifications)
       */
      @SuppressWarnings({ "rawtypes", "unchecked" })
      public final List<List<Object>> getEvents() {
          List<List<Object>> result = new ArrayList<List<Object>>();
  
          result.add((List)values());
  
          result.add((List)errors());
  
          List<Object> completeList = new ArrayList<Object>();
          for (long i = 0; i < completions; i++) {
              completeList.add(Notification.createOnComplete());
          }
          result.add(completeList);
  
          return result;
      }
  
      /**
       * Assert that the onSubscribe method was called exactly once.
       * @return this;
       */
      public abstract U assertSubscribed();
  
      /**
       * Assert that the onSubscribe method hasn't been called at all.
       * @return this;
       */
      public abstract U assertNotSubscribed();
  
      /**
       * Assert that the upstream signalled the specified values in order and
       * completed normally.
       * @param values the expected values, asserted in order
       * @return this
       * @see #assertFailure(Class, Object...)
       * @see #assertFailure(Predicate, Object...)
       * @see #assertFailureAndMessage(Class, String, Object...)
       */
      public final U assertResult(T... values) {
          return assertSubscribed()
                  .assertValues(values)
                  .assertNoErrors()
                  .assertComplete();
      }
  
      /**
       * Assert that the upstream signalled the specified values in order
       * and then failed with a specific class or subclass of Throwable.
       * @param error the expected exception (parent) class
       * @param values the expected values, asserted in order
       * @return this
       */
      public final U assertFailure(Class<? extends Throwable> error, T... values) {
          return assertSubscribed()
                  .assertValues(values)
                  .assertError(error)
                  .assertNotComplete();
      }
  
      /**
       * Assert that the upstream signalled the specified values in order and then failed
       * with a Throwable for which the provided predicate returns true.
       * @param errorPredicate
       *            the predicate that receives the error Throwable
       *            and should return true for expected errors.
       * @param values the expected values, asserted in order
       * @return this
       */
      public final U assertFailure(Predicate<Throwable> errorPredicate, T... values) {
          return assertSubscribed()
                  .assertValues(values)
                  .assertError(errorPredicate)
                  .assertNotComplete();
      }
  
      /**
       * Assert that the upstream signalled the specified values in order,
       * then failed with a specific class or subclass of Throwable
       * and with the given exact error message.
       * @param error the expected exception (parent) class
       * @param message the expected failure message
       * @param values the expected values, asserted in order
       * @return this
       */
      public final U assertFailureAndMessage(Class<? extends Throwable> error,
              String message, T... values) {
          return assertSubscribed()
                  .assertValues(values)
                  .assertError(error)
                  .assertErrorMessage(message)
                  .assertNotComplete();
      }
  
      /**
       * Awaits until the internal latch is counted down.
       * <p>If the wait times out or gets interrupted, the TestObserver/TestObserver is cancelled.
       * @param time the waiting time
       * @param unit the time unit of the waiting time
       * @return this
       * @throws RuntimeException wrapping an InterruptedException if the wait is interrupted
       */
      @SuppressWarnings("unchecked")
      public final U awaitDone(long time, TimeUnit unit) {
          try {
              if (!done.await(time, unit)) {
                  timeout = true;
                  dispose();
              }
          } catch (InterruptedException ex) {
              dispose();
              throw ExceptionHelper.wrapOrThrow(ex);
          }
          return (U)this;
      }
  
  
      /**
       * Assert that the TestObserver/TestObserver has received a Disposable but no other events.
       * @return this
       */
      public final U assertEmpty() {
          return assertSubscribed()
                  .assertNoValues()
                  .assertNoErrors()
                  .assertNotComplete();
      }
  
      /**
       * Set the tag displayed along with an assertion failure's
       * other state information.
       * <p>History: 2.0.7 - experimental
       * @param tag the string to display (null won't print any tag)
       * @return this
       * @since 2.1
       */
      @SuppressWarnings("unchecked")
      public final U withTag(CharSequence tag) {
          this.tag = tag;
          return (U)this;
      }
  
      /**
       * Enumeration of default wait strategies when waiting for a specific number of
       * items in {@link TestConsumer#awaitCount(int, Runnable)}.
       * <p>History: 2.0.7 - experimental
       * @since 2.1
       */
      public enum TestWaitStrategy implements Runnable {
          /** The wait loop will spin as fast as possible. */
          SPIN {
              @Override
              public void run() {
                  // nothing to do
              }
          },
          /** The current thread will be yielded. */
          YIELD {
              @Override
              public void run() {
                  Thread.yield();
              }
          },
          /** The current thread sleeps for 1 millisecond. */
          SLEEP_1MS {
              @Override
              public void run() {
                  sleep(1);
              }
          },
          /** The current thread sleeps for 10 milliseconds. */
          SLEEP_10MS {
              @Override
              public void run() {
                  sleep(10);
              }
          },
          /** The current thread sleeps for 100 milliseconds. */
          SLEEP_100MS {
              @Override
              public void run() {
                  sleep(100);
              }
          },
          /** The current thread sleeps for 1000 milliseconds. */
          SLEEP_1000MS {
              @Override
              public void run() {
                  sleep(1000);
              }
          }
          ;
  
          @Override
          public abstract void run();
  
          static void sleep(int millis) {
              try {
                  Thread.sleep(millis);
              } catch (InterruptedException ex) {
                  throw new RuntimeException(ex);
              }
          }
      }
  
  
      /**
       * Await until the TestObserver/TestObserver receives the given
       * number of items or terminates by sleeping 10 milliseconds at a time
       * up to 5000 milliseconds of timeout.
       * <p>History: 2.0.7 - experimental
       * @param atLeast the number of items expected at least
       * @return this
       * @see #awaitCount(int, Runnable, long)
       * @since 2.1
       */
      public final U awaitCount(int atLeast) {
          return awaitCount(atLeast, TestWaitStrategy.SLEEP_10MS, 5000);
      }
  
      /**
       * Await until the TestObserver/TestObserver receives the given
       * number of items or terminates by waiting according to the wait
       * strategy and up to 5000 milliseconds of timeout.
       * <p>History: 2.0.7 - experimental
       * @param atLeast the number of items expected at least
       * @param waitStrategy a Runnable called when the current received count
       *                     hasn't reached the expected value and there was
       *                     no terminal event either, see {@link TestWaitStrategy}
       *                     for examples
       * @return this
       * @see #awaitCount(int, Runnable, long)
       * @since 2.1
       */
      public final U awaitCount(int atLeast, Runnable waitStrategy) {
          return awaitCount(atLeast, waitStrategy, 5000);
      }
  
      /**
       * Await until the TestObserver/TestObserver receives the given
       * number of items or terminates.
       * <p>History: 2.0.7 - experimental
       * @param atLeast the number of items expected at least
       * @param waitStrategy a Runnable called when the current received count
       *                     hasn't reached the expected value and there was
       *                     no terminal event either, see {@link TestWaitStrategy}
       *                     for examples
       * @param timeoutMillis if positive, the await ends if the specified amount of
       *                      time has passed no matter how many items were received
       * @return this
       * @since 2.1
       */
      @SuppressWarnings("unchecked")
      public final U awaitCount(int atLeast, Runnable waitStrategy, long timeoutMillis) {
          long start = System.currentTimeMillis();
          for (;;) {
              if (timeoutMillis > 0L && System.currentTimeMillis() - start >= timeoutMillis) {
                  timeout = true;
                  break;
              }
              if (done.getCount() == 0L) {
                  break;
              }
              if (values.size() >= atLeast) {
                  break;
              }
  
              waitStrategy.run();
          }
          return (U)this;
      }
  
      /**
       * <p>History: 2.0.7 - experimental
       * @return true if one of the timeout-based await methods has timed out.
       * @see #clearTimeout()
       * @see #assertTimeout()
       * @see #assertNoTimeout()
       * @since 2.1
       */
      public final boolean isTimeout() {
          return timeout;
      }
  
      /**
       * Clears the timeout flag set by the await methods when they timed out.
       * <p>History: 2.0.7 - experimental
       * @return this
       * @since 2.1
       * @see #isTimeout()
       */
      @SuppressWarnings("unchecked")
      public final U clearTimeout() {
          timeout = false;
          return (U)this;
      }
  
      /**
       * Asserts that some awaitX method has timed out.
       * <p>History: 2.0.7 - experimental
       * @return this
       * @since 2.1
       */
      @SuppressWarnings("unchecked")
      public final U assertTimeout() {
          if (!timeout) {
              throw fail("No timeout?!");
          }
          return (U)this;
      }
  
  
      /**
       * Asserts that some awaitX method has not timed out.
       * <p>History: 2.0.7 - experimental
       * @return this
       * @since 2.1
       */
      @SuppressWarnings("unchecked")
      public final U assertNoTimeout() {
          if (timeout) {
              throw fail("Timeout?!");
          }
          return (U)this;
      }
  }
  