一、框架思想
-
觀察者模式
- 觀察者自下而上注入被觀察者
-
被觀察者自上而下發(fā)射事件
觀察者模式
-
裝飾器模式
- 自上而下,被觀察者被一層層裝飾
-
自下而上,觀察者被一層層裝飾
裝飾者模式
-
策略模式
- 函數(shù)式編程
二、最簡單的訂閱模型
以下是由just方法和subscribe方法組成的一個(gè)最簡單的訂閱模型
Observable
.just("Hello RxJava")
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
}
@Override
public void onNext(String s) {
Log.d(TAG, s);
}
@Override
public void onError(Throwable e) {
}
@Override
public void onComplete() {
Log.d(TAG, "Complete");
}
});
對(duì)于just方法
public static <T> Observable<T> just(T item) {
// 參數(shù)合法判斷欲芹,不允許傳入空item
ObjectHelper.requireNonNull(item, "The item is null");
// hook裝飾
return RxJavaPlugins.onAssembly(
// 返回真正的ObservableJust對(duì)象
new ObservableJust<T>(item));
}
public static <T> Observable<T> onAssembly(@NonNull Observable<T> source) {
Function<? super Observable, ? extends Observable> f = onObservableAssembly;
if (f != null) {
// 如果hook方法存在,則調(diào)用hook方法
return apply(f, source);
}
return source;
}
public final class ObservableJust<T> extends Observable<T> implements ScalarCallable<T> {
private final T value;
public ObservableJust(final T value) {
// 保存原始的value
this.value = value;
}
// 實(shí)際的訂閱方法的實(shí)現(xiàn)
@Override
protected void subscribeActual(Observer<? super T> downStream) {
// 傳入下游和value
ScalarDisposable<T> sd = new ScalarDisposable<T>(downStream, value);
// 訂閱
downStream.onSubscribe(sd);
// Runnable.run()吟吝;
sd.run();
}
}
public static final class ScalarDisposable<T> extends AtomicInteger implements QueueDisposable<T>, Runnable {
// 省略很多跟隊(duì)列使用菱父,線程安全相關(guān)的方法
@Override
public void run() {
if (get() == START && compareAndSet(START, ON_NEXT)) {
downStream.onNext(value);
if (get() == ON_NEXT) {
lazySet(ON_COMPLETE);
downStream.onComplete();
}
}
}
}
對(duì)于subscribe方法
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete, Consumer<? super Disposable> onSubscribe) {
// 參數(shù)合法檢查
ObjectHelper.requireNonNull(onNext, "onNext is null");
ObjectHelper.requireNonNull(onError, "onError is null");
ObjectHelper.requireNonNull(onComplete, "onComplete is null");
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
// 把4個(gè)Lambda表達(dá)式合并創(chuàng)建LambdaObserver
LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
// 執(zhí)行訂閱
subscribe(ls);
// 返回
return ls;
}
@Override
public final void subscribe(Observer<? super T> downStream) {
// 檢查參數(shù)合法性
ObjectHelper.requireNonNull(downStream, "observer is null");
try {
// hook
downStream = RxJavaPlugins.onSubscribe(this, downStream);
// hook后再次檢查Observer不為null
ObjectHelper.requireNonNull(downStream, "The RxJavaPlugins.onSubscribe hook returned a null Observer. Please change the handler provided to RxJavaPlugins.setOnObservableSubscribe for invalid null returns. Further reading: https://github.com/ReactiveX/RxJava/wiki/Plugins");
// 執(zhí)行實(shí)際的subscribe,傳入下游Observer
subscribeActual(downStream);
} catch (NullPointerException e) { // NOPMD
throw e;
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// can't call onError because no way to know if a Disposable has been set or not
// can't call onSubscribe because the call might have set a Subscription already
RxJavaPlugins.onError(e);
NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
npe.initCause(e);
throw npe;
}
}
三剑逃、操作符的實(shí)現(xiàn)
最簡單的操作符 map
public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
// 檢查參數(shù)合法性
ObjectHelper.requireNonNull(mapper, "mapper is null");
// hook
return RxJavaPlugins.onAssembly(
// 返回真正的 ObservableMap浙宜,傳入了上游引用和mapper方法
new ObservableMap<T, R>(this, mapper));
}
ObservableMap等操作符一般繼承AbstractObservableWithUpstream,這種Observable持有了上游引用
public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;
public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}
@Override
public void subscribeActual(Observer<? super U> downStream) {
// 合并下游和mapper蛹磺,創(chuàng)建新的Observer:MapObserver
source.subscribe(new MapObserver<T, U>(downStream, function));
}
static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;
MapObserver(Observer<? super U> downStream, Function<? super T, ? extends U> mapper) {
super(downStream);
this.mapper = mapper;
}
// 重新實(shí)現(xiàn)onNext
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != NONE) {
downStream.onNext(null);
return;
}
U v;
try {
// 檢查mapper的輸出結(jié)果粟瞬,合法則賦值給
v = ObjectHelper.requireNonNull(
mapper.apply(t), "The mapper function returned a null value.");
} catch (Throwable ex) {
fail(ex);
return;
}
// 下游onNext傳入經(jīng)過mapper的結(jié)果
downStream.onNext(v);
}
}
對(duì)于操作符而言
- 自上而下
- 裝飾器模式。操作符創(chuàng)建了一個(gè)新的 Observable
- 該 Observable 重寫了subscribeActual方法
- subscribeActual 的實(shí)現(xiàn)是讓所持有的上游source調(diào)用subscribe方法去訂閱 被重新裝飾的下游
- 自下而上
- 裝飾器模式萤捆。操作符使下游Observer被裝飾后形成新的Observer
- 重寫了Observer的幾個(gè)方法裙品,向下傳遞經(jīng)過操作符處理后的數(shù)值結(jié)果
自定義操作符
// compose()操作符,自上而下俗或,封裝Observable
public interface ObservableTransformer<Upstream, Downstream> {
ObservableSource<Downstream> apply(Observable<Upstream> upstream);
}
// lift()操作符市怎,自下而上,封裝Observer
public interface ObservableOperator<Downstream, Upstream> {
Observer<? super Upstream> apply(@NonNull Observer<? super Downstream> observer) throws Exception;
}
- lift
- 創(chuàng)建并返回新的Observable辛慰,即ObservableLift
- 訂閱發(fā)生在operator.apply(s)区匠,即得到新的Observable之后
- 意圖是封裝一個(gè)操作符,類似于create
- compose
- 意圖是封裝一系列操作符昆雀,方便復(fù)用
四辱志、線程調(diào)度
舉例 .subscribeOn(AndroidSchedulers.mainThread())
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(
// 返回Observable蝠筑,傳入上游this和調(diào)度器scheduler
new ObservableSubscribeOn<T>(this, scheduler));
}
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
@Override
public void subscribeActual(final Observer<? super T> downStream) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(downStream);
// SubscribeOnObserver實(shí)現(xiàn)了Disposable狞膘,將其傳遞給下游的onSubscribe
downStream.onSubscribe(parent);
// 把source的訂閱放在Runnable中,由scheduler調(diào)度
parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
@Override
public void run() {
source.subscribe(parent);
}
}));
}
}
講解
- subscribeOn 線程調(diào)度即發(fā)生在subscribe時(shí)
- subscribeOn 只能生效一次
- 因?yàn)橥暾挠嗛嗊^程是自下而上訂閱什乙,數(shù)據(jù)源發(fā)射事件在自上而下傳遞挽封,所以真正發(fā)射事件所在的線程,是有最接近上游的一次subscribeOn來決定的臣镣,其他的都會(huì)被覆蓋
- 從代碼上看辅愿,就是經(jīng)歷了n個(gè)線程的傳遞后,把source.subscribe(parent)放在了最上游的那個(gè)線程中去發(fā)射忆某,如果沒有observeOn影響点待,整個(gè)事件流都會(huì)在那個(gè)線程完成
- 為什么flatMap能改變subscribeOn的這種特性?因?yàn)閒latMap等操作符創(chuàng)建了新的Observable弃舒,而不是單純傳遞上下游
對(duì)于調(diào)度器 AndroidSchedulers.mainThread()癞埠。其實(shí)際上是 new HandlerScheduler(new Handler(Looper.getMainLooper()))
從上面看出状原,實(shí)現(xiàn)線程調(diào)度的是scheduler.scheduleDirect方法
@Override
public Disposable scheduleDirect(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
run = RxJavaPlugins.onSchedule(run);
// ScheduledRunnable 實(shí)現(xiàn)了Disposable接口
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
// 大家熟悉的handler.postDelayed
handler.postDelayed(scheduled, Math.max(0L, unit.toMillis(delay)));
return scheduled;
}
// 一般的Schedule都是用createWorker().schedule(scheduled, delay, unit)的方式去實(shí)現(xiàn)線程調(diào)度
// Worker實(shí)現(xiàn)相應(yīng)的disposable接口,便于取消訂閱時(shí)停止執(zhí)行尚未執(zhí)行的Runnable苗踪,并裝飾相應(yīng)的hook颠区。
// 這個(gè)類的實(shí)現(xiàn)比較有點(diǎn)劃水,可能因?yàn)槭侵骶€程通铲,所以不擔(dān)心泄露
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
private static final class HandlerWorker extends Worker {
private final Handler handler;
private volatile boolean disposed;
HandlerWorker(Handler handler) {
this.handler = handler;
}
@Override
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException("run == null");
if (unit == null) throw new NullPointerException("unit == null");
if (disposed) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
@Override
public void dispose() {
disposed = true;
handler.removeCallbacksAndMessages(this /* token */);
}
@Override
public boolean isDisposed() {
return disposed;
}
}
舉例 .observeOn(AndroidSchedulers.mainThread())
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
return RxJavaPlugins.onAssembly(
new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}
// 以上代碼都很熟悉了毕莱,暫不贅述
public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler; // 線程調(diào)度器
this.delayError = delayError; // 出現(xiàn)錯(cuò)誤是否立刻中斷線程
this.bufferSize = bufferSize; //緩沖區(qū)大小,默認(rèn)128
}
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
// 默認(rèn)線程則不做線程調(diào)度颅夺,直接在當(dāng)前線程中調(diào)用
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(
// 用Worker朋截、和相關(guān)參數(shù)裝飾observer,得到新的Observer注入上游
new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
//省略部分代碼
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
//省略部分代碼吧黄,創(chuàng)建緩沖隊(duì)列
queue = new SpscLinkedArrayQueue<T>(bufferSize);
actual.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t); //上游的數(shù)據(jù)全部先入隊(duì)列
}
//執(zhí)行調(diào)度
schedule();
}
void schedule() {
if (getAndIncrement() == 0) {
// 隊(duì)列如果已經(jīng)空了质和,則再次調(diào)度
worker.schedule(this);
}
}
@Override
public void run() {
// Fused 熔斷機(jī)制,默認(rèn)false
if (outputFused) {
drainFused();
} else {
drainNormal();
}
}
//該函數(shù)在Runnable所在的線程執(zhí)行稚字,從緩沖隊(duì)列里拿出事件饲宿,向下游發(fā)射
void drainNormal() {
int missed = 1;
final SimpleQueue<T> q = queue;
final Observer<? super T> a = actual;
for (;;) {
// 如果設(shè)置了errorDelay,則不管隊(duì)列是否為空胆描,發(fā)生了錯(cuò)誤都會(huì)中斷發(fā)射瘫想,并調(diào)用observer的onError
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}
for (;;) {
boolean d = done;
T v;
try {
v = q.poll(); //隊(duì)列中取數(shù)據(jù)
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
s.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;
if (checkTerminated(d, empty, a)) {
return;
}
if (empty) {
break;
}
// 向下游發(fā)射數(shù)據(jù)
a.onNext(v);
}
missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}
}
}
講解
- 訂閱發(fā)生在調(diào)度前,說明線程調(diào)度不影響訂閱過程
- ObserveOnObserver持有下游observer和調(diào)度器,并實(shí)現(xiàn)Runnable接口
- 訂閱時(shí)(onSubscribe)會(huì)創(chuàng)建一個(gè)緩沖隊(duì)列挚冤,當(dāng)上游數(shù)據(jù)到來先放在隊(duì)列威始,接著在調(diào)度線程中取出并發(fā)射到下游
- 由上看observeOn可以多次生效
五、背壓
- 默認(rèn)策略判斷是否觸發(fā)背壓的因素:
- 同步場景中车吹,有發(fā)射數(shù)是否超出響應(yīng)式拉取值 request 決定
- 異步場景中,由是否超出緩沖池 queue 的承受能力決定醋闭。需要下游的request方法拉取queue的數(shù)據(jù)
- observeOn允許我們設(shè)置緩沖隊(duì)列的容量大小
- 在onSubscribe(Subscription s)回調(diào)提供的s可以調(diào)用request方法來增加拉取數(shù)窄驹;如果不重寫,默認(rèn)執(zhí)行s.request(Long.MAX_VALUE)
- 背壓策略
- ERROR 觸發(fā)背壓直接拋異常 MissingBackpressureException
- BUFFER: queue無限大证逻,知道OOM
- DROP: 超載則拋棄之后的數(shù)據(jù)乐埠,不拋異常
- LATEST:超載后拋棄之后數(shù)據(jù),且是專用有一個(gè)額外空間保留當(dāng)前最新一次數(shù)據(jù)
舉例 onBackpressureDrop 方法
public final class FlowableOnBackpressureDrop<T> extends AbstractFlowableWithUpstream<T, T> implements Consumer<T> {
final Consumer<? super T> onDrop;
@Override
protected void subscribeActual(Subscriber<? super T> s) {
this.source.subscribe(new BackpressureDropSubscriber<T>(s, onDrop));
}
static final class BackpressureDropSubscriber<T> extends AtomicLong implements FlowableSubscriber<T>, Subscription {
@Override
public void onNext(T t) {
if (done) {
return;
}
long r = get();
if (r != 0L) {
actual.onNext(t);
BackpressureHelper.produced(this, 1);
} else {
try {
onDrop.accept(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
cancel();
onError(e);
}
}
}
@Override
public void request(long n) {
if (SubscriptionHelper.validate(n)) {
BackpressureHelper.add(this, n);
}
}
}
}
BackpressureDropSubscriber繼承了AtomicLong囚企,實(shí)現(xiàn)了Subscriber丈咐。只有判斷自身不為0時(shí)才會(huì)向下游發(fā)射元素,否則將被拋棄龙宏。這個(gè)數(shù)值的計(jì)算在BackpressureHelper中計(jì)算
// 每執(zhí)行一次onNext棵逊,當(dāng)前值減一
public static long produced(AtomicLong requested, long n) {
for (;;) {
long current = requested.get();
if (current == Long.MAX_VALUE) {
return Long.MAX_VALUE;
}
long update = current - n;
if (update < 0L) {
RxJavaPlugins.onError(new IllegalStateException("More produced than requested: " + update));
update = 0L;
}
if (requested.compareAndSet(current, update)) {
return update;
}
}
}
// request調(diào)用時(shí),值會(huì)加到當(dāng)前值上银酗。
public static long add(AtomicLong requested, long n) {
for (;;) {
long r = requested.get();
if (r == Long.MAX_VALUE) {
return Long.MAX_VALUE;
}
long u = addCap(r, n);
if (requested.compareAndSet(r, u)) {
return r;
}
}
}
這樣辆影,就實(shí)現(xiàn)了發(fā)生背壓則拋棄新的值
onBackpressureLatest的實(shí)現(xiàn)和前者類似掩浙,只是多了一個(gè)對(duì)象來存儲(chǔ)最新一次的值:
final AtomicReference<T> current = new AtomicReference<T>();
@Override
public void onNext(T t) {
current.lazySet(t);
drain();
}
六、Subject
Subject 繼承Observable秸歧,實(shí)現(xiàn)Observer厨姚。具有自攻自受的特性,相當(dāng)于是一個(gè)中間層
一般Observable觀察的對(duì)象往往是靜態(tài)的键菱,如一個(gè)常量谬墙、一個(gè)文件。有確定的開頭和結(jié)尾
Subject是動(dòng)態(tài)的经备,監(jiān)聽沒有被中斷的時(shí)候拭抬,可以有不確定的事件傳來
- Observable特性
- 訂閱:如果下游可用,則加入到訂閱隊(duì)列中
@Override
public void subscribeActual(Observer<? super T> t) {
PublishDisposable<T> ps = new PublishDisposable<T>(t, this);
t.onSubscribe(ps);
if (add(ps)) {
// if cancellation happened while a successful add, the remove() didn't work
// so we need to do it again
if (ps.isDisposed()) {
remove(ps);
}
} else {
Throwable ex = error;
if (ex != null) {
t.onError(ex);
} else {
t.onComplete();
}
}
}
boolean add(PublishDisposable<T> ps) {
for (;;) {
PublishDisposable<T>[] a = subscribers.get();
if (a == TERMINATED) {
return false;
}
int n = a.length;
@SuppressWarnings("unchecked")
PublishDisposable<T>[] b = new PublishDisposable[n + 1];
System.arraycopy(a, 0, b, 0, n);
b[n] = ps;
if (subscribers.compareAndSet(a, b)) {
return true;
}
}
}
- Observer 特性
- 如果本身可用侵蒙,切存在可用的觀察者造虎,則向下傳遞事件
// 但舉例onNext方法
@Override
public void onNext(T t) {
if (subscribers.get() == TERMINATED) {
return;
}
if (t == null) {
onError(new NullPointerException("onNext called with null. Null values are generally not allowed in 2.x operators and sources."));
return;
}
// 遍歷下游觀察者隊(duì)列,并逐個(gè)調(diào)用onNext
for (PublishDisposable<T> s : subscribers.get()) {
s.onNext(t);
}
}
七纷闺、性能問題
.map(x -> x + 1) 和 .flatMap(x -> Observable.just(x + 1))有什么區(qū)別
- map和flatMap的區(qū)別
- map沒有創(chuàng)建新的Observable算凿,flatMap創(chuàng)建了新的Observable,相當(dāng)于創(chuàng)建了新的流
- map在上游和下游之間仍是線性的犁功。flatMap已經(jīng)上升了一個(gè)階氓轰,即每一個(gè)元素進(jìn)來,都會(huì)變成一個(gè)新的source浸卦,而下游也會(huì)變成一個(gè)新的observer
循環(huán)的靜態(tài)Observable和動(dòng)態(tài)的Subject之間的區(qū)別
void callback(int item) {
Observable
.<Integer>create(emitter -> {
emitter.onNext(item);
})
.map(i -> i + 1)
.subscribe(i -> Log.d(TAG, "i = " + i));
}
Subject<Integer> mSubject = PublishSubject.create();
void init() {
mSuject
.map(i -> i + 1)
.subscribe(i -> Log.d(TAG, "i = " + i));
}
void callback(int item) {
mSuject.onNext(item);
}
- 循環(huán)創(chuàng)建的事件流署鸡,所有操作符都涉及了new對(duì)象,如果callback被頻繁調(diào)用限嫌,則會(huì)產(chǎn)生十分多的臨時(shí)對(duì)象造成內(nèi)存抖動(dòng)
八靴庆、Rx擴(kuò)展
RxBinding
final class ViewClickObservable extends Observable<Object> {
private final View view;
ViewClickObservable(View view) {
this.view = view;
}
@Override protected void subscribeActual(Observer<? super Object> observer) {
if (!checkMainThread(observer)) {
return;
}
//訂閱階段執(zhí)行的邏輯,創(chuàng)建listener綁定view
Listener listener = new Listener(view, observer);
observer.onSubscribe(listener);
view.setOnClickListener(listener);
}
static final class Listener extends MainThreadDisposable implements OnClickListener {
private final View view;
private final Observer<? super Object> observer;
Listener(View view, Observer<? super Object> observer) {
this.view = view;
this.observer = observer;
}
@Override public void onClick(View v) {
if (!isDisposed()) {
//發(fā)生點(diǎn)擊時(shí)傳遞事件給下游
observer.onNext(Notification.INSTANCE);
}
}
@Override protected void onDispose() {
view.setOnClickListener(null);
}
}
}
RxPermissions
public <T> ObservableTransformer<T, Boolean> ensure(final String... permissions) {
return new ObservableTransformer<T, Boolean>() {
@Override
public ObservableSource<Boolean> apply(Observable<T> o) {
return request(o, permissions)
// 轉(zhuǎn)換 Observable<Permission> 為 Observable<Boolean>
.buffer(permissions.length)
.flatMap(new Function<List<Permission>, ObservableSource<Boolean>>() {
@Override
public ObservableSource<Boolean> apply(List<Permission> permissions) {
if (permissions.isEmpty()) {
// Occurs during orientation change, when the subject receives onComplete.
// In that case we don't want to propagate that empty list to the
// subscriber, only the onComplete.
return Observable.empty();
}
// Return true if all permissions are granted.
for (Permission p : permissions) {
if (!p.granted) {
return Observable.just(false);
}
}
return Observable.just(true);
}
});
}
};
}