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

package org.mercury_im.messenger.data.repository;

import org.mercury_im.messenger.data.mapping.AccountMapping;
import org.mercury_im.messenger.data.mapping.DirectChatMapping;
import org.mercury_im.messenger.data.mapping.PeerMapping;
import org.mercury_im.messenger.data.model.DirectChatModel;
import org.mercury_im.messenger.data.repository.dao.DirectChatDao;
import org.mercury_im.messenger.data.util.Optional;
import org.mercury_im.messenger.entity.chat.DirectChat;
import org.mercury_im.messenger.entity.chat.IDirectChat;
import org.mercury_im.messenger.entity.contact.Peer;
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;
import io.requery.reactivex.ReactiveResult;

public class XmppDirectChatRepository
        extends RequeryRepository
        implements DirectChatRepository {

    private final AccountMapping accountMapping;

    private final PeerMapping peerMapping;

    private final DirectChatMapping directChatMapping;

    private final DirectChatDao dao;

    @Inject
    public XmppDirectChatRepository(
            ReactiveEntityStore<Persistable> data,
            @Named(value = ThreadUtils.SCHEDULER_IO) Scheduler subscriberScheduler,
            @Named(value = ThreadUtils.SCHEDULER_UI) Scheduler observerScheduler,
            AccountMapping accountMapping,
            PeerMapping peerMapping,
            DirectChatMapping directChatMapping) {
        super(data, subscriberScheduler, observerScheduler);
        this.accountMapping = accountMapping;
        this.peerMapping = peerMapping;
        this.directChatMapping = directChatMapping;

        this.dao = new DirectChatDao(data);
    }

    @Override
    public Single<DirectChat> insertDirectChat(DirectChat chat) {
        return Single.just(chat)
                // map entity to model
                .map(directChatMapping::toModel)
                .flatMap(dao::insert)
                // map back to entity
                .map(model -> directChatMapping.toEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<DirectChat>> observeDirectChat(long chatId) {
        return data().select(DirectChatModel.class)
                .where(DirectChatModel.ID.eq(chatId))
                .get().observableResult()
                .map(ResultDelegate::firstOrNull)
                .map(directChatMapping::toEntity)
                .map(Optional::new)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<DirectChat> getDirectChat(long chatId) {
        return data().select(DirectChatModel.class)
                .where(DirectChatModel.ID.eq(chatId))
                .get().maybe()
                .map(result -> directChatMapping.toEntity(result, new IDirectChat()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<DirectChat> getOrCreateChatWithPeer(Peer peer) {
        return getDirectChatByPeer(peer)
                .switchIfEmpty(insertDirectChat(new IDirectChat(){
                    {
                        setAccount(peer.getAccount());
                        setPeer(peer);
                    }
                }))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<DirectChat>> observeDirectChatByPeer(Peer peer) {
        return data().select(DirectChatModel.class)
                .where(DirectChatModel.PEER_ID.eq(peer.getId()))
                .get().observableResult()
                .map(ReactiveResult::firstOrNull)
                .map(model -> directChatMapping.toEntity(model, new IDirectChat()))
                .map(Optional::new)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<DirectChat> getDirectChatByPeer(Peer peer) {
        return data().select(DirectChatModel.class)
                .where(DirectChatModel.PEER_ID.eq(peer.getId()))
                .get().maybe()
                .map(model -> directChatMapping.toEntity(model, new IDirectChat()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<List<DirectChat>> observeAllDirectChats() {
        return data().select(DirectChatModel.class)
                .get().observableResult()
                .map(ResultDelegate::toList)
                .map(list -> {
                    List<DirectChat> entities = new ArrayList<>(list.size());
                    for (DirectChatModel model : list) {
                        entities.add(directChatMapping.toEntity(model, new IDirectChat()));
                    }
                    return entities;
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<DirectChat> updateDirectChat(DirectChat chat) {
        return dao.get(chat.getId())
                // fail if not exists
                .toSingle()
                .map(model -> directChatMapping.toModel(chat, model))
                .flatMap(data()::update)
                .map(model -> directChatMapping.toEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    Single<DirectChatModel> getOrCreateDirectChatModel(DirectChat chat) {
        return data()
                // fetch existing chat
                .select(DirectChatModel.class)
                .where(DirectChatModel.ID.eq(chat.getId()))
                .get().maybe()
                .switchIfEmpty(
                        // If not exists, insert chat as new model and return
                        data().insert(directChatMapping.toModel(chat, new DirectChatModel()))
                )
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<DirectChat> upsertDirectChat(DirectChat chat) {
        return getOrCreateDirectChatModel(chat)
                .map(directChatModel -> directChatMapping.toModel(chat, directChatModel))
                .flatMap(data()::update)
                .map(model -> directChatMapping.toEntity(model, chat))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

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