Akka Streams - combine latest operation

0
votes

I'd like to combine-latest with Akka Streams as described here.

I can't figure out how to do it - please help!

Thanks, Ryan.

1
scalaakkarx-javaakka-streamrx-scala
Could you give some more specifics? What is the input, what is the consumer. Here I have an interop benchmark that shows how RxJava 2 can work with Akka-Stream: github.com/akarnokd/akarnokd-misc/blob/master/src/jmh/java/hu/… - akarnokd
What kind of specifics? I'd like to be able to take two streams exactly like the zip function would, but instead of zip semantics I'm after the combine-latest semantics that I linked. Checked you link and I'm not sure how it helps me? Perhaps we're talking at cross purposes? - Ryan Worsley
I thought you want to combine two Akka Streams but it lacks the operator so you'd want to reuse RxJava's combineLatest operator. - akarnokd
Oh no, I don't want a dependency on RxJava, I just want to achieve the same by extending Akka Streams, or ideally, by using existing functionality. - Ryan Worsley
Did u find the answer? - NieMaszNic

1 Answers

2
votes

I just implemented it quickly. Not sure if its bugless, but worth a try :) https://gist.github.com/tg44/2e75d45c234ca02d91cfdac35f41a5a2 Comments under the gist is welcomed!

As we spoke on the gitter channel, it can't achieved with build in stages, but you can write the functionality with a custom stage. You will need two input and one output (can be extend to N input), so it's a fan in shape.

I save the incoming elements to options, and whenever an input is ready (aka. send an element) I save the given element to the option. Whenever the output need an element (and we already have one element from both inputs) I give it the values from the options as a tupple. This is the backpressure aware approach.

For a backpressured approach (where you produce all the pairs) you need to handle the waiting for "other" output element then the last one, and need to handle the input pulls. I think my implementation still not handle the too fast producers with slow consumer case (we can miss one element, can be handle with emits), and can deadlocked if both input produces the same element multiple times (maybe emits can handle this too).

If you want to extend my code functionality or want to write other custom stages read this: http://doc.akka.io/docs/akka/2.5/scala/stream/stream-customize.html