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.
