reactor.test.subscriber

类 AssertSubscriber<T>

java.lang.Object

reactor.test.subscriber.AssertSubscriber<T>

类型参数:

T - the value type.

所有已实现的接口:

org.reactivestreams.Subscriber<T>, org.reactivestreams.Subscription, CoreSubscriber<T>

public class AssertSubscriber<T>
extends java.lang.Object
implements CoreSubscriber<T>, org.reactivestreams.Subscription

A Subscriber implementation that hosts assertion tests for its state and allows asynchronous cancellation and requesting.

To create a new instance of AssertSubscriber, you have the choice between these static methods:

• create(): create a new AssertSubscriber and requests an unbounded number of elements.
• create(long): create a new AssertSubscriber 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:

 
 AssertSubscriber
   .subscribe(publisher)
   .await()
   .assertValues("ABC", "DEF");
 
 

作者:

Sebastien Deleuze, David Karnok, Anatoly Kadyshev, Stephane Maldini, Brian Clozel

字段概要

字段

限定符和类型	字段和说明
static java.time.Duration	DEFAULT_VALUES_TIMEOUT Default timeout for waiting next values to be received

构造器概要

构造器

构造器和说明
AssertSubscriber()
AssertSubscriber(Context context)
AssertSubscriber(Context context, long n)
AssertSubscriber(long n)

方法概要

限定符和类型	方法和说明
AssertSubscriber<T>	assertComplete() Assert a complete successfully signal has been received.
AssertSubscriber<T>	assertContainValues(java.util.Set<? extends T> expectedValues) Assert the specified values have been received.
AssertSubscriber<T>	assertError() Assert an error signal has been received.
AssertSubscriber<T>	assertError(java.lang.Class<? extends java.lang.Throwable> clazz) Assert an error signal has been received.
AssertSubscriber<T>	assertErrorMessage(java.lang.String message)
AssertSubscriber<T>	assertErrorWith(java.util.function.Consumer<? super java.lang.Throwable> expectation) Assert an error signal has been received.
AssertSubscriber<T>	assertFuseableSource() Assert that the upstream was a Fuseable source.
AssertSubscriber<T>	assertFusionEnabled() Assert that the fusion mode was granted.
AssertSubscriber<T>	assertFusionMode(int expectedMode)
AssertSubscriber<T>	assertFusionRejected() Assert that the fusion mode was granted.
AssertSubscriber<T>	assertIncomplete(T... values)
AssertSubscriber<T>	assertNoError() Assert no error signal has been received.
AssertSubscriber<T>	assertNoEvents()
AssertSubscriber<T>	assertNonFuseableSource() Assert that the upstream was not a Fuseable source.
AssertSubscriber<T>	assertNotComplete() Assert no complete successfully signal has been received.
AssertSubscriber<T>	assertNotSubscribed() Assert no subscription occurred.
AssertSubscriber<T>	assertNotTerminated() Assert no complete successfully or error signal has been received.
AssertSubscriber<T>	assertNoValues() Assert no values have been received.
AssertSubscriber<T>	assertSubscribed() Assert subscription occurred (once).
AssertSubscriber<T>	assertTerminated() Assert either complete successfully or error signal has been received.
AssertSubscriber<T>	assertValueCount(long n) Assert n values has been received.
AssertSubscriber<T>	assertValues(T... expectedValues) Assert the specified values have been received in the declared order.
AssertSubscriber<T>	assertValueSequence(java.lang.Iterable<? extends T> expectedSequence) Assert the specified values have been received in the same order read by the passed Iterable.
AssertSubscriber<T>	assertValuesWith(java.util.function.Consumer<T>... expectations) Assert the specified values have been received in the declared order.
AssertSubscriber<T>	await() Blocking method that waits until a complete successfully or error signal is received.
AssertSubscriber<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.
AssertSubscriber<T>	awaitAndAssertNextValueCount(long n) Blocking method that waits until n next values have been received.
AssertSubscriber<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.
AssertSubscriber<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()
AssertSubscriber<T>	configureValuesStorage(boolean enabled) Enable or disabled the values storage.
AssertSubscriber<T>	configureValuesTimeout(java.time.Duration timeout) Configure the timeout in seconds for waiting next values to be received (3 seconds by default).
static <T> AssertSubscriber<T>	create() Create a new AssertSubscriber that requests an unbounded number of elements.
static <T> AssertSubscriber<T>	create(long n) Create a new AssertSubscriber that requests initially n elements.
Context	currentContext() Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.
int	establishedFusionMode() Returns the established fusion mode or -1 if it was not enabled
boolean	isTerminated()
void	onComplete()
void	onError(java.lang.Throwable t)
void	onNext(T t)
void	onSubscribe(org.reactivestreams.Subscription s) Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long).
void	request(long n)
AssertSubscriber<T>	requestedFusionMode(int requestMode) Setup what fusion mode should be requested from the incoming Subscription if it happens to be QueueSubscription
org.reactivestreams.Subscription	upstream()
java.util.List<T>	values()

从类继承的方法 java.lang.Object

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

字段详细资料

DEFAULT_VALUES_TIMEOUT

public static final java.time.Duration DEFAULT_VALUES_TIMEOUT

Default timeout for waiting next values to be received

构造器详细资料

AssertSubscriber

public AssertSubscriber()

AssertSubscriber

public AssertSubscriber(long n)

AssertSubscriber

public AssertSubscriber(Context context)

AssertSubscriber

public AssertSubscriber(Context context,
                        long n)

方法详细资料

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.

参数:

timeout - the timeout duration

errorMessageSupplier - the error message supplier

conditionSupplier - condition to break out of the wait loop

抛出:

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.

参数:

timeout - the timeout duration

errorMessage - the error message

conditionSupplier - condition to break out of the wait loop

抛出:

java.lang.AssertionError

create

public static <T> AssertSubscriber<T> create()

Create a new AssertSubscriber 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.

类型参数:

T - the observed value type

返回:

a fresh AssertSubscriber instance

create

public static <T> AssertSubscriber<T> create(long n)

Create a new AssertSubscriber 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.

类型参数:

T - the observed value type

参数:

n - Number of elements to request (can be 0 if you want no initial demand).

返回:

a fresh AssertSubscriber instance

configureValuesStorage

public final AssertSubscriber<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.

参数:

enabled - enable value storage?

返回:

this

configureValuesTimeout

public final AssertSubscriber<T> configureValuesTimeout(java.time.Duration timeout)

Configure the timeout in seconds for waiting next values to be received (3 seconds by default).

参数:

timeout - the new default value timeout duration

返回:

this

establishedFusionMode

public final int establishedFusionMode()

Returns the established fusion mode or -1 if it was not enabled

返回:

the fusion mode, see Fuseable constants

assertComplete

public final AssertSubscriber<T> assertComplete()

Assert a complete successfully signal has been received.

返回:

this

assertContainValues

public final AssertSubscriber<T> assertContainValues(java.util.Set<? extends T> expectedValues)

Assert the specified values have been received. Values storage should be enabled to use this method.

参数:

expectedValues - the values to assert

返回:

this

另请参阅:

configureValuesStorage(boolean)

assertError

public final AssertSubscriber<T> assertError()

Assert an error signal has been received.

返回:

this

assertError

public final AssertSubscriber<T> assertError(java.lang.Class<? extends java.lang.Throwable> clazz)

Assert an error signal has been received.

参数:

clazz - The class of the exception contained in the error signal

返回:

this

assertErrorMessage

public final AssertSubscriber<T> assertErrorMessage(java.lang.String message)

assertErrorWith

public final AssertSubscriber<T> assertErrorWith(java.util.function.Consumer<? super java.lang.Throwable> expectation)

Assert an error signal has been received.

参数:

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.

返回:

this

assertFuseableSource

public final AssertSubscriber<T> assertFuseableSource()

Assert that the upstream was a Fuseable source.

返回:

this

assertFusionEnabled

public final AssertSubscriber<T> assertFusionEnabled()

Assert that the fusion mode was granted.

返回:

this

assertFusionMode

public final AssertSubscriber<T> assertFusionMode(int expectedMode)

assertFusionRejected

public final AssertSubscriber<T> assertFusionRejected()

Assert that the fusion mode was granted.

返回:

this

assertNoError

public final AssertSubscriber<T> assertNoError()

Assert no error signal has been received.

返回:

this

assertNoValues

public final AssertSubscriber<T> assertNoValues()

Assert no values have been received.

返回:

this

assertNonFuseableSource

public final AssertSubscriber<T> assertNonFuseableSource()

Assert that the upstream was not a Fuseable source.

返回:

this

assertNotComplete

public final AssertSubscriber<T> assertNotComplete()

Assert no complete successfully signal has been received.

返回:

this

assertNotSubscribed

public final AssertSubscriber<T> assertNotSubscribed()

Assert no subscription occurred.

返回:

this

assertNotTerminated

public final AssertSubscriber<T> assertNotTerminated()

Assert no complete successfully or error signal has been received.

返回:

this

assertSubscribed

public final AssertSubscriber<T> assertSubscribed()

Assert subscription occurred (once).

返回:

this

assertTerminated

public final AssertSubscriber<T> assertTerminated()

Assert either complete successfully or error signal has been received.

返回:

this

assertValueCount

public final AssertSubscriber<T> assertValueCount(long n)

Assert n values has been received.

参数:

n - the expected value count

返回:

this

assertValueSequence

public final AssertSubscriber<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.

参数:

expectedSequence - the values to assert

返回:

this

另请参阅:

configureValuesStorage(boolean)

assertValues

@SafeVarargs
public final AssertSubscriber<T> assertValues(T... expectedValues)

Assert the specified values have been received in the declared order. Values storage should be enabled to use this method.

参数:

expectedValues - the values to assert

返回:

this

另请参阅:

configureValuesStorage(boolean)

assertValuesWith

@SafeVarargs
public final AssertSubscriber<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.

参数:

expectations - One or more methods that can verify the values and throw a exception (like an AssertionError) if the value is not valid.

返回:

this

另请参阅:

configureValuesStorage(boolean)

await

public final AssertSubscriber<T> await()

Blocking method that waits until a complete successfully or error signal is received.

返回:

this

await

public final AssertSubscriber<T> await(java.time.Duration timeout)

Blocking method that waits until a complete successfully or error signal is received or until a timeout occurs.

参数:

timeout - The timeout value

返回:

this

awaitAndAssertNextValueCount

public final AssertSubscriber<T> awaitAndAssertNextValueCount(long n)

Blocking method that waits until n next values have been received.

参数:

n - the value count to assert

返回:

this

awaitAndAssertNextValues

@SafeVarargs
public final AssertSubscriber<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.

参数:

values - the values to assert

返回:

this

awaitAndAssertNextValuesWith

@SafeVarargs
public final AssertSubscriber<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.

参数:

expectations - One or more methods that can verify the values and throw a exception (like an AssertionError) if the value is not valid.

返回:

this

cancel

public void cancel()

指定者:

cancel 在接口中 org.reactivestreams.Subscription

isTerminated

public final boolean isTerminated()

onComplete

public void onComplete()

指定者:

onComplete 在接口中 org.reactivestreams.Subscriber<T>

onError

public void onError(java.lang.Throwable t)

指定者:

onError 在接口中 org.reactivestreams.Subscriber<T>

onNext

public void onNext(T t)

指定者:

onNext 在接口中 org.reactivestreams.Subscriber<T>

onSubscribe

public void onSubscribe(org.reactivestreams.Subscription s)

从接口复制的说明: CoreSubscriber

Implementors should initialize any state used by Subscriber.onNext(Object) before calling Subscription.request(long). Should further onNext related state modification occur, thread-safety will be required.

Note that an invalid request <= 0 will not produce an onError and will simply be ignored or reported through a debug-enabled Logger.

指定者:

onSubscribe 在接口中 org.reactivestreams.Subscriber<T>

指定者:

onSubscribe 在接口中 CoreSubscriber<T>

request

public void request(long n)

指定者:

request 在接口中 org.reactivestreams.Subscription

currentContext

@Nonnull
public Context currentContext()

从接口复制的说明: CoreSubscriber

Request a Context from dependent components which can include downstream operators during subscribing or a terminal Subscriber.

指定者:

currentContext 在接口中 CoreSubscriber<T>

返回:

a resolved context or Context.empty()

requestedFusionMode

public final AssertSubscriber<T> requestedFusionMode(int requestMode)

Setup what fusion mode should be requested from the incoming Subscription if it happens to be QueueSubscription

参数:

requestMode - the mode to request, see Fuseable constants

返回:

this

upstream

public org.reactivestreams.Subscription upstream()

values

public java.util.List<T> values()

assertNoEvents

public final AssertSubscriber<T> assertNoEvents()

assertIncomplete

@SafeVarargs
public final AssertSubscriber<T> assertIncomplete(T... values)