OK, I understand the problem with implementing a perfectly predictable
merge preferred.
For completeness, I'll report the result for two sources which continuously
produce data as fast as possible: the order of elements after the merge is:
121211111...22222.
So the merge preferred behaves un-intuitively only for the first 4
elements. This result is confirmed for any number of elements, it's always
exactly the same.
Anyway, thank you for your time and explanations.
On Monday, January 18, 2016 at 11:37:02 AM UTC+1, Quizzie Fogg wrote:
>
> Hi.
>
> I think MergePreferred might not be working correctly in streams version
> 2.0._
>
> Old working test code (version 1.0):
> def mergePreferredMat[A, M1, M2, M3](sourceA: Source[A, M1], sourceB:
> Source[A, M2])(combine: (M1, M2) => M3): Source[A, M3] = {
> val merger = FlowGraph.partial(sourceB) { implicit b =>
> sourceB =>
> val merge = b.add(MergePreferred[A](1))
> sourceB ~> merge.in(0)
> FlowShape(merge.preferred, merge.out)
> }
> sourceA.viaMat(merger)(combine)
> }
>
> def testPreferred(): Unit = {
> val mergedSource = mergePreferredMat(TestPublisher.Probe[ByteString],
> TestPublisher.Probe[ByteString])(Keep.both)
> val (in1, in2, out): (TestPublisher.Probe[ByteString],
> TestPublisher.Probe[ByteString], TestSubscriber.Probe[ByteString]) =
> mergedSource.toMat(TestSubscriber.Probe[ByteString]) {
> case ((in1, in2), out) => (in1, in2, out)
> }.run()
>
> in2.sendNext(data2)
> in1.sendNext(data1)
> in2.sendNext(data2)
> in1.sendNext(data1)
>
> out.request(4)
> out.expectNext(data1)
> out.expectNext(data1)
> out.expectNext(data2)
> out.expectNext(data2)
> }
> The data is received in the correct order: first all the data from the
> preferred port 1, then all data from port 2.
>
> But in version 2.0._ (tested in 2.0.1 and 2.02) this is the test that
> passes:
> def mergePreferredMat[A, M1, M2, M3](sourceA: Source[A, M1], sourceB:
> Source[A, M2])(combine: (M1, M2) => M3): Source[A, M3] = {
> val merger = GraphDSL.create(sourceB) { implicit b =>
> sourceB =>
> val merge = b.add(MergePreferred[A](1))
> sourceB ~> merge.in(0)
> FlowShape(merge.preferred, merge.out)
> }
> sourceA.viaMat(merger)(combine)
> }
>
> def testPreferred(): Unit = {
> val mergedSource = mergePreferredMat(TestPublisher.Probe[ByteString],
> TestPublisher.Probe[ByteString])(Keep.both)
> val (in1, in2, out): (TestPublisher.Probe[ByteString],
> TestPublisher.Probe[ByteString], TestSubscriber.Probe[ByteString]) =
> mergedSource.toMat(TestSubscriber.Probe[ByteString]) {
> case ((in1, in2), out) => (in1, in2, out)
> }.run()
>
> in2.sendNext(data2)
> in1.sendNext(data1)
> in2.sendNext(data2)
> in1.sendNext(data1)
>
> out.request(4)
> out.expectNext(data1)
> out.expectNext(data2)
> out.expectNext(data1)
> out.expectNext(data2)
> }
> So the first piece of data is taken from the preferred port 1, but still
> ports 1 and 2 take turns instead of draining the preferred one completely
> first.
>
> Question: was the definition of merge preferred changed, is this a bug, or
> am I doing something wrong?
>
>
>
--
>>>>>>>>>> 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 [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
