Создание наблюдаемых из обычных событий Java

Каков наилучший способ создания Rx-Java Observable из классического шаблона событий Java? То есть, учитывая

class FooEvent { ... }

interface FooListener {
  void fooHappened(FooEvent arg);
}

class Bar {
  public void addFooListener(FooListener l);
  public void removeFooListener(FooListener l);
}

Я хочу реализовать

Observable<FooEvent> fooEvents(Bar bar);

Реализация, с которой я столкнулся, это:

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);
        }
      };
    }
  }); 
}

Однако мне это не очень нравится:

он довольно многословный;
требуется слушатель на Observer (в идеале не должно быть ни одного слушателя, если в нем нет наблюдателей, а один слушатель - в противном случае). Это можно улучшить, сохранив счетчик наблюдателей в качестве поля в OnSubscribeFunc, увеличивая его на подписку и декремент при отписании.

Есть ли лучшее решение?

Требования:

Работа с существующими реализациями шаблонов событий без их изменения (если бы я контролировал этот код, я мог бы уже записать его для возврата Observable, который мне нужен).
Получение ошибок компилятора, если/при изменении исходного API. Не работает с Object вместо фактического типа аргумента события или с строками имен свойств.

Ответ 1

Я не думаю, что есть способ создать общий наблюдаемый для каждого возможного события, но вы можете, конечно, использовать их там, где вам нужно.

В источнике RxJava есть несколько удобных примеров создания наблюдаемых событий мыши, событий кнопок и т.д. Взгляните на этот класс, который создает их из KeyEvents: KeyEventSource.java.

Ответ 2

Ваша реализация абсолютно правильна.

это довольно многословный

Он становится намного менее подробным с lambdas (пример для 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));
    }); 
}

в идеале не должно быть слушателей, если нет наблюдателей, а один слушатель в противном случае

Вы можете использовать оператор share(), который делает ваши наблюдаемые горячие, т.е. все подписчики используют одну подписку. Он автоматически подписывается на первого абонента и отменяет подписку, если последний отменил подписку:

fooEvents(bar).share()

Ответ 3

Я предполагаю, что вы можете иметь тот же суп, просто повторно нагретый, если вы используете другой слой слушателей в качестве моста между фактическими обратными вызовами и вашим наблюдателем. Actual callback → bridge callback → Observer.

Преимущества:

более линейный код
один экземпляр фактического обратного вызова, вне наблюдателя
выглядит особенно хорошо с функциями высокого порядка, например функцией литералы в котлине:

Ex (обратите внимание, как мало создается наблюдаемое закрытие):

class LocationService @Inject constructor(private val googleApiClient: GoogleApiClient) : ConnectionCallbacks{

    val locationObservable: Observable<Location>    

    private var passToObservable: (Location?) -> Unit = {}

    init {
        locationObservable = Observable.create<Location> { subscription ->
            passToObservable = { location ->
                subscription.onNext(location)
            }
        }.doOnSubscribe {
            googleApiClient.registerConnectionCallbacks(this)
            googleApiClient.connect()
        }.doOnUnsubscribe {
            googleApiClient.unregisterConnectionCallbacks(this)
        }
    }

    override fun onConnected(connectionHint: Bundle?) {
        val location = LocationServices.FusedLocationApi.getLastLocation(googleApiClient)
        passToObservable(location)
    }

    override fun onConnectionSuspended(cause: Int) {
        //...
    }
}

Ответ 4

Если вам нужно что-то простое и встроенное, дайте этому подходу попробовать http://examples.javacodegeeks.com/core-java/beans/bean-property-change-event-listener/

java.beans.PropertyChangeEvent;
java.beans.PropertyChangeListener;
java.beans.PropertyChangeSupport;

С сайта есть фрагмент, который показывает, как его использовать

package com.javacodegeeks.snippets.core;

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.beans.PropertyChangeSupport;

public class BeanPropertyChangeEventListener {

    public static void main(String[] args) throws Exception {

        Bean bean = new Bean();
        bean.addPropertyChangeListener(new MyPropertyChangeListener());

        bean.setProperty1("newProperty1");
        bean.setProperty2(123);


        bean.setProperty1("newnewProperty1");
        bean.setProperty2(234);

    }

    public static class MyPropertyChangeListener implements PropertyChangeListener {
        // This method is called every time the property value is changed
        public void propertyChange(PropertyChangeEvent evt) {
            System.out.println("Name = " + evt.getPropertyName());

  System.out.println("Old Value = " + evt.getOldValue());

  System.out.println("New Value = " + evt.getNewValue());

  System.out.println("**********************************");
        }
    }

    public static class Bean {

        private PropertyChangeSupport pcs = new PropertyChangeSupport(this);

        // Property property1
        private String property1;
        // Property property2
        private int property2;

        public String getProperty1() {
            return property1;
        }
        public void setProperty1(String property1) {
            pcs.firePropertyChange("property1", this.property1, property1);
            this.property1 = property1;
        }

        public int getProperty2() {
            return property2;
        }
        public void setProperty2(int property2) {
            pcs.firePropertyChange("property2", this.property2, property2);
            this.property2 = property2;
        }

        public void addPropertyChangeListener(PropertyChangeListener listener) {
            pcs.addPropertyChangeListener(listener);
        }

    }

}

это довольно просто