RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
@mlatu Why do you need map+filter? Just filter can already do the filtering and possible data manipulations (just nothing persistent).
flatMap is for one to many tranformations, map for one to one.
About preserving order, do you really need that in a reactive system? concatMap also evaluate everything serial i think, flatMap has the possibility to go concurrent if you need it.
Identical code, just schematic difference.
Oh right, flatMap is one to zero or more, here it's used for 0 or 1 element. Map is also one on one.
Very hard to say what's faster. You're probably talking about nanoseconds here, but you can run it 10.000.000 times in a loop and see.
Just measure
Observable.range(10000000)./* map or flatmap */.subscribe(() -> endMeasure())
@mlatu
I do have the feeling the flatMap one could be written cleaner also.
o.flatMap(x -> Observalbe.just(f(x)).filter(y -> y != null) )
lol :)
map is still cleaner.
Do you know any article which explain the backpressure with Rx, and how to write a correct Producer ? I'm reading articles from the blog http://akarnokd.blogspot.fr/. But I'm only understood that writing a Producer is hard and complicated.
@dwursteisen Have you read https://github.com/ReactiveX/RxJava/wiki/Backpressure and http://reactivex.io/documentation/operators/backpressure.html ?
And there are more.
If you understand the operators, producers should be easier. Continue with none-backpresure producers and then backpresure shouldnt be too hard. Still i dont know any article for just that.
I may have not fully understood operators, as I don't feel that producers are easy :smile:
@DmitriyZaitsev Sorry for the late response. Though your code works just perfectly find, I am unable to understand what is
lock and how is it working here?
whst notifyAll doing?
@tusharmath
Since the reactive code executes asynchronously on a different thread, method
As soon as the Observable is completed, it unlocks the
lock.wait() causes the current thread to wait until another thread invokes lock.notify() or lock.notifyAll().Since the reactive code executes asynchronously on a different thread, method
main() can not be finished until rx.Observable emits elements.As soon as the Observable is completed, it unlocks the
lock object, which will cause the main thread to continue till the end.
Ah sorry, i got distracted. But looks like you already have an answer. I was going to point at the Java guarded blocks tutorial as these are Java methods available on every Object. The only scala part is using
new AnyRef instead of new Object().
How many threads would I have created in the above code?
is it one per
interval call?
your observables
a, b, c emit elements on different threads
every interval creates a new thread, which is quite essentially idle and waiting for a timeout. Eventually when it fires it goes to the main thread to get consumed?
However, the only thing it guarantees is that events run serialized, different events might still run parallel for different steps, and flatMap might even run everything concurrent (just that it starts serialized but if the inner observable has a delay, it already starts the next one).
Im not sure what the
computation does, but @DmitriyZaitsev might be right that it starts multiple threads.
As that's often optimal for computation work.
In the above example
Will the
map functions be called parallely ?