gistfile1.java

@Override
    public void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        mSubscriptionChatNotSynced = Observable
                .combineLatest(
                        mMessagesListModel.isVisibleObservable(),
                        mMessagesListModel.isChatSyncedObervable(),
                        new Func2<Boolean, Boolean, Boolean>() {
                            @Override
                            public Boolean call(Boolean isFragmentVisible, Boolean isChatSynced) {
                                return isFragmentVisible && !isChatSynced;
                            }
                        })
                .filter(new Func1<Boolean, Boolean>() {
                    @Override
                    public Boolean call(Boolean isShouldLoad) {
                        return isShouldLoad;
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        new Action1<Boolean>() {
                            @Override
                            public void call(Boolean shouldSync) {
                                mSubscriptionChatLoad = chatLoadObservable()
                                        .subscribeOn(Schedulers.newThread())
                                        .observeOn(AndroidSchedulers.mainThread())
                                        .subscribe(
                                                new Action1<Chat>() {
                                                    @Override
                                                    public void call(Chat chat) {
                                                        mMessagesListModel.setChat(chat);
                                                    }
                                                },
                                                new Action1<Throwable>() {
                                                    @Override
                                                    public void call(Throwable throwable) {
                                                        Toast.makeText(getApplicationContext(), throwable.getMessage(), Toast.LENGTH_SHORT).show();
                                                    }
                                                });
                            }
                        });

        mSubscriptionShow = Observable
                .combineLatest(
                        mMessagesListModel.isVisibleObservable(),
                        mMessagesListModel.chatObservable(),
                        new Func2<Boolean, Chat, Pair<Boolean, Chat>>() {
                            @Override
                            public Pair<Boolean, Chat> call(Boolean isFragmentVisible, Chat chat) {
                                return new Pair<Boolean, Chat>(isFragmentVisible, chat);
                            }
                        })
                .filter(new Func1<Pair<Boolean, Chat>, Boolean>() {
                    @Override
                    public Boolean call(Pair<Boolean, Chat> booleanChatPair) {
                        return booleanChatPair.first;
                    }})
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Action1<Pair<Boolean, Chat>>() {
                    @Override
                    public void call(Pair<Boolean, Chat> booleanChatPair) {
                        Toast.makeText(getApplicationContext(), booleanChatPair.second.getCreatorId(), Toast.LENGTH_SHORT).show();
                    }});
    }

    private Observable<Chat> chatLoadObservable() {
        return Observable.create(new Observable.OnSubscribeFunc<Chat>() {
            @Override
            public Subscription onSubscribe(final Observer<? super Chat> observer) {
                final Future<Chat> future = mDatabaseThreadPool.submit(new Callable<Chat>() {
                    @Override
                    public Chat call() throws Exception {
                        try {
                            observer.onNext(mChatsStorage.fetchChatByLocalId(mMessagesListModel.getChatLocalId()));
                        } catch (Throwable throwable) {
                            observer.onError(throwable);
                        }
                        observer.onCompleted();
                        return null;
                    }
                });
                return Subscriptions.create(new Action0() {
                    @Override
                    public void call() {
                        future.cancel(true);
                    }
                });
            }
        });
    }