Как разместить событие в ReactiveX Observable?

Question

У меня есть метод обратного вызова события

public void onSomethingHappened(Information info) { 

, которая называется некоторыми обычным эмитентом событий.

Теперь я хочу обернуть это в наблюдаемый ReactiveX.

Как это сделать?

UPDATE

Я нашел PublishSubject<T> класс и использовал это следующим образом обернуть LocationListener из Android API:

public class LocationPublisher implements LocationListener { 


    @Override 
    public void onLocationChanged(Location location) { 
     LocationPublisher.onNext(location); 
    } 

    @Override 
    public void onStatusChanged(String provider, int status, Bundle extras) { 
     StatusPublisher.onNext(new BundleEx(provider, status, extras)); 
    } 

    @Override 
    public void onProviderEnabled(String provider) { 
     EnablenessPublisher.onNext(new Enableness(provider, true)); 
    } 

    @Override 
    public void onProviderDisabled(String provider) { 
     EnablenessPublisher.onNext(new Enableness(provider, false)); 
    } 

    public final PublishSubject<Location> LocationPublisher = PublishSubject.create(); 

    public final PublishSubject<BundleEx> StatusPublisher = PublishSubject.create(); 

    public final PublishSubject<Enableness> EnablenessPublisher = PublishSubject.create(); 


    public static class Enableness { 

     private final String provider; 

     private final boolean enabled; 


     public Enableness(String provider, boolean enabled) { 
      this.provider = provider; 
      this.enabled = enabled; 
     } 

     public String getProvider() { 
      return provider; 
     } 

     public boolean isEnabled() { 
      return enabled; 
     } 
    } 


    public static class BundleEx { 

     private final String provider; 

     private final int status; 

     private final Bundle extras; 

     public BundleEx(String provider, int status, Bundle extras) { 
      this.provider = provider; 
      this.status = status; 
      this.extras = extras; 
     } 

     public String getProvider() { 
      return provider; 
     } 

     public int getStatus() { 
      return status; 
     } 

     public Bundle getExtras() { 
      return extras; 
     } 
    } 
} 

Является ли это правильный подход? Теперь я могу подписаться на LocationPublisher, StatusPublisher и EnablenessPublisher Я?

Answer 1

Вы считаете источник этого события концептуальным наблюдаемым, а слушатель - концептуальным подписчиком.

Самое большое несоответствие в том, что у вас есть бесконечный источник событий, то есть он не будет заканчиваться на своем собственном. Не факт, что это незаконно или невозможно иметь дело, но это не естественно для RX.

Таким образом, вы используете Observable.create(). OnSubscribe - это действие, которое добавляет слушателя к этому традиционному источнику, в то время как вы реализуете прослушиватель, который вызывает подписчик onNext().

// Окончательный источник EventSource = ...

Observable.create(s -> { 
    source.addFooBarListener(info -> { //could be a full anon innerclass too if more than 1 method 
    s.onNext(info); 
    }); 
}); 

Я иду по памяти, у меня нет компилятора, ни APIs на руках. Прости. Отсутствуют строки кода для правильного выполнения onComplete()/onError(), конечно.