TestSubscriber (reactor-core 3.0.0.RC1 API)

java.lang.Object
- reactor.test.TestSubscriber<T>

Type Parameters:

T - the value type.

All Implemented Interfaces:

org.reactivestreams.Subscriber<T>, org.reactivestreams.Subscription, Receiver, Trackable
```
public class TestSubscriber<T>
extends java.lang.Object
implements org.reactivestreams.Subscriber<T>, org.reactivestreams.Subscription, Trackable, Receiver
```
A Subscriber implementation that hosts assertion tests for its state and allows asynchronous cancellation and requesting.
To create a new instance of TestSubscriber, you have the choice between these static methods:
- subscribe(Publisher): create a new TestSubscriber, subscribe to it with the specified Publisher and requests an unbounded number of elements.
- subscribe(Publisher, long): create a new TestSubscriber, subscribe to it with the specified Publisher and requests n elements (can be 0 if you want no initial demand).
- create(): create a new TestSubscriber and requests an unbounded number of elements.
- create(long): create a new TestSubscriber and requests n elements (can be 0 if you want no initial demand).
If you are testing asynchronous publishers, don't forget to use one of the await*() methods to wait for the data to assert.
You can extend this class but only the onNext, onError and onComplete can be overridden. You can call request(long) and cancel() from any thread or from within the overridable methods but you should avoid calling the assertXXX methods asynchronously.
Usage:
```
 
 TestSubscriber
   .subscribe(publisher)
   .await()
   .assertValues("ABC", "DEF");
 
 
```

Field Summary

Fields
Modifier and Type Field and Description

static java.time.Duration DEFAULT_VALUES_TIMEOUT
Default timeout for waiting next values to be received
- Fields inherited from interface reactor.core.Trackable
  UNSPECIFIED

Fields
Modifier and Type	Field and Description
`static java.time.Duration`	`DEFAULT_VALUES_TIMEOUT` Default timeout for waiting next values to be received

Method Summary

All Methods Static Methods Instance Methods Concrete Methods
Modifier and Type	Method and Description
`TestSubscriber<T>`	`assertComplete()` Assert a complete successfully signal has been received.
`TestSubscriber<T>`	`assertContainValues(java.util.Set<? extends T> expectedValues)` Assert the specified values have been received.
`TestSubscriber<T>`	`assertError()` Assert an error signal has been received.
`TestSubscriber<T>`	`assertError(java.lang.Class<? extends java.lang.Throwable> clazz)` Assert an error signal has been received.
`TestSubscriber<T>`	`assertErrorMessage(java.lang.String message)`
`TestSubscriber<T>`	`assertErrorWith(java.util.function.Consumer<? super java.lang.Throwable> expectation)` Assert an error signal has been received.
`TestSubscriber<T>`	`assertFuseableSource()` Assert that the upstream was a Fuseable source.
`TestSubscriber<T>`	`assertFusionEnabled()` Assert that the fusion mode was granted.
`TestSubscriber<T>`	`assertFusionMode(int expectedMode)`
`TestSubscriber<T>`	`assertFusionRejected()` Assert that the fusion mode was granted.
`protected void`	`assertionError(java.lang.String message, java.lang.Throwable cause)` Prepares and throws an AssertionError exception based on the message, cause, the active state and the potential errors so far.
`TestSubscriber<T>`	`assertNoError()` Assert no error signal has been received.
`TestSubscriber<T>`	`assertNonFuseableSource()` Assert that the upstream was not a Fuseable source.
`TestSubscriber<T>`	`assertNotComplete()` Assert no complete successfully signal has been received.
`TestSubscriber<T>`	`assertNotSubscribed()` Assert no subscription occurred.
`TestSubscriber<T>`	`assertNotTerminated()` Assert no complete successfully or error signal has been received.
`TestSubscriber<T>`	`assertNoValues()` Assert no values have been received.
`TestSubscriber<T>`	`assertSubscribed()` Assert subscription occurred (once).
`TestSubscriber<T>`	`assertTerminated()` Assert either complete successfully or error signal has been received.
`TestSubscriber<T>`	`assertValueCount(long n)` Assert `n` values has been received.
`TestSubscriber<T>`	`assertValues(T... expectedValues)` Assert the specified values have been received in the declared order.
`TestSubscriber<T>`	`assertValueSequence(java.lang.Iterable<? extends T> expectedSequence)` Assert the specified values have been received in the same order read by the passed `Iterable`.
`TestSubscriber<T>`	`assertValuesWith(java.util.function.Consumer<T>... expectations)` Assert the specified values have been received in the declared order.
`TestSubscriber<T>`	`await()` Blocking method that waits until a complete successfully or error signal is received.
`TestSubscriber<T>`	`await(java.time.Duration timeout)` Blocking method that waits until a complete successfully or error signal is received or until a timeout occurs.
`static void`	`await(java.time.Duration timeout, java.lang.String errorMessage, java.util.function.BooleanSupplier conditionSupplier)` Blocking method that waits until `conditionSupplier` returns true, or if it does not before the specified timeout, throw an `AssertionError` with the specified error message.
`static void`	`await(java.time.Duration timeout, java.util.function.Supplier<java.lang.String> errorMessageSupplier, java.util.function.BooleanSupplier conditionSupplier)` Blocking method that waits until `conditionSupplier` returns true, or if it does not before the specified timeout, throws an `AssertionError` with the specified error message supplier.
`TestSubscriber<T>`	`awaitAndAssertNextValueCount(long n)` Blocking method that waits until `n` next values have been received.
`TestSubscriber<T>`	`awaitAndAssertNextValues(T... values)` Blocking method that waits until `n` next values have been received (n is the number of values provided) to assert them.
`TestSubscriber<T>`	`awaitAndAssertNextValuesWith(java.util.function.Consumer<T>... expectations)` Blocking method that waits until `n` next values have been received (n is the number of expectations provided) to assert them.
`void`	`cancel()`
`TestSubscriber<T>`	`configureValuesStorage(boolean enabled)` Enable or disabled the values storage.
`TestSubscriber<T>`	`configureValuesTimeout(java.time.Duration timeout)` Configure the timeout in seconds for waiting next values to be received (3 seconds by default).
`static <T> TestSubscriber<T>`	`create()` Create a new `TestSubscriber` that requests an unbounded number of elements.
`static <T> TestSubscriber<T>`	`create(long n)` Create a new `TestSubscriber` that requests initially `n` elements.
`int`	`establishedFusionMode()` Returns the established fusion mode or -1 if it was not enabled
`protected java.lang.String`	`fusionModeName(int mode)`
`boolean`	`isCancelled()`
`boolean`	`isStarted()` Has this upstream started or "onSubscribed" ?
`boolean`	`isTerminated()` Has this upstream finished or "completed" / "failed" ?
`protected void`	`normalRequest(long n)`
`void`	`onComplete()`
`void`	`onError(java.lang.Throwable t)`
`void`	`onNext(T t)`
`void`	`onSubscribe(org.reactivestreams.Subscription s)`
`void`	`request(long n)`
`protected void`	`requestDeferred()` Requests the deferred amount if not zero.
`long`	`requestedFromDownstream()` Return defined element capacity, used to drive new `Subscription` request needs.
`TestSubscriber<T>`	`requestedFusionMode(int requestMode)` Setup what fusion mode should be requested from the incomining Subscription if it happens to be QueueSubscription
`protected boolean`	`set(org.reactivestreams.Subscription s)` Atomically sets the single subscription and requests the missed amount from it.
`protected boolean`	`setWithoutRequesting(org.reactivestreams.Subscription s)` Sets the Subscription once but does not request anything.
`static <T> TestSubscriber<T>`	`subscribe(org.reactivestreams.Publisher<T> publisher)` Create a new `TestSubscriber` that requests an unbounded number of elements, and make the specified `publisher` subscribe to it.
`static <T> TestSubscriber<T>`	`subscribe(org.reactivestreams.Publisher<T> publisher, long n)` Create a new `TestSubscriber` that requests initially `n` elements, and make the specified `publisher` subscribe to it.
`org.reactivestreams.Subscription`	`upstream()` Return the direct source of data, Supports reference.
`protected java.lang.String`	`valueAndClass(java.lang.Object o)`

Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface reactor.core.Trackable
expectedFromUpstream, getCapacity, getError, getPending, limit

- Field Detail
  - DEFAULT_VALUES_TIMEOUT
```
public static final java.time.Duration DEFAULT_VALUES_TIMEOUT
```
    Default timeout for waiting next values to be received
- Method Detail
  - await
```
public static void await(java.time.Duration timeout,
                         java.util.function.Supplier<java.lang.String> errorMessageSupplier,
                         java.util.function.BooleanSupplier conditionSupplier)
```
    Blocking method that waits until conditionSupplier returns true, or if it does not before the specified timeout, throws an AssertionError with the specified error message supplier.
    
    Parameters:
    
    timeout - the timeout duration
    
    errorMessageSupplier - the error message supplier
    
    conditionSupplier - condition to break out of the wait loop
    
    Throws:
    
    java.lang.AssertionError
  - await
```
public static void await(java.time.Duration timeout,
                         java.lang.String errorMessage,
                         java.util.function.BooleanSupplier conditionSupplier)
```
    Blocking method that waits until conditionSupplier returns true, or if it does not before the specified timeout, throw an AssertionError with the specified error message.
    
    Parameters:
    
    timeout - the timeout duration
    
    errorMessage - the error message
    
    conditionSupplier - condition to break out of the wait loop
    
    Throws:
    
    java.lang.AssertionError
  - create
```
public static <T> TestSubscriber<T> create()
```
    Create a new TestSubscriber that requests an unbounded number of elements.
    Be sure at least a publisher has subscribed to it via Publisher.subscribe(Subscriber) before use assert methods.
    
    Type Parameters:
    
    T - the observed value type
    
    Returns:
    
    a fresh TestSubscriber instance
    
    See Also:
    
    subscribe(Publisher)
  - create
```
public static <T> TestSubscriber<T> create(long n)
```
    Create a new TestSubscriber that requests initially n elements. You can then manage the demand with Subscription.request(long).
    Be sure at least a publisher has subscribed to it via Publisher.subscribe(Subscriber) before use assert methods.
    
    Type Parameters:
    
    T - the observed value type
    
    Parameters:
    
    n - Number of elements to request (can be 0 if you want no initial demand).
    
    Returns:
    
    a fresh TestSubscriber instance
    
    See Also:
    
    subscribe(Publisher, long)
  - subscribe
```
public static <T> TestSubscriber<T> subscribe(org.reactivestreams.Publisher<T> publisher)
```
    Create a new TestSubscriber that requests an unbounded number of elements, and make the specified publisher subscribe to it.
    
    Type Parameters:
    
    T - the observed value type
    
    Parameters:
    
    publisher - The publisher to subscribe with
    
    Returns:
    
    a fresh TestSubscriber instance
  - subscribe
```
public static <T> TestSubscriber<T> subscribe(org.reactivestreams.Publisher<T> publisher,
                                              long n)
```
    Create a new TestSubscriber that requests initially n elements, and make the specified publisher subscribe to it. You can then manage the demand with Subscription.request(long).
    
    Type Parameters:
    
    T - the observed value type
    
    Parameters:
    
    publisher - The publisher to subscribe with
    
    n - Number of elements to request (can be 0 if you want no initial demand).
    
    Returns:
    
    a fresh TestSubscriber instance
  - configureValuesStorage
```
public final TestSubscriber<T> configureValuesStorage(boolean enabled)
```
    Enable or disabled the values storage. It is enabled by default, and can be disable in order to be able to perform performance benchmarks or tests with a huge amount values.
    
    Parameters:
    
    enabled - enable value storage?
    
    Returns:
    
    this
  - configureValuesTimeout
```
public final TestSubscriber<T> configureValuesTimeout(java.time.Duration timeout)
```
    Configure the timeout in seconds for waiting next values to be received (3 seconds by default).
    
    Parameters:
    
    timeout - the new default value timeout duration
    
    Returns:
    
    this
  - establishedFusionMode
```
public final int establishedFusionMode()
```
    Returns the established fusion mode or -1 if it was not enabled
    
    Returns:
    
    the fusion mode, see Fuseable constants
  - assertComplete
```
public final TestSubscriber<T> assertComplete()
```
    Assert a complete successfully signal has been received.
    
    Returns:
    
    this
  - assertContainValues
```
public final TestSubscriber<T> assertContainValues(java.util.Set<? extends T> expectedValues)
```
    Assert the specified values have been received. Values storage should be enabled to use this method.
    
    Parameters:
    
    expectedValues - the values to assert
    
    Returns:
    
    this
    
    See Also:
    
    configureValuesStorage(boolean)
  - assertError
```
public final TestSubscriber<T> assertError()
```
    Assert an error signal has been received.
    
    Returns:
    
    this
  - assertError
```
public final TestSubscriber<T> assertError(java.lang.Class<? extends java.lang.Throwable> clazz)
```
    Assert an error signal has been received.
    
    Parameters:
    
    clazz - The class of the exception contained in the error signal
    
    Returns:
    
    this
  - assertErrorMessage
```
public final TestSubscriber<T> assertErrorMessage(java.lang.String message)
```
  - assertErrorWith
```
public final TestSubscriber<T> assertErrorWith(java.util.function.Consumer<? super java.lang.Throwable> expectation)
```
    Assert an error signal has been received.
    
    Parameters:
    
    expectation - A method that can verify the exception contained in the error signal and throw an exception (like an AssertionError) if the exception is not valid.
    
    Returns:
    
    this
  - assertFuseableSource
```
public final TestSubscriber<T> assertFuseableSource()
```
    Assert that the upstream was a Fuseable source.
    
    Returns:
    
    this
  - assertFusionEnabled
```
public final TestSubscriber<T> assertFusionEnabled()
```
    Assert that the fusion mode was granted.
    
    Returns:
    
    this
  - assertFusionMode
```
public final TestSubscriber<T> assertFusionMode(int expectedMode)
```
  - assertFusionRejected
```
public final TestSubscriber<T> assertFusionRejected()
```
    Assert that the fusion mode was granted.
    
    Returns:
    
    this
  - assertNoError
```
public final TestSubscriber<T> assertNoError()
```
    Assert no error signal has been received.
    
    Returns:
    
    this
  - assertNoValues
```
public final TestSubscriber<T> assertNoValues()
```
    Assert no values have been received.
    
    Returns:
    
    this
  - assertNonFuseableSource
```
public final TestSubscriber<T> assertNonFuseableSource()
```
    Assert that the upstream was not a Fuseable source.
    
    Returns:
    
    this
  - assertNotComplete
```
public final TestSubscriber<T> assertNotComplete()
```
    Assert no complete successfully signal has been received.
    
    Returns:
    
    this
  - assertNotSubscribed
```
public final TestSubscriber<T> assertNotSubscribed()
```
    Assert no subscription occurred.
    
    Returns:
    
    this
  - assertNotTerminated
```
public final TestSubscriber<T> assertNotTerminated()
```
    Assert no complete successfully or error signal has been received.
    
    Returns:
    
    this
  - assertSubscribed
```
public final TestSubscriber<T> assertSubscribed()
```
    Assert subscription occurred (once).
    
    Returns:
    
    this
  - assertTerminated
```
public final TestSubscriber<T> assertTerminated()
```
    Assert either complete successfully or error signal has been received.
    
    Returns:
    
    this
  - assertValueCount
```
public final TestSubscriber<T> assertValueCount(long n)
```
    Assert n values has been received.
    
    Parameters:
    
    n - the expected value count
    
    Returns:
    
    this
  - assertValueSequence
```
public final TestSubscriber<T> assertValueSequence(java.lang.Iterable<? extends T> expectedSequence)
```
    Assert the specified values have been received in the same order read by the passed Iterable. Values storage should be enabled to use this method.
    
    Parameters:
    
    expectedSequence - the values to assert
    
    Returns:
    
    this
    
    See Also:
    
    configureValuesStorage(boolean)
  - assertValues
```
@SafeVarargs
public final TestSubscriber<T> assertValues(T... expectedValues)
```
    Assert the specified values have been received in the declared order. Values storage should be enabled to use this method.
    
    Parameters:
    
    expectedValues - the values to assert
    
    Returns:
    
    this
    
    See Also:
    
    configureValuesStorage(boolean)
  - assertValuesWith
```
@SafeVarargs
public final TestSubscriber<T> assertValuesWith(java.util.function.Consumer<T>... expectations)
```
    Assert the specified values have been received in the declared order. Values storage should be enabled to use this method.
    
    Parameters:
    
    expectations - One or more methods that can verify the values and throw a exception (like an AssertionError) if the value is not valid.
    
    Returns:
    
    this
    
    See Also:
    
    configureValuesStorage(boolean)
  - await
```
public final TestSubscriber<T> await()
```
    Blocking method that waits until a complete successfully or error signal is received.
    
    Returns:
    
    this
  - await
```
public final TestSubscriber<T> await(java.time.Duration timeout)
```
    Blocking method that waits until a complete successfully or error signal is received or until a timeout occurs.
    
    Parameters:
    
    timeout - The timeout value
    
    Returns:
    
    this
  - awaitAndAssertNextValueCount
```
public final TestSubscriber<T> awaitAndAssertNextValueCount(long n)
```
    Blocking method that waits until n next values have been received.
    
    Parameters:
    
    n - the value count to assert
    
    Returns:
    
    this
  - awaitAndAssertNextValues
```
@SafeVarargs
public final TestSubscriber<T> awaitAndAssertNextValues(T... values)
```
    Blocking method that waits until n next values have been received (n is the number of values provided) to assert them.
    
    Parameters:
    
    values - the values to assert
    
    Returns:
    
    this
  - awaitAndAssertNextValuesWith
```
@SafeVarargs
public final TestSubscriber<T> awaitAndAssertNextValuesWith(java.util.function.Consumer<T>... expectations)
```
    Blocking method that waits until n next values have been received (n is the number of expectations provided) to assert them.
    
    Parameters:
    
    expectations - One or more methods that can verify the values and throw a exception (like an AssertionError) if the value is not valid.
    
    Returns:
    
    this
  - cancel
```
public void cancel()
```
    Specified by:
    
    cancel in interface org.reactivestreams.Subscription
  - isCancelled
```
public final boolean isCancelled()
```
    Specified by:
    
    isCancelled in interface Trackable
    
    Returns:
    
    has the downstream "cancelled" and interrupted its consuming ?
  - isStarted
```
public final boolean isStarted()
```
    Description copied from interface: Trackable
    
    Has this upstream started or "onSubscribed" ?
    
    Specified by:
    
    isStarted in interface Trackable
    
    Returns:
    
    has this upstream started or "onSubscribed" ?
  - isTerminated
```
public final boolean isTerminated()
```
    Description copied from interface: Trackable
    
    Has this upstream finished or "completed" / "failed" ?
    
    Specified by:
    
    isTerminated in interface Trackable
    
    Returns:
    
    has this upstream finished or "completed" / "failed" ?
  - onComplete
```
public void onComplete()
```
    Specified by:
    
    onComplete in interface org.reactivestreams.Subscriber<T>
  - onError
```
public void onError(java.lang.Throwable t)
```
    Specified by:
    
    onError in interface org.reactivestreams.Subscriber<T>
  - onNext
```
public void onNext(T t)
```
    Specified by:
    
    onNext in interface org.reactivestreams.Subscriber<T>
  - onSubscribe
```
public void onSubscribe(org.reactivestreams.Subscription s)
```
    Specified by:
    
    onSubscribe in interface org.reactivestreams.Subscriber<T>
  - request
```
public void request(long n)
```
    Specified by:
    
    request in interface org.reactivestreams.Subscription
  - requestedFromDownstream
```
public final long requestedFromDownstream()
```
    Description copied from interface: Trackable
    
    Return defined element capacity, used to drive new Subscription request needs. This is the maximum in-flight data allowed to transit to this elements.
    
    Specified by:
    
    requestedFromDownstream in interface Trackable
    
    Returns:
    
    long capacity
  - requestedFusionMode
```
public final TestSubscriber<T> requestedFusionMode(int requestMode)
```
    Setup what fusion mode should be requested from the incomining Subscription if it happens to be QueueSubscription
    
    Parameters:
    
    requestMode - the mode to request, see Fuseable constants
    
    Returns:
    
    this
  - upstream
```
public org.reactivestreams.Subscription upstream()
```
    Description copied from interface: Receiver
    
    Return the direct source of data, Supports reference.
    
    Specified by:
    
    upstream in interface Receiver
    
    Returns:
    
    the direct source of data, Supports reference.
  - normalRequest
```
protected final void normalRequest(long n)
```
  - requestDeferred
```
protected final void requestDeferred()
```
    Requests the deferred amount if not zero.
  - set
```
protected final boolean set(org.reactivestreams.Subscription s)
```
    Atomically sets the single subscription and requests the missed amount from it.
    
    Parameters:
    
    s -
    
    Returns:
    
    false if this arbiter is cancelled or there was a subscription already set
  - setWithoutRequesting
```
protected final boolean setWithoutRequesting(org.reactivestreams.Subscription s)
```
    Sets the Subscription once but does not request anything.
    
    Parameters:
    
    s - the Subscription to set
    
    Returns:
    
    true if successful, false if the current subscription is not null
  - assertionError
```
protected final void assertionError(java.lang.String message,
                                    java.lang.Throwable cause)
```
    Prepares and throws an AssertionError exception based on the message, cause, the active state and the potential errors so far.
    
    Parameters:
    
    message - the message
    
    cause - the optional Throwable cause
    
    Throws:
    
    java.lang.AssertionError - as expected
  - fusionModeName
```
protected final java.lang.String fusionModeName(int mode)
```
  - valueAndClass
```
protected final java.lang.String valueAndClass(java.lang.Object o)
```

Class TestSubscriber<T>

Field Summary

Fields inherited from interface reactor.core.Trackable

Method Summary

Methods inherited from class java.lang.Object

Methods inherited from interface reactor.core.Trackable

Field Detail

DEFAULT_VALUES_TIMEOUT

Method Detail

await

await

create

create

subscribe

subscribe

configureValuesStorage

configureValuesTimeout

establishedFusionMode

assertComplete

assertContainValues

assertError

assertError

assertErrorMessage

assertErrorWith

assertFuseableSource

assertFusionEnabled

assertFusionMode

assertFusionRejected

assertNoError

assertNoValues

assertNonFuseableSource

assertNotComplete

assertNotSubscribed

assertNotTerminated

assertSubscribed

assertTerminated

assertValueCount

assertValueSequence

assertValues

assertValuesWith

await

await

awaitAndAssertNextValueCount

awaitAndAssertNextValues

awaitAndAssertNextValuesWith

cancel

isCancelled

isStarted

isTerminated

onComplete

onError

onNext

onSubscribe

request

requestedFromDownstream

requestedFusionMode

upstream

normalRequest

requestDeferred

set

setWithoutRequesting

assertionError

fusionModeName

valueAndClass