What is the best way to create an Rx-Java Observable
from the classical Java event pattern? That is, given
class FooEvent { ... } interface FooListener { void fooHappened(FooEvent arg); } class Bar { public void addFooListener(FooListener l); public void removeFooListener(FooListener l); }
I want to implement
Observable<FooEvent> fooEvents(Bar bar);
The implementation I came up with is:
Observable<FooEvent> fooEvents(Bar bar) { return Observable.create(new OnSubscribeFunc<FooEvent>() { public Subscription onSubscribe(Observer<? super FooEvent> obs) { FooListener l = new FooListener() { public void fooHappened(FooEvent arg) { obs.onNext(arg); } }; bar.addFooListener(l); return new Subscription() { public void unsubscribe() { bar.removeFooListener(l); } }; } }); }
However, I don't really like it:
it's quite verbose;
requires a listener per Observer
(ideally there should be no listeners if there are no observers, and one listener otherwise). This can be improved by keeping an observer count as a field in the OnSubscribeFunc
, incrementing it on subscribe and decrementing on unsubscribe.
Is there a better solution?
Requirements:
Working with existing implementations of event patterns without changing them (if I was controlling that code, I could already write it to return Observable
I needed).
Getting compiler errors if/when the source API changes. No working with Object
instead of actual event argument type or with property name strings.
just() This is one of the easiest and convenient ways to create observable. just() constructs a reactive type by taking a pre-existing object and emitting that specific object to the downstream consumer upon subscription. The just operator converts an item into an Observable that emits that item.
Transforming Single to Observable is simple, as Single satisfies Observable's contract. Just call single. toObservable() and you're good.
There are two main methods to create Observables in RxJS. Subjects and Operators. We will take a look at both of these!
Ans: The Observable class and the Observer interface have been deprecated in Java 9 because the event model supported by Observer and Observable is quite limited, the order of notifications delivered by Observable is unspecified, and state changes are not in one-for-one correspondence with notifications.
I don't think there's a way to create a generic observable for every possible event, but you can certainly use them wherever you need.
The RxJava source has some handy examples of how to create observables from mouse events, button events, etc. Take a look at this class, which creates them from KeyEvents: KeyEventSource.java.
Your implementation is absolutely correct.
it's quite verbose
It gets much less verbose with lambdas (example for RxJava 2):
Observable<FooEvent> fooEvents(Bar bar) { return Observable.create(emitter -> { FooListener listener = event -> emitter.onNext(event); bar.addFooListener(listener); emitter.setCancellable(() -> bar.removeFooListener(listener)); }); }
ideally there should be no listeners if there are no observers, and one listener otherwise
You can use share()
operator, which makes your observable hot, i.e. all subscribers share single subscription. It automatically subscribes with the first subscriber, and unsubscribes when last one unsubscribes:
fooEvents(bar).share()
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With