RxJava basic usage case sharing source code analysis [with code]-javaTutorial-php.cn

This article details the basic usage of RxJava. RxJava is a magical framework with very simple usage, but the internal implementation is a bit complicated and the code logic is a bit convoluted. Online articles about RxJava source code analysis, source code posts are few and incomplete. The complete source code analysis is listed below for reference.

1. Basic usage of RxJava

Observable.create(new Observable.OnSubscribe () { @Override public void call(Subscriber subscriber) { } }).subscribe(new Observer () { @Override public void onCompleted() {

}

@Override public void onError(Throwable e) {

}

@Override public void onNext(Object o) {

} }); Copy after login 2. First look at the .subscribe(new Observer ()) method

public final Subscription subscribe(final Observer observer) { if (observer instanceof Subscriber) { return subscribe((Subscriber )observer); } if (observer == null) { throw new NullPointerException("observer is null"); } return subscribe(new ObserverSubscriber (observer)); } Copy after login This is just the passed in observer parameter A simple encapsulation (ObserverableSubscriber)

Continue to look at the subscribe method

public final Subscription subscribe(Subscriber subscriber) { return Observable.subscribe(subscriber, this); } Copy after login static Subscription subscribe(Subscriber subscriber, Observable observable) { // validate and proceed if (subscriber == null) { throw new IllegalArgumentException("subscriber can not be null"); } if (observable.onSubscribe == null) { throw new IllegalStateException("onSubscribe function can not be null."); /* * the subscribe function can also be overridden but generally that's not the appropriate approach * so I won't mention that in the exception */ }

// new Subscriber so onStart it subscriber.onStart();

/* * See https://github.com/ReactiveX/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls * to user code from within an Observer" */ // if not already wrapped if (!(subscriber instanceof SafeSubscriber)) { // assign to `observer` so we return the protected version subscriber = new SafeSubscriber (subscriber); }

// The code below is exactly the same an unsafeSubscribe but not used because it would // add a significant depth to already huge call stacks. try { // allow the hook to intercept and/or decorate RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber); return RxJavaHooks.onObservableReturn(subscriber); } catch (Throwable e) { // special handling for certain Throwable/Error/Exception types Exceptions.throwIfFatal(e); // in case the subscriber can't listen to exceptions anymore if (subscriber.isUnsubscribed()) { RxJavaHooks.onError(RxJavaHooks.onObservableError(e)); } else { // if an unhandled error occurs executing the onSubscribe we will propagate it try { subscriber.onError(RxJavaHooks.onObservableError(e)); } catch (Throwable e2) { Exceptions.throwIfFatal(e2); // if this happens it means the onError itself failed (perhaps an invalid function implementation) // so we are unable to propagate the error correctly and will just throw RuntimeException r = new OnErrorFailedException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2); // TODO could the hook be the cause of the error in the on error handling. RxJavaHooks.onObservableError(r); // TODO why aren't we throwing the hook's return value. throw r; // NOPMD } } return Subscriptions.unsubscribed(); } } Copy after login You only need to pay attention here

RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber); Copy after login public static Observable.OnSubscribe onObservableStart(Observable instance, Observable.OnSubscribe onSubscribe) { Func2 f = onObservableStart; if (f != null) { return f.call(instance, onSubscribe); } return onSubscribe; } Copy after login RxJavaHooks.onObservableStart(observable, observable.onSubscribe) 这个方法返回的是它的第二个参数，也就是Observable它自己的onSubscribe 对象， 所以在subscribe 方法里面调用了 onSubscribe.call(subscriber)方法 Copy after login The subscriber here is the parameter passed in

protected Observable(OnSubscribe f) { this.onSubscribe = f; } Copy after login public static Observable create(OnSubscribe f) { return new Observable (RxJavaHooks.onCreate(f)); } Copy after login It can be seen that the onSubscribe object is the parameter passed in by create, then the whole process is very clear

The entire process will only be executed when the subscribe method is called: subscribe ===>Call onSubscribe.call(observer ) method also passes the observer in

Related recommendations:

RxJava Operator (8) Aggregate_PHP Tutorial

Video: JavaScript Basic strengthening video tutorial

RxJava basic usage case sharing source code analysis [with code]

Hot AI Tools

Undresser.AI Undress

AI Clothes Remover

Undress AI Tool

Clothoff.io

Video Face Swap

Hot Article

Hot Tools

Notepad++7.3.1

SublimeText3 Chinese version

Zend Studio 13.0.1

Dreamweaver CS6

SublimeText3 Mac version

Hot Topics