Package akka.stream.testkit

Class 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.

    • Method Detail

      • expectNext

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

        Expect multiple stream elements.

      • expectNextUnordered

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

        Expect multiple stream elements in arbitrary order.

      • expectSubscription

        public org.reactivestreams.Subscription expectSubscription()
        Expect and return a Subscription.

      • expectNext

        public I expectNext()
        Expect and return a stream element.

      • expectNext

        public I expectNext​(scala.concurrent.duration.FiniteDuration d)
        Expect and return a stream element during specified time or timeout.

      • expectNext

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

        Expect a stream element during specified time or timeout.

      • expectNext

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

        Expect multiple stream elements.

      • expectNextUnordered

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

        Expect multiple stream elements in arbitrary order.

      • expectNextN

        public scala.collection.immutable.Seq<I> expectNextN​(long n)
        Expect and return the next n stream elements.

      • expectNextN

        public TestSubscriber.ManualProbe expectNextN​(scala.collection.immutable.Seq<I> all)
        Fluent DSL Expect the given elements to be signalled in order.

      • expectNextUnorderedN

        public TestSubscriber.ManualProbe expectNextUnorderedN​(scala.collection.immutable.Seq<I> all)
        Fluent DSL Expect the given elements to be signalled in any order.

      • expectError

        public java.lang.Throwable expectError()
        Expect and return the signalled Throwable.

      • expectSubscriptionAndError

        public java.lang.Throwable expectSubscriptionAndError()
        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.

      • expectSubscriptionAndError

        public java.lang.Throwable expectSubscriptionAndError​(boolean signalDemand)
        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 the signalDemand parameter to false.

        See also expectSubscriptionAndError().

      • 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 {@link #expectSubscriptionAndComplete(cause: Throwable, signalDemand: Boolean)} if no demand should be signalled.

      • 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 {@link #expectSubscriptionAndError(cause: Throwable)}.

      • 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 {@link #expectSubscriptionAndComplete(signalDemand: Boolean)} if no demand should be signalled.

      • 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 immediately 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().

      • expectNextOrError

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

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

      • 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.

      • expectNextOrComplete

        public scala.util.Either<TestSubscriber.OnComplete$,​I> expectNextOrComplete()
        Expect next element or stream completion - returning whichever was signalled.

      • expectNextOrComplete

        public TestSubscriber.ManualProbe expectNextOrComplete​(I element)
        Fluent DSL

        Expect given next element or stream completion.

      • expectNoMessage

        public TestSubscriber.ManualProbe expectNoMessage​(scala.concurrent.duration.FiniteDuration remaining)
        Fluent DSL

        Assert that no message is received for the specified time.

      • expectNoMessage

        public TestSubscriber.ManualProbe expectNoMessage()
        Fluent DSL

        Assert that no message is received for the specified time. Waits for the default period configured as akka.test.expect-no-message-default. That timeout is scaled using the configuration entry "akka.test.timefactor".

      • expectNoMessage

        public TestSubscriber.ManualProbe expectNoMessage​(java.time.Duration remaining)
        Java API: Assert that no message is received for the specified time.

      • expectNextPF

        public <T> T expectNextPF​(scala.PartialFunction<java.lang.Object,​T> f)
        Expect a stream element and test it with partial function.

      • expectNextWithTimeoutPF

        public <T> T expectNextWithTimeoutPF​(scala.concurrent.duration.Duration max,
                                     scala.PartialFunction<java.lang.Object,​T> f)
        Expect a stream element and test it with partial function.

        Parameters:
        max - wait no more than max time, otherwise throw AssertionError

      • expectNextChainingPF

        public TestSubscriber.ManualProbe expectNextChainingPF​(scala.concurrent.duration.Duration max,
                                                       scala.PartialFunction<java.lang.Object,​java.lang.Object> f)
        Expect a stream element during specified time or timeout and test it with partial function.

        Allows chaining probe methods.

        Parameters:
        max - wait no more than max time, otherwise throw AssertionError

      • expectNextChainingPF

        public TestSubscriber.ManualProbe expectNextChainingPF​(scala.PartialFunction<java.lang.Object,​java.lang.Object> f)
        Expect a stream element during specified time or timeout and test it with partial function.

        Allows chaining probe methods.

      • expectEventWithTimeoutPF

        public <T> T expectEventWithTimeoutPF​(scala.concurrent.duration.Duration max,
                                      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.

      • receiveWhile$default$1

        public <T> scala.concurrent.duration.Duration receiveWhile$default$1()

      • receiveWhile$default$2

        public <T> scala.concurrent.duration.Duration receiveWhile$default$2()

      • receiveWhile$default$3

        public <T> int receiveWhile$default$3()

      • receiveWithin

        public scala.collection.immutable.Seq<I> receiveWithin​(scala.concurrent.duration.FiniteDuration max,
                                                       int messages)
        Drains a given number of messages

      • receiveWithin$default$2

        public int receiveWithin$default$2()

      • 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!'''

      • 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])
 }

      • within

        public <T> T within​(scala.concurrent.duration.FiniteDuration max,
                    scala.Function0<T> f)
        Same as calling within(0 seconds, max)(f).

      • 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>