Creating Observable from normal Java events

Question

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:

1. it's quite verbose;

2. 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:

1. 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).

2. Getting compiler errors if/when the source API changes. No working with Object instead of actual event argument type or with property name strings.

Answer 1

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.

Answer 2

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()