akka.stream.testkit

Class TestSubscriber.ManualProbe<I>

java.lang.Object

akka.stream.testkit.TestSubscriber.ManualProbe<I>

All Implemented Interfaces:

org.reactivestreams.Subscriber<I>

Direct Known Subclasses:

TestSubscriber.Probe

Enclosing class:

TestSubscriber

public static class TestSubscriber.ManualProbe<I>
extends java.lang.Object
implements org.reactivestreams.Subscriber<I>

Implementation of Subscriber that allows various assertions.

All timeouts are dilated automatically, for more details about time dilation refer to TestKit.

Constructor Summary

Constructors

Constructor and Description
ManualProbe(ActorSystem system)

Method Summary

Modifier and Type	Method and Description
TestSubscriber.ManualProbe	expectComplete() Fluent DSL
java.lang.Throwable	expectError() Expect and return the signalled Throwable.
TestSubscriber.ManualProbe	expectError(java.lang.Throwable cause) Fluent DSL
TestSubscriber.SubscriberEvent	expectEvent() Expect and return TestSubscriber.SubscriberEvent (any of: OnSubscribe, OnNext, OnError or OnComplete).
TestSubscriber.SubscriberEvent	expectEvent(scala.concurrent.duration.FiniteDuration max) Expect and return TestSubscriber.SubscriberEvent (any of: OnSubscribe, OnNext, OnError or OnComplete).
TestSubscriber.ManualProbe	expectEvent(TestSubscriber.SubscriberEvent event) Fluent DSL
<T> T	expectEventPF(scala.PartialFunction<TestSubscriber.SubscriberEvent,T> f)
I	expectNext() Expect and return a stream element.
I	expectNext(scala.concurrent.duration.FiniteDuration d) Expect and return a stream element during specified time or timeout.
TestSubscriber.ManualProbe	expectNext(scala.concurrent.duration.FiniteDuration d, I element) Fluent DSL
TestSubscriber.ManualProbe	expectNext(I element) Fluent DSL
TestSubscriber.ManualProbe	expectNext(I e1, I e2, java.lang.Object... es) Fluent DSL
TestSubscriber.ManualProbe	expectNext(I e1, I e2, scala.collection.Seq<I> es) Fluent DSL
TestSubscriber.ManualProbe	expectNextChainingPF(scala.PartialFunction<java.lang.Object,java.lang.Object> f) Expect next element and test it with partial function.
scala.collection.immutable.Seq<I>	expectNextN(long n) Expect and return the next n stream elements.
TestSubscriber.ManualProbe	expectNextN(scala.collection.immutable.Seq<I> all) Fluent DSL Expect the given elements to be signalled in order.
scala.util.Either<TestSubscriber.OnComplete$,I>	expectNextOrComplete() Expect next element or stream completion - returning whichever was signalled.
TestSubscriber.ManualProbe	expectNextOrComplete(I element) Fluent DSL
scala.util.Either<java.lang.Throwable,I>	expectNextOrError() Fluent DSL
scala.util.Either<java.lang.Throwable,I>	expectNextOrError(I element, java.lang.Throwable cause) Fluent DSL Expect given next element or error signal.
<T> T	expectNextPF(scala.PartialFunction<java.lang.Object,T> f)
TestSubscriber.ManualProbe	expectNextUnordered(I e1, I e2, java.lang.Object... es) Fluent DSL
TestSubscriber.ManualProbe	expectNextUnordered(I e1, I e2, scala.collection.Seq<I> es) Fluent DSL
TestSubscriber.ManualProbe	expectNextUnorderedN(scala.collection.immutable.Seq<I> all) Fluent DSL Expect the given elements to be signalled in any order.
TestSubscriber.ManualProbe	expectNoMsg() Fluent DSL
TestSubscriber.ManualProbe	expectNoMsg(scala.concurrent.duration.FiniteDuration remaining) Fluent DSL
org.reactivestreams.Subscription	expectSubscription() Expect and return a Subscription.
TestSubscriber.ManualProbe	expectSubscriptionAndComplete() Fluent DSL
TestSubscriber.ManualProbe	expectSubscriptionAndComplete(boolean signalDemand) Fluent DSL
java.lang.Throwable	expectSubscriptionAndError() Expect subscription to be followed immediatly by an error signal.
java.lang.Throwable	expectSubscriptionAndError(boolean signalDemand) Expect subscription to be followed immediatly by an error signal.
TestSubscriber.ManualProbe	expectSubscriptionAndError(java.lang.Throwable cause) Fluent DSL
TestSubscriber.ManualProbe	expectSubscriptionAndError(java.lang.Throwable cause, boolean signalDemand) Fluent DSL
void	onComplete()
void	onError(java.lang.Throwable cause)
void	onNext(I element)
void	onSubscribe(org.reactivestreams.Subscription subscription)
<T> scala.collection.immutable.Seq<T>	receiveWhile(scala.concurrent.duration.Duration max, scala.concurrent.duration.Duration idle, int messages, scala.PartialFunction<TestSubscriber.SubscriberEvent,T> f) Receive messages for a given duration or until one does not match a given partial function.
scala.collection.immutable.Seq<I>	receiveWithin(scala.concurrent.duration.FiniteDuration max, int messages) Drains a given number of messages
scala.collection.immutable.Seq<I>	toStrict(scala.concurrent.duration.FiniteDuration atMost) Attempt to drain the stream into a strict collection (by requesting Long.MaxValue elements).
<T> T	within(scala.concurrent.duration.FiniteDuration min, scala.concurrent.duration.FiniteDuration max, scala.Function0<T> f) Execute code block while bounding its execution time between min and max.
<T> T	within(scala.concurrent.duration.FiniteDuration max, scala.Function0<T> f) Same as calling within(0 seconds, max)(f).

Methods inherited from class java.lang.Object

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

Constructor Detail

ManualProbe

public ManualProbe(ActorSystem system)

Method Detail

expectNext

public TestSubscriber.ManualProbe expectNext(I e1,
                                             I e2,
                                             java.lang.Object... es)

Fluent DSL

Expect multiple stream elements.

Parameters:

e1 - (undocumented)

e2 - (undocumented)

es - (undocumented)

Returns:

(undocumented)

expectNextUnordered

public TestSubscriber.ManualProbe expectNextUnordered(I e1,
                                                      I e2,
                                                      java.lang.Object... es)

Fluent DSL

Expect multiple stream elements in arbitrary order.

Parameters:

e1 - (undocumented)

e2 - (undocumented)

es - (undocumented)

Returns:

(undocumented)

expectSubscription

public org.reactivestreams.Subscription expectSubscription()

Expect and return a Subscription.

Returns:

(undocumented)

expectEvent

public TestSubscriber.SubscriberEvent expectEvent()

Expect and return TestSubscriber.SubscriberEvent (any of: OnSubscribe, OnNext, OnError or OnComplete).

Returns:

(undocumented)

expectEvent

public TestSubscriber.SubscriberEvent expectEvent(scala.concurrent.duration.FiniteDuration max)

Expect and return TestSubscriber.SubscriberEvent (any of: OnSubscribe, OnNext, OnError or OnComplete).

Parameters:

max - (undocumented)

Returns:

(undocumented)

expectEvent

public TestSubscriber.ManualProbe expectEvent(TestSubscriber.SubscriberEvent event)

Fluent DSL

Expect TestSubscriber.SubscriberEvent (any of: OnSubscribe, OnNext, OnError or OnComplete).

Parameters:

event - (undocumented)

Returns:

(undocumented)

expectNext

public I expectNext()

Expect and return a stream element.

Returns:

(undocumented)

expectNext

public I expectNext(scala.concurrent.duration.FiniteDuration d)

Expect and return a stream element during specified time or timeout.

Parameters:

d - (undocumented)

Returns:

(undocumented)

expectNext

public TestSubscriber.ManualProbe expectNext(I element)

Fluent DSL

Expect a stream element.

Parameters:

element - (undocumented)

Returns:

(undocumented)

expectNext

public TestSubscriber.ManualProbe expectNext(scala.concurrent.duration.FiniteDuration d,
                                             I element)

Fluent DSL

Expect a stream element during specified time or timeout.

Parameters:

d - (undocumented)

element - (undocumented)

Returns:

(undocumented)

expectNext

public TestSubscriber.ManualProbe expectNext(I e1,
                                             I e2,
                                             scala.collection.Seq<I> es)

Fluent DSL

Expect multiple stream elements.

Parameters:

e1 - (undocumented)

e2 - (undocumented)

es - (undocumented)

Returns:

(undocumented)

expectNextUnordered

public TestSubscriber.ManualProbe expectNextUnordered(I e1,
                                                      I e2,
                                                      scala.collection.Seq<I> es)

Fluent DSL

Expect multiple stream elements in arbitrary order.

Parameters:

e1 - (undocumented)

e2 - (undocumented)

es - (undocumented)

Returns:

(undocumented)

expectNextN

public scala.collection.immutable.Seq<I> expectNextN(long n)

Expect and return the next n stream elements.

Parameters:

n - (undocumented)

Returns:

(undocumented)

expectNextN

public TestSubscriber.ManualProbe expectNextN(scala.collection.immutable.Seq<I> all)

Fluent DSL Expect the given elements to be signalled in order.

Parameters:

all - (undocumented)

Returns:

(undocumented)

expectNextUnorderedN

public TestSubscriber.ManualProbe expectNextUnorderedN(scala.collection.immutable.Seq<I> all)

Fluent DSL Expect the given elements to be signalled in any order.

Parameters:

all - (undocumented)

Returns:

(undocumented)

expectComplete

public TestSubscriber.ManualProbe expectComplete()

Fluent DSL

Expect completion.

Returns:

(undocumented)

expectError

public java.lang.Throwable expectError()

Expect and return the signalled Throwable.

Returns:

(undocumented)

expectError

public TestSubscriber.ManualProbe expectError(java.lang.Throwable cause)

Fluent DSL

Expect given Throwable.

Parameters:

cause - (undocumented)

Returns:

(undocumented)

expectSubscriptionAndError

public java.lang.Throwable expectSubscriptionAndError()

Expect subscription to be followed immediatly by an error signal.

By default 1 demand will be signalled in order to wake up a possibly lazy upstream.

See also #expectSubscriptionAndError(Boolean) if no demand should be signalled.

Returns:

(undocumented)

expectSubscriptionAndError

public java.lang.Throwable expectSubscriptionAndError(boolean signalDemand)

Expect subscription to be followed immediatly by an error signal.

Depending on the signalDemand parameter demand may be signalled immediatly after obtaining the subscription in order to wake up a possibly lazy upstream. You can disable this by setting the signalDemand parameter to false.

See also expectSubscriptionAndError().

Parameters:

signalDemand - (undocumented)

Returns:

(undocumented)

expectSubscriptionAndError

public TestSubscriber.ManualProbe expectSubscriptionAndError(java.lang.Throwable cause)

Fluent DSL

Expect subscription followed by immediate stream completion.

By default 1 demand will be signalled in order to wake up a possibly lazy upstream.

See also #expectSubscriptionAndComplete(cause: Throwable, signalDemand: Boolean) if no demand should be signalled.

Parameters:

cause - (undocumented)

Returns:

(undocumented)

expectSubscriptionAndError

public TestSubscriber.ManualProbe expectSubscriptionAndError(java.lang.Throwable cause,
                                                             boolean signalDemand)

Fluent DSL

Expect subscription followed by immediate stream completion. By default 1 demand will be signalled in order to wake up a possibly lazy upstream

See also #expectSubscriptionAndError(cause: Throwable).

Parameters:

cause - (undocumented)

signalDemand - (undocumented)

Returns:

(undocumented)

expectSubscriptionAndComplete

public TestSubscriber.ManualProbe expectSubscriptionAndComplete()

Fluent DSL

Expect subscription followed by immediate stream completion. By default 1 demand will be signalled in order to wake up a possibly lazy upstream

See also #expectSubscriptionAndComplete(signalDemand: Boolean) if no demand should be signalled.

Returns:

(undocumented)

expectSubscriptionAndComplete

public TestSubscriber.ManualProbe expectSubscriptionAndComplete(boolean signalDemand)

Fluent DSL

Expect subscription followed by immediate stream completion.

Depending on the signalDemand parameter demand may be signalled immediatly after obtaining the subscription in order to wake up a possibly lazy upstream. You can disable this by setting the signalDemand parameter to false.

See also expectSubscriptionAndComplete().

Parameters:

signalDemand - (undocumented)

Returns:

(undocumented)

expectNextOrError

public scala.util.Either<java.lang.Throwable,I> expectNextOrError()

Fluent DSL

Expect given next element or error signal, returning whichever was signalled.

Returns:

(undocumented)

expectNextOrError

public scala.util.Either<java.lang.Throwable,I> expectNextOrError(I element,
                                                                  java.lang.Throwable cause)

Fluent DSL Expect given next element or error signal.

Parameters:

element - (undocumented)

cause - (undocumented)

Returns:

(undocumented)

expectNextOrComplete

public scala.util.Either<TestSubscriber.OnComplete$,I> expectNextOrComplete()

Expect next element or stream completion - returning whichever was signalled.

Returns:

(undocumented)

expectNextOrComplete

public TestSubscriber.ManualProbe expectNextOrComplete(I element)

Fluent DSL

Expect given next element or stream completion.

Parameters:

element - (undocumented)

Returns:

(undocumented)

expectNoMsg

public TestSubscriber.ManualProbe expectNoMsg()

Fluent DSL

Same as expectNoMsg(remaining), but correctly treating the timeFactor.

Returns:

(undocumented)

expectNoMsg

public TestSubscriber.ManualProbe expectNoMsg(scala.concurrent.duration.FiniteDuration remaining)

Fluent DSL

Assert that no message is received for the specified time.

Parameters:

remaining - (undocumented)

Returns:

(undocumented)

expectNextPF

public <T> T expectNextPF(scala.PartialFunction<java.lang.Object,T> f)

expectNextChainingPF

public TestSubscriber.ManualProbe expectNextChainingPF(scala.PartialFunction<java.lang.Object,java.lang.Object> f)

Expect next element and test it with partial function.

Allows chaining probe methods.

Parameters:

f - (undocumented)

Returns:

(undocumented)

expectEventPF

public <T> T expectEventPF(scala.PartialFunction<TestSubscriber.SubscriberEvent,T> f)

receiveWhile

public <T> scala.collection.immutable.Seq<T> receiveWhile(scala.concurrent.duration.Duration max,
                                                          scala.concurrent.duration.Duration idle,
                                                          int messages,
                                                          scala.PartialFunction<TestSubscriber.SubscriberEvent,T> f)

Receive messages for a given duration or until one does not match a given partial function.

Parameters:

max - (undocumented)

idle - (undocumented)

messages - (undocumented)

f - (undocumented)

Returns:

(undocumented)

receiveWithin

public scala.collection.immutable.Seq<I> receiveWithin(scala.concurrent.duration.FiniteDuration max,
                                                       int messages)

Drains a given number of messages

Parameters:

max - (undocumented)

messages - (undocumented)

Returns:

(undocumented)

toStrict

public scala.collection.immutable.Seq<I> toStrict(scala.concurrent.duration.FiniteDuration atMost)

Attempt to drain the stream into a strict collection (by requesting Long.MaxValue elements).

'''Use with caution: Be warned that this may not be a good idea if the stream is infinite or its elements are very large!'''

Parameters:

atMost - (undocumented)

Returns:

(undocumented)

within

public <T> T within(scala.concurrent.duration.FiniteDuration min,
                    scala.concurrent.duration.FiniteDuration max,
                    scala.Function0<T> f)

Execute code block while bounding its execution time between min and max. within blocks may be nested. All methods in this trait which take maximum wait times are available in a version which implicitly uses the remaining time governed by the innermost enclosing within block.

Note that the timeout is scaled using Duration.dilated, which uses the configuration entry "akka.test.timefactor", while the min Duration is not.

 val ret = within(50 millis) {
   test ! "ping"
   expectMsgClass(classOf[String])
 }
 

Parameters:

min - (undocumented)

max - (undocumented)

f - (undocumented)

Returns:

(undocumented)

within

public <T> T within(scala.concurrent.duration.FiniteDuration max,
                    scala.Function0<T> f)

Same as calling within(0 seconds, max)(f).

Parameters:

max - (undocumented)

f - (undocumented)

Returns:

(undocumented)

onSubscribe

public void onSubscribe(org.reactivestreams.Subscription subscription)

Specified by:

onSubscribe in interface org.reactivestreams.Subscriber<I>

onNext

public void onNext(I element)

Specified by:

onNext in interface org.reactivestreams.Subscriber<I>

onComplete

public void onComplete()

Specified by:

onComplete in interface org.reactivestreams.Subscriber<I>

onError

public void onError(java.lang.Throwable cause)

Specified by:

onError in interface org.reactivestreams.Subscriber<I>