Hi Gabriel,
testing such a Graph as a black box is rather simple: just attach a Source
that emits the test inputs and attach a Sink to each output that collects
all emitted elements (e.g. using
Flow[...].grouped(1000).toMat(Sink.head)(Keep.right)). When running that
flow, the output Futures will eventually be completed and you can assert
that they contain the right collections.
def collector[T] = Flow[T].grouped(1000).toMat(Sink.head)(Keep.right)
val (out0, out1) = FlowGraph.closed(collector[Event], collector[Event]) {
implicit builder => (out0, out1) =>
val eif = builder.add(eventInputFlow)
Source(ev1, ev2, ev3) ~> eif.in
eif.out(0) ~> out0
eif.out(1) ~> out1
}.run()
Regards,
Roland
On Thu, Aug 13, 2015 at 1:44 PM, Gabriel Volpe <volpegabr...@gmail.com>
wrote:
> Hi Viktor,
>
> I've been reading the official docs (
> http://doc.akka.io/docs/akka-stream-and-http-experimental/1.0/scala/stream-testkit.html)
> and researching into the scala docs but it seems there is no way to test
> neither a FanInShape or a FanOutShape. The examples in the documentation
> are trivial (only testing a simple Flow).
>
> I tried to wrap the FanOutShape into two Partial Flow Shapes using Out(0)
> for one and Out(1) for the other one. But when I try to use this FlowShape
> (using Flow.wrap(theFlowShape)) connected to a Source and a Sink I got an
> error like this one:
>
> Element must not be null, rule 2.13
> java.lang.NullPointerException: Element must not be null, rule 2.13
> at
> akka.stream.impl.ReactiveStreamsCompliance$.elementMustNotBeNullException(ReactiveStreamsCompliance.scala:44)
> at
> akka.stream.impl.ReactiveStreamsCompliance$.requireNonNullElement(ReactiveStreamsCompliance.scala:70)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.push(Interpreter.scala:287)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.push(Interpreter.scala:434)
> at akka.stream.impl.fusing.Map.onPush(Ops.scala:24)
> at akka.stream.impl.fusing.Map.onPush(Ops.scala:23)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.run(Interpreter.scala:436)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.progress(Interpreter.scala:245)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$$anon$1.progress(Interpreter.scala:434)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter.akka$stream$impl$fusing$OneBoundedInterpreter$$execute(Interpreter.scala:580)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$State$class.execute(Interpreter.scala:241)
> at
> akka.stream.impl.fusing.OneBoundedInterpreter$EntryState.execute(Interpreter.scala:666)
> at akka.stream.stage.AbstractStage.enterAndPush(Stage.scala:66)
> at
> akka.stream.impl.fusing.BatchingActorInputBoundary$$anonfun$upstreamRunning$1.applyOrElse(ActorInterpreter.scala:157)
> at
> scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
> at akka.stream.impl.SubReceive.apply(Transfer.scala:16)
> at akka.stream.impl.SubReceive.apply(Transfer.scala:12)
> at scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
> at akka.stream.impl.SubReceive.applyOrElse(Transfer.scala:12)
> at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
> at akka.actor.Actor$class.aroundReceive(Actor.scala:467)
> at
> akka.stream.impl.fusing.ActorInterpreter.aroundReceive(ActorInterpreter.scala:366)
> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238)
> at akka.dispatch.Mailbox.run(Mailbox.scala:220)
> at
> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397)
> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> at
> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> at
> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> at
> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>
>
>
> El jueves, 13 de agosto de 2015, 12:09:45 (UTC+1), √ escribió:
>>
>> Hi Gabriel,
>>
>> what did you try and in what way doesn't it do what you expected?
>>
>> On Thu, Aug 13, 2015 at 12:39 PM, Gabriel Volpe <volpeg...@gmail.com>
>> wrote:
>>
>>> Hi hakkers!
>>>
>>> I found difficult to test a Partial Flow Graph. I can test the different
>>> single pieces, but not the whole partial flow. I'm separating the flows
>>> depending on it's functionality and I wish to test every partial flow like
>>> a black box.
>>>
>>> For instance, given the following partial flow:
>>>
>>> lazy val eventInputFlow = FlowGraph.partial() { implicit builder =>
>>> val headerEnricherFlow = builder.add(Flow[Event] map (e =>
>>> applyHeaders(e)))
>>> val eventFilter = builder.add(eventFilterFlow)
>>> val eventOriginFilter = builder.add(eventOriginFilterFlow)
>>> val eventDeletedLogger = builder.add(eventDeletedLoggerFlow)
>>> val sinkIgnore = builder.add(Sink.ignore)
>>>
>>> headerEnricherFlow ~> eventFilter
>>> eventFilter.out(0) ~> eventOriginFilter
>>> eventFilter.out(1) ~> eventDeletedLogger ~> sinkIgnore
>>>
>>> UniformFanOutShape(headerEnricherFlow.inlet, eventOriginFilter.out(0),
>>> eventOriginFilter.out(1))
>>> }.named("eventInputFlow")
>>>
>>> It would be nice to test the 1 Input and the 2 outputs of this graph.
>>>
>>> I don't really know how to achieve this.
>>>
>>> Any help would be appreciated.
>>>
>>> Thanks,
>>> Gabriel.
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ:
>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>> >>>>>>>>>> Search the archives:
>>> https://groups.google.com/group/akka-user
>>> ---
>>> You received this message because you are subscribed to the Google
>>> Groups "Akka User List" group.
>>> To unsubscribe from this group and stop receiving emails from it, send
>>> an email to akka-user+...@googlegroups.com.
>>> To post to this group, send email to akka...@googlegroups.com.
>>> Visit this group at http://groups.google.com/group/akka-user.
>>> For more options, visit https://groups.google.com/d/optout.
>>>
>>
>>
>>
>> --
>> Cheers,
>> √
>>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at http://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Akka Team
Typesafe - Reactive apps on the JVM
Blog: letitcrash.com
Twitter: @akkateam
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
