Как создать свою собственную горячую Наблюдаемую с нуля?

Question

Как создать свой собственный горячий Наблюдаемый с нуля?

Я хотел бы создать свою собственную функцию, возвращая наблюдаемые, возвращающиеся места:

public static Observable<Location> locationObservable(Context context, String provider, long minTime, float minDistance) { 

Это для Android. Для этих целей рекомендуется использовать Observable.create(), но example shows just passing constant list of integers каждому абоненту, который не является горячим.

Если я что-то еще здесь, например, помню список подписчиков, то как я могу реализовать отмену подписки и многие другие функции?

I.e. абсолютно не идея, что делать внутри Observable.OnSubscribe<Integer>() реализации?

Answer 1

Обычно для создания горячей наблюдаемый вы используете какой-то Subject: PublishSubject, BehaviorSubject и т.д.

Смотрите примеры для BehaviorSubjecthere.

class LocationService { 
    private Subject<Location> subject = BehaviorSubject.create(); 

    Observable<Location> locationObservable(...) { 
     return subject; 
    } 

    void onNewLocationListener(Location newLocation) { 
     subject.onNext(newLocation); 
    } 
} 

Answer 2

Посмотрите на этот прекрасный пример взят непосредственно из Realm-х RealmObservableFactory:

@Override 
public Observable<Realm> from(Realm realm) { 
    final RealmConfiguration realmConfig = realm.getConfiguration(); 
    return Observable.create(new Observable.OnSubscribe<Realm>() { // create new observable 
     @Override 
     public void call(final Subscriber<? super Realm> subscriber) { // this is executed on `subscribeOn(Scheduler)` 
      final Realm observableRealm = Realm.getInstance(realmConfig); 
      final RealmChangeListener<Realm> listener = new RealmChangeListener<Realm>() { 
       @Override 
       public void onChange(Realm realm) { 
        if (!subscriber.isUnsubscribed()) { // always check if subscriber is unsubscribed! 
         subscriber.onNext(observableRealm); 
        } 
       } 
      }; 
      subscriber.add(Subscriptions.create(new Action0() { // add unsubscription first! thread specified by unsubscribeOn(Scheduler) 
       @Override 
       public void call() { 
        observableRealm.removeChangeListener(listener); // remove listener 
        observableRealm.close(); 
       } 
      })); 
      observableRealm.addChangeListener(listener); // add listener 
      subscriber.onNext(observableRealm); // initial value 
     } 
    }); 
} 

И читать комментарии, это очень хороший пример.

Answer 3

Не рекомендуется писать свои собственные, по крайней мере, до тех пор, пока вы не овладеете существующими и не потребуете своеобразного шаблона кэширования/излучения, не охватываемого стандартными 5 (Async, Behavior, Publish, Replay, Unicast).

У меня есть 3 части серии по этому вопросу (каламбур), если вы действительно хотите:

• Part 1
• Part 2
• Part 3