Mercury-IM/data/src/main/java/org/mercury_im/messenger/data/repository/XmppGroupChatRepository.java

package org.mercury_im.messenger.data.repository;

import org.mercury_im.messenger.data.mapping.GroupChatMapping;
import org.mercury_im.messenger.data.model.GroupChatModel;
import org.mercury_im.messenger.data.util.Optional;
import org.mercury_im.messenger.entity.Account;
import org.mercury_im.messenger.entity.chat.GroupChat;
import org.mercury_im.messenger.entity.chat.IGroupChat;
import org.mercury_im.messenger.entity.message.Message;
import org.mercury_im.messenger.util.ThreadUtils;

import java.util.ArrayList;
import java.util.List;

import javax.inject.Inject;
import javax.inject.Named;

import io.reactivex.Completable;
import io.reactivex.Maybe;
import io.reactivex.Observable;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.requery.Persistable;
import io.requery.query.ResultDelegate;
import io.requery.reactivex.ReactiveEntityStore;

public class XmppGroupChatRepository
        extends RequeryRepository
        implements GroupChatRepository {

    @Inject
    GroupChatMapping groupChatMapping;

    @Inject
    public XmppGroupChatRepository(
            ReactiveEntityStore<Persistable> data,
            @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
            @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler) {
        super(data, subscriberScheduler, observerScheduler);
    }

    @Override
    public Single<GroupChat> insertGroupChat(GroupChat chat) {
        return data().insert(groupChatMapping.entityToModel(chat, new GroupChatModel()))
                .map(model -> groupChatMapping.modelToEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<GroupChat>> observeGroupChat(long chatId) {
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ID.eq(chatId))
                .get().observableResult()
                .map(result -> new Optional<>(
                        groupChatMapping.modelToEntity(result.firstOrNull(), new IGroupChat())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<GroupChat> getGroupChat(long chatId) {
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ID.eq(chatId))
                .get().maybe()
                .map(model -> groupChatMapping.modelToEntity(model, new IGroupChat()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<GroupChat> getOrCreateGroupChat(Account account, String roomAddress) {
        return getGroupChatByRoomAddress(account, roomAddress)
                .switchIfEmpty(insertGroupChat(new IGroupChat(){
                    {
                        setAccount(account);
                        setRoomAddress(roomAddress);
                    }
                }))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<GroupChat>> observeGroupChatByRoomAddress(long accountId, String roomAddress) {
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ADDRESS.eq(roomAddress))
                .and(GroupChatModel.ACCOUNT_ID.eq(accountId))
                .get().observableResult()
                .map(result -> new Optional<>(
                        groupChatMapping.modelToEntity(result.firstOrNull(), new IGroupChat())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<GroupChat> getGroupChatByRoomAddress(long accountId, String roomAddress) {
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ADDRESS.eq(roomAddress))
                .and(GroupChatModel.ACCOUNT_ID.eq(accountId))
                .get().maybe()
                .map(model -> groupChatMapping.modelToEntity(model, new IGroupChat()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<List<GroupChat>> observeAllGroupChats() {
        return data().select(GroupChatModel.class)
                .get().observableResult()
                .map(ResultDelegate::toList)
                .map(list -> {
                    List<GroupChat> entities = new ArrayList<>(list.size());
                    for (GroupChatModel model : list) {
                        entities.add(groupChatMapping.modelToEntity(model, new IGroupChat()));
                    }
                    return entities;
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<GroupChat> updateGroupChat(GroupChat chat) {
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ID.eq(chat.getId()))
                .get().maybe().toSingle() // to single
                .flatMap(model -> {
                    // copy changes from entity to the model
                    model = groupChatMapping.entityToModel(chat, model);
                    // write the updated model back
                    return data().update(model);
                })
                .map(model -> groupChatMapping.modelToEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<GroupChat> upsertGroupChat(GroupChat chat) {
        // Try to fetch model
        return data().select(GroupChatModel.class)
                .where(GroupChatModel.ID.eq(chat.getId()))
                .get().maybe()
                // If it does not exist, create a new model from the entity
                .switchIfEmpty(data().insert(groupChatMapping.entityToModel(chat, new GroupChatModel())))
                .flatMap(model -> {
                    // update the model
                    model = groupChatMapping.entityToModel(chat, model);
                    // write the updated model back
                    return data().update(model);
                })
                .map(model -> groupChatMapping.modelToEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Completable deleteGroupChat(GroupChat chat) {
        return data().delete(GroupChatModel.class)
                .where(GroupChatModel.ID.eq(chat.getId()))
                .get().single().ignoreElement()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Message> insertMessage(GroupChat chat, Message message) {
        return null;
    }

    @Override
    public Observable<List<Message>> observeMessages(GroupChat chat) {
        return null;
    }

    @Override
    public Single<Message> updateMessage(Message message) {
        return null;
    }

    @Override
    public Single<Message> upsertMessage(GroupChat chat, Message message) {
        return null;
    }

    @Override
    public Completable deleteMessage(Message message) {
        return null;
    }
}