Recently I found myself wondering about the ways of combining two awesome libraries: Akka and RxJava. The most straightforward use case for me was about converting a stream of messages received by an Actor into an Observable.
I’ll quickly write down how that works, and maybe this will be the start of a new little series of blogs. And maybe, I’ll learn something along the way about how to best create an Akka module for RxJava, something I’d love to do…
But I’ll start very very small. This is not going to go into any kind of error handling, or explore the Akka Event Bus, for example. – All this will have to wait for later.
First, I’ll need a small build.sbt that’s mostly there to import the two dependencies for me:
Next, I’ll need to define a simple message…
…and a basic actor that receives them:
Now let’s add a simple Main and try it out:
Up to now, this is pure Akka, and of course it works as expected. Now, for making an Observable out of this, I’ll need a mechanism for subscribing and unsubscribing. For subscribing, I want to be able to specify a callback onNext that will be called each time a message is received.
This means that I need a few more messages:
My actor has to handle these new messages, of course. I’ll use the Become mechanism for this in order to avoid any kind of mutable state. And very important: I want to call the new callback upon receiving messages instead of just logging them. All this means that the new version of the actor looks like this:
At last, I’m prepared to create an Observable. This is the step where the “magic” happens. I’ll add a small utility function for this:
What happens here is that in order to create an Observable, I need a function that takes an Observer (which provides the required callback(s) and wants to get notified about messages) and returns a Subscription (for unsubscribing).
On subscription, we send the Subscribe message to the actor, telling it about the callback. And on unsubscription, we send the Unsubscribe message. That’s all there is to it.
Let’s extend our Main object to try it out:
Aaaand it still does it’s thing. I’m good. For now.
In conclusion, I have to say that this was surprisingly easy. Of course, maybe that’s just because I left out all the more interesting things…
Thanks for reading! :-)
For reference, here’s the complete code of the example: