网站个人空间怎么做今日国内新闻10则

Observable.just("Hello world").subscribe(word -> {System.out.println("got " + word + " @ " + Thread.currentThread().getName());});

// Observable.java
public static <T> Observable<T> just(final T value) {return ScalarSynchronousObservable.create(value);
}// ScalarSynchronousObservable.java
public static <T> ScalarSynchronousObservable<T> create(T t) {return new ScalarSynchronousObservable<T>(t);           // 1
}protected ScalarSynchronousObservable(final T t) {super(RxJavaHooks.onCreate(new JustOnSubscribe<T>(t))); // 2this.t = t;
}

public final Subscription subscribe(final Action1<? super T> onNext) {// 省略参数检查代码Action1<Throwable> onError = InternalObservableUtils.ERROR_NOT_IMPLEMENTED;Action0 onCompleted = Actions.empty();return subscribe(new ActionSubscriber<T>(onNext, onError, onCompleted));                             // 1
}public final Subscription subscribe(Subscriber<? super T> subscriber) {return Observable.subscribe(subscriber, this);
}static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {// 省略参数检查代码subscriber.onStart();                                   // 2if (!(subscriber instanceof SafeSubscriber)) {subscriber = new SafeSubscriber<T>(subscriber);     // 3}try {RxJavaHooks.onObservableStart(observable, observable.onSubscribe).call(subscriber);       // 4return RxJavaHooks.onObservableReturn(subscriber);  // 5} catch (Throwable e) {// 省略错误处理代码}
}

// ScalarSynchronousObservable.java
static final class JustOnSubscribe<T> implements OnSubscribe<T> {// ...@Overridepublic void call(Subscriber<? super T> s) {s.setProducer(createProducer(s, value));}
}static <T> Producer createProducer(Subscriber<? super T> s, T v) {// ...return new WeakSingleProducer<T>(s, v);
}

// Subscriber.java
public void setProducer(Producer p) {long toRequest;boolean passToSubscriber = false;synchronized (this) {toRequest = requested;producer = p;if (subscriber != null) {             // 1if (toRequest == NOT_SET) {passToSubscriber = true;}}}if (passToSubscriber) {                   // 2subscriber.setProducer(producer);} else {if (toRequest == NOT_SET) {           // 3producer.request(Long.MAX_VALUE);} else {producer.request(toRequest);}}
}

// ScalarSynchronousObservable.java
static final class WeakSingleProducer<T> implements Producer {// ...@Overridepublic void request(long n) {// 省略状态检查代码Subscriber<? super T> a = actual;if (a.isUnsubscribed()) {return;}T v = value;try {a.onNext(v);} catch (Throwable e) {Exceptions.throwOrReport(e, a, v);return;}if (a.isUnsubscribed()) {return;}a.onCompleted();}
}

Observable.just("Hello world").map(String::length).subscribe(word -> {System.out.println("got " + word + " @ "+ Thread.currentThread().getName());});

// Observable.java
public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {return create(new OnSubscribeMap<T, R>(this, func));
}public static <T> Observable<T> create(OnSubscribe<T> f) {return new Observable<T>(RxJavaHooks.onCreate(f));
}

public final class OnSubscribeMap<T, R> implements OnSubscribe<R> {final Observable<T> source;final Func1<? super T, ? extends R> transformer;public OnSubscribeMap(Observable<T> source, Func1<? super T, ? extends R> transformer) {this.source = source;this.transformer = transformer;}@Overridepublic void call(final Subscriber<? super R> o) {MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);   // 1o.add(parent);                                 // 2source.unsafeSubscribe(parent);                // 3}
}

static final class MapSubscriber<T, R> extends Subscriber<T> {final Subscriber<? super R> actual;final Func1<? super T, ? extends R> mapper;boolean done;public MapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {this.actual = actual;this.mapper = mapper;}@Overridepublic void onNext(T t) {R result;try {result = mapper.call(t);        // 1} catch (Throwable ex) {Exceptions.throwIfFatal(ex);unsubscribe();onError(OnErrorThrowable.addValueAsLastCause(ex, t));return;}actual.onNext(result);              // 2}// 省略 onError，onCompleted 和 setProducer
}

Observable.just("Hello world").map(String::length).subscribeOn(Schedulers.computation()).observeOn(AndroidSchedulers.mainThread()).subscribe(len -> {System.out.println("got " + len + " @ " + Thread.currentThread().getName());});

public final Observable<T> subscribeOn(Scheduler scheduler) {if (this instanceof ScalarSynchronousObservable) {return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);}return create(new OperatorSubscribeOn<T>(this, scheduler));
}

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> {final Scheduler scheduler;final Observable<T> source;public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {this.scheduler = scheduler;this.source = source;}@Overridepublic void call(final Subscriber<? super T> subscriber) {final Worker inner = scheduler.createWorker();subscriber.add(inner);                            // 1inner.schedule(new Action0() {@Overridepublic void call() {final Thread t = Thread.currentThread();Subscriber<T> s = new Subscriber<T>(subscriber) {@Overridepublic void onNext(T t) {subscriber.onNext(t);             // 2}// 省略 onError 和 onCompleted@Overridepublic void setProducer(final Producer p) {subscriber.setProducer(new Producer() {@Overridepublic void request(final long n) {if (t == Thread.currentThread()) {p.request(n);         // 3} else {inner.schedule(new Action0() {@Overridepublic void call() {p.request(n); // 4}});}}});}};source.unsafeSubscribe(s);                // 5}});}
}

public final Observable<T> observeOn(Scheduler scheduler) {return observeOn(scheduler, RxRingBuffer.SIZE);
}public final Observable<T> observeOn(Scheduler scheduler, int bufferSize) {return observeOn(scheduler, false, bufferSize);
}public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {if (this instanceof ScalarSynchronousObservable) {return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);}return lift(new OperatorObserveOn<T>(scheduler, delayError, bufferSize));
}public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {return create(new OnSubscribeLift<T, R>(onSubscribe, operator));
}

public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {final OnSubscribe<T> parent;final Operator<? extends R, ? super T> operator;public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {this.parent = parent;this.operator = operator;}@Overridepublic void call(Subscriber<? super R> o) {Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);st.onStart();parent.call(st);// 省略了异常处理代码}
}

public final class OperatorObserveOn<T> implements Operator<T, T> {// ...@Overridepublic Subscriber<? super T> call(Subscriber<? super T> child) {if (scheduler instanceof ImmediateScheduler) {// avoid overhead, execute directlyreturn child;} else if (scheduler instanceof TrampolineScheduler) {// avoid overhead, execute directlyreturn child;} else {ObserveOnSubscriber<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError, bufferSize);parent.init();return parent;}}// ...
}

Observable.create(subscriber -> {Worker worker = scheduler.createWorker();subscriber.add(worker);source.unsafeSubscribe(new Subscriber<T>(subscriber) {@Overridepublic void onNext(T t) {worker.schedule(() -> subscriber.onNext(t));}@Overridepublic void onError(Throwable e) {worker.schedule(() -> subscriber.onError(e));}@Overridepublic void onCompleted() {worker.schedule(() -> subscriber.onCompleted());}            });
});