Yahoo Pipes was a nice project, but as its popularity grew, it started getting blocked more and more. It was also hard to build and maintain pipelines with more than a few steps.
[1] https://github.com/olviko/RssPercolator
[2] https://github.com/olviko/RssPercolator/blob/master/RssPerco...
But since generators can only be "pulled" into one destination, you have to copy a stream (subsequently converting it into a list) if you want more than one destination [5]. This works fine if the data can fit in memory, but if it can't then you're out of luck!
[1] http://www.dabeaz.com/coroutines/copipe.py
[2] http://www.dabeaz.com/coroutines/
[3] http://www.dabeaz.com/generators/retuple.py
[4] http://www.dabeaz.com/generators
[5] https://github.com/nerevu/riko/blob/master/riko/modules/spli...
But, I see what you mean. I had to deal with similar issues in commercial projects and the "pull" model (generators in Python ~ "yield return" in C#) almost never a good idea, especially when you have to have concurrent consumers. While callbacks are hard to combine, in C# it can be nicely abstracted with “async/await”, not sure how it is handled in Python, I stopped using it around 2.5
I've been working on a similar project and I've also found the push model easier.
1. The code is tiny with 90% of it dealing with RSS parsing and filtering. Using RX.NET wouldn't really simplify anything.
2. I wanted a library that I can integrate into my apps and run locally to avoid throttling, robots.txt and other BS Yahoo Pipes was suffering from.
I am also not a huge fan of RX... to put it mildly
Mind sharing why you don't like it ?
I personally like how it allows me to express complex high level operations cleanly. For example - I have a observable configuration variable that can come from different sources and the source change dynamically. I need to listen to latest source until a new one becomes active - in Rx I only need to push the new source trough IObservable<IObservable<ConfigurationValue>> and then use http://reactivex.io/documentation/operators/switch.html which returns IObservable<ConfigurationValue> which will push values from the latest source - Rx will handle unsubscribing from previous active source, synchronizing state and making sure everything is thread safe. And there are a bunch of operators like this that would be tedious and hard implement correctly with all the edge cases in a thread safe way - and here they are abstracted in to high level operators.