class ManualProbe[I] extends Subscriber[I]
Implementation of org.reactivestreams.Subscriber that allows various assertions.
All timeouts are dilated automatically, for more details about time dilation refer to akka.testkit.TestKit.
- Source
- StreamTestKit.scala
- Alphabetic
- By Inheritance
- ManualProbe
- Subscriber
- AnyRef
- Any
- by any2stringadd
- by StringFormat
- by Ensuring
- by ArrowAssoc
- Hide All
- Show All
- Public
- All
Type Members
- abstract type Self <: ManualProbe[I]
Value Members
-
def
expectComplete(): Self
Fluent DSL
Fluent DSL
Expect completion.
-
def
expectError(cause: Throwable): Self
Fluent DSL
Fluent DSL
Expect given Throwable.
-
def
expectError(): Throwable
Expect and return the signalled Throwable.
-
def
expectEvent(event: SubscriberEvent): Self
Fluent DSL
Fluent DSL
Expect SubscriberEvent (any of:
OnSubscribe
,OnNext
,OnError
orOnComplete
). -
def
expectEvent(max: FiniteDuration): SubscriberEvent
Expect and return SubscriberEvent (any of:
OnSubscribe
,OnNext
,OnError
orOnComplete
). -
def
expectEvent(): SubscriberEvent
Expect and return SubscriberEvent (any of:
OnSubscribe
,OnNext
,OnError
orOnComplete
). - def expectEventPF[T](f: PartialFunction[SubscriberEvent, T]): T
- def expectEventWithTimeoutPF[T](max: Duration, f: PartialFunction[SubscriberEvent, T]): T
-
def
expectNext(e1: I, e2: I, es: I*): Self
Fluent DSL
Fluent DSL
Expect multiple stream elements.
- Annotations
- @varargs()
-
def
expectNext(d: FiniteDuration, element: I): Self
Fluent DSL
Fluent DSL
Expect a stream element during specified time or timeout.
-
def
expectNext(element: I): Self
Fluent DSL
Fluent DSL
Expect a stream element.
-
def
expectNext(d: FiniteDuration): I
Expect and return a stream element during specified time or timeout.
-
def
expectNext(): I
Expect and return a stream element.
-
def
expectNextChainingPF(f: PartialFunction[Any, Any]): Self
Expect a stream element during specified time or timeout and test it with partial function.
Expect a stream element during specified time or timeout and test it with partial function.
Allows chaining probe methods.
-
def
expectNextChainingPF(max: Duration, f: PartialFunction[Any, Any]): Self
Expect a stream element during specified time or timeout and test it with partial function.
Expect a stream element during specified time or timeout and test it with partial function.
Allows chaining probe methods.
- max
wait no more than max time, otherwise throw AssertionError
-
def
expectNextN(all: Seq[I]): Self
Fluent DSL Expect the given elements to be signalled in order.
-
def
expectNextN(n: Long): Seq[I]
Expect and return the next
n
stream elements. -
def
expectNextOrComplete(element: I): Self
Fluent DSL
Fluent DSL
Expect given next element or stream completion.
-
def
expectNextOrComplete(): Either[OnComplete.type, I]
Expect next element or stream completion - returning whichever was signalled.
-
def
expectNextOrError(element: I, cause: Throwable): Either[Throwable, I]
Fluent DSL Expect given next element or error signal.
-
def
expectNextOrError(): Either[Throwable, I]
Fluent DSL
Fluent DSL
Expect given next element or error signal, returning whichever was signalled.
-
def
expectNextPF[T](f: PartialFunction[Any, T]): T
Expect a stream element and test it with partial function.
-
def
expectNextUnordered(e1: I, e2: I, es: I*): Self
Fluent DSL
Fluent DSL
Expect multiple stream elements in arbitrary order.
- Annotations
- @varargs()
-
def
expectNextUnorderedN(all: Seq[I]): Self
Fluent DSL Expect the given elements to be signalled in any order.
-
def
expectNextWithTimeoutPF[T](max: Duration, f: PartialFunction[Any, T]): T
Expect a stream element and test it with partial function.
Expect a stream element and test it with partial function.
- max
wait no more than max time, otherwise throw AssertionError
-
def
expectNoMessage(remaining: Duration): Self
Java API: Assert that no message is received for the specified time.
-
def
expectNoMessage(remaining: FiniteDuration): Self
Fluent DSL
Fluent DSL
Assert that no message is received for the specified time.
-
def
expectSubscription(): Subscription
Expect and return a org.reactivestreams.Subscription.
-
def
expectSubscriptionAndComplete(signalDemand: Boolean): Self
Fluent DSL
Fluent DSL
Expect subscription followed by immediate stream completion.
Depending on the
signalDemand
parameter demand may be signalled immediately after obtaining the subscription in order to wake up a possibly lazy upstream. You can disable this by setting thesignalDemand
parameter tofalse
.See also #expectSubscriptionAndComplete.
-
def
expectSubscriptionAndComplete(): Self
Fluent DSL
Fluent DSL
Expect subscription followed by immediate stream completion. By default
1
demand will be signalled in order to wake up a possibly lazy upstreamSee also Boolean) if no demand should be signalled.
-
def
expectSubscriptionAndError(cause: Throwable, signalDemand: Boolean): Self
Fluent DSL
Fluent DSL
Expect subscription followed by immediate stream completion. By default
1
demand will be signalled in order to wake up a possibly lazy upstreamSee also Throwable).
-
def
expectSubscriptionAndError(cause: Throwable): Self
Fluent DSL
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 Throwable, signalDemand: Boolean) if no demand should be signalled.
-
def
expectSubscriptionAndError(signalDemand: Boolean): Throwable
Expect subscription to be followed immediately by an error signal.
Expect subscription to be followed immediately by an error signal.
Depending on the
signalDemand
parameter demand may be signalled immediately after obtaining the subscription in order to wake up a possibly lazy upstream. You can disable this by setting thesignalDemand
parameter tofalse
.See also #expectSubscriptionAndError().
-
def
expectSubscriptionAndError(): Throwable
Expect subscription to be followed immediately by an error signal.
Expect subscription to be followed immediately 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.
-
def
onComplete(): Unit
- Definition Classes
- ManualProbe → Subscriber
-
def
onError(cause: Throwable): Unit
- Definition Classes
- ManualProbe → Subscriber
-
def
onNext(element: I): Unit
- Definition Classes
- ManualProbe → Subscriber
-
def
onSubscribe(subscription: Subscription): Unit
- Definition Classes
- ManualProbe → Subscriber
-
def
receiveWhile[T](max: Duration = Duration.Undefined, idle: Duration = Duration.Inf, messages: Int = Int.MaxValue)(f: PartialFunction[SubscriberEvent, T]): Seq[T]
Receive messages for a given duration or until one does not match a given partial function.
-
def
receiveWithin(max: FiniteDuration, messages: Int = Int.MaxValue): Seq[I]
Drains a given number of messages
-
def
toStrict(atMost: FiniteDuration): Seq[I]
Attempt to drain the stream into a strict collection (by requesting
Long.MaxValue
elements).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!
-
def
within[T](max: FiniteDuration)(f: ⇒ T): T
Same as calling
within(0 seconds, max)(f)
. -
def
within[T](min: FiniteDuration, max: FiniteDuration)(f: ⇒ T): T
Execute code block while bounding its execution time between
min
andmax
.Execute code block while bounding its execution time between
min
andmax
.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 enclosingwithin
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]) }
Deprecated Value Members
-
def
expectNoMsg(remaining: FiniteDuration): Self
Fluent DSL
Fluent DSL
Assert that no message is received for the specified time. NOTE! Timeout value is automatically multiplied by timeFactor.
- Annotations
- @deprecated
- Deprecated
(Since version 2.5.5) Use expectNoMessage instead
-
def
expectNoMsg(): Self
Fluent DSL
Fluent DSL
Same as
expectNoMsg(remaining)
, but correctly treating the timeFactor. NOTE! Timeout value is automatically multiplied by timeFactor.- Annotations
- @deprecated
- Deprecated
(Since version 2.5.5) Use expectNoMessage instead