I'm not familiar with the API you're using, but it...

# rx

[tschuchort]

I'm not familiar with the API you're using, but it's imperative, so you may not be able to translate it completely into an elegant functional stream