Real-World Case of RxJava Subject

Question

There are many great articles in the RxJava article. But almost none of them explains the concept with a real example.

So, I basically understand the concept of the RxJava theme as pipe , and it is both observable and observer .

But I don’t see what will use this RxJava Subject in the real world of Android development in the real world. Could you come up with something?

Answer 1

In my case, it was because I had one Observable, which was waiting for the appearance of another observable Investigator, which was asynchronous, since it was an interval.

 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. } 

Here, as you can see, we use subject.first () to wait for the first emission of the observed interval, which is executed in another thread.

If you want to see more examples of "hotObservables" https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/connectable/HotObservable.java

Answer 2

Subjects have many “real world” applications, especially when you gradually transform your code base from an imperative to a reactive style. It can serve as a bridge between these two worlds, where you can influence a stream with non-reactive code that goes beyond the stream.

But as you asked for an example. Recently, I have been performing user behavior when a user tries to return from an action. RxJava provided me with a very elegant solution to the problem I encountered, so I needed to create an event stream that would match the user who wants to return. I deliberately avoided the phrase “press the back button, because there are several places in the code base where I can simulate a return situation, and it always goes through the onBackPressed() method.

Including this in a single thread will require massive refactoring, which is currently not in the budget. But I did not want to give up the solution through RxJava, as this could make the code much more concise. Using BehaviorSubject gave the answer, because I just needed to emit an event in the onBackPressed() method.

Answer 3

I created a generic ReportDownloadManager for an Android application where we needed to use an Observable, whose Observer would load and store the file locally. The event of a successful or unsuccessful download had to be handled by the manager, but it was also necessary to open the Monitoring of actions / services that use this DownloadManager. I believe this was a good use case for using the subject, so that both use the original Observable, but also generate events for the Observable client.

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