Rx subscribeOn order

        return endpoints.getRecentConversations(channel, page, Constants.CONVERSATIONS_PAGE_SIZE)
                .subscribeOn(Schedulers.io()) //if I put this on the bottom, everything just works
                .flatMap(new CheckResponse<Conversation>())
                .toMap(new Func1<Conversation, String>() {
                    @Override
                    public String call(Conversation conversation) {
                        String userId = profile.isMerchant() ? profile.getStore().getUid() : profile.getUid();
                        return utils.convertChannelNameToUid(userId.contains(conversation.getReceiverId()) ? conversation.getSenderId() : conversation.getReceiverId());
                    }
                }).flatMap(new Func1<Map<String, Conversation>, Observable<Response<ArrayList<UserInfo>>>>() {
                    @Override
                    public Observable<Response<ArrayList<UserInfo>>> call(Map<String, Conversation> stringConversationMap) {
                        userIdConversationHashMap = stringConversationMap;
                        return endpoints.getDisplayNames(profile.isMerchant() ? Endpoints.ROLE_USERS : Endpoints.ROLE_STORES, new ArrayList<>(stringConversationMap.keySet()));
                    }
                })
                .flatMap(new CheckResponse<UserInfo>())
                .map(new Func1<UserInfo, Conversation>() {
                    @Override
                    public Conversation call(UserInfo userInfo) {
                        Conversation conversation = userIdConversationHashMap.get(userInfo.getId());
                        conversation.setOppositeInfo(userInfo);
                        return conversation;
                    }
                })
                .filter(new NotNullAndHasOpposite())
                .map(new Func1<Conversation, Conversation>() {
                    @Override
                    public Conversation call(Conversation conversation) {
                        getReadStatusRequest.addChannelId(new ChannelId(conversation.getOppositeInfo().getUid()));
                        return conversation;
                    }
                }).zipWith(Endpoints.getConversationStatus(getReadStatusRequest).flatMap(new CheckResponse<Status>()), new AssignStatus())
                .observeOn(AndroidSchedulers.mainThread()).doOnNext(new Action1<Conversation>() {
                    @Override
                    public void call(final Conversation conversation) {
                        handler.post(new Runnable() {
                            @Override
                            public void run() {
                                repository.addConversation(conversation);
                            }
                        });
                    }
                });
 
 
// HELPER CLASSES
 
      private static class CheckResponse<T> implements Func1<Response<ArrayList<T>>, Observable<T>> {
        @Override
        public Observable<T> call(Response<ArrayList<T>> response) {
            if (!response.isSuccessful()) {
                Log.e(TAG, response.message());
                return Observable.error(new HttpException(response));
            }

            return Observable.from(response.body());
        }
    }