Real World Use case of RxJava Subject

Question

Есть много отличных статей, объясняющих около RxJava. Но почти никто из них не объясняет концепцию реальным примером.

Так что я в основном понимаю концепцию RxJava Тема как трубы и одновременно observable и observer.

Но я не вижу, что будет в реальном мире использования этого RxJava Subject в Android Development мире. Не могли бы вы что-то придумать?

Answer 1

В моем случае это было потому, что у меня было одно Наблюдаемое, ожидавшее, что элемент другого Наблюдаемого испускает, который был асинхронным, поскольку это был Интервал.

Scheduler scheduler = RxHelper.scheduler(vertx.getOrCreateContext()); 


Observable.just(callAnotherObservable) 
      .subscribe(item -> System.out.println(item) 

public Observable<String> callAnotherObservable(Scheduler scheduler,){ 
      Subject subject = ReplaySubject.create(1); 
      Observable.interval(100,TimeUnit.MILLISECONDS) 
      .map(i->"item to be passed to the other observable") 
      .subscribe(subject); 
      return subject.observeOn(scheduler).first();//Here we wait for the first emission of the interval Observable. 

} 

Здесь, как вы можете видеть, что мы используем subject.first() ждать первой эмиссии интервала наблюдаемой, который работает в другом потоке.

Если вы хотите увидеть больше примеров «hotObservables» https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/connectable/HotObservable.java

Answer 2

Субъекты имеет много «реальный мир» приложения, особенно, когда вы постепенно превратить ваш кодовый от императива реактивной стиле. Он может служить мостом между этими двумя мирами, где вы можете влиять на поток с нереактивным кодом, выходящим за пределы потока.

Но как вы попросили пример. В последнее время я выполнял пользовательское поведение, когда пользователь пытается вернуться из действия. RxJava предоставил мне очень элегантное решение проблемы, с которой я столкнулся, поэтому мне нужно было создать поток событий, который соответствовал бы пользователю, желающему вернуться. Я сознательно избегал фразы «нажмите кнопку« Назад », потому что в кодовой базе есть несколько мест, где я смогу имитировать ситуацию возврата, и она всегда проходит через метод onBackPressed().

Чтобы превратить это в один поток, потребуется массивный рефакторинг, который сейчас не находится в бюджете. Но я не хотел отказываться от решения через RxJava, поскольку это могло сделать код намного более кратким. Использование BehaviorSubject дало ответ, поскольку мне просто нужно было испустить событие в методе onBackPressed().

Answer 3

Я создавал общий ReportDownloadManager для приложения для Android, где нам нужно было использовать Observable, чей Observer будет загружать и хранить файл локально. Событие успешной или неудачной загрузки необходимо было обработать менеджером, но также необходимо было открыть Наблюдение за действиями/услугами, которые используют этот DownloadManager. Я считаю, что это было хорошим вариантом для использования Субъекта, чтобы оба использовали первоначальный Observable, но также генерировали события для клиента Observable.

import android.app.DownloadManager; 
import android.content.Context; 
import android.webkit.MimeTypeMap; 

import java.io.File; 

import io.reactivex.Observable; 
import io.reactivex.functions.Consumer; 
import io.reactivex.subjects.PublishSubject; 

public class ReportDownloadManager { 

    private final DownloadManager platformDownloadManager; 

    public ReportDownloadManager(Context context) { 
     this.platformDownloadManager = (DownloadManager) context.getSystemService(Context.DOWNLOAD_SERVICE); 
    } 

    public Observable<Object> download(final File file, DownloadAction downloadAction) { 
     final PublishSubject<Object> subject = PublishSubject.create(); 
     downloadAction.execute(file.getName()) 
       .subscribe(new Consumer<Object>() { 
        @Override 
        public void accept(Object o) throws Exception { 
         platformDownloadManager.addCompletedDownload(file.getName(), "No description", false, 
           MimeTypeMap.getSingleton().getMimeTypeFromExtension("pdf"), file.getAbsolutePath(), 
           file.length(), true); 
         subject.onNext(new Object()); 
         subject.onComplete(); 
        } 
       }, new Consumer<Throwable>() { 
        @Override 
        public void accept(Throwable throwable) throws Exception { 
         subject.onError(throwable); 
        } 
       }); 
     return subject; 
    } 

    interface DownloadAction { 
     Observable<Object> execute(String fileAbsolutePath); 
    } 

}