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

package org.mercury_im.messenger.data.repository;

import org.mercury_im.messenger.data.mapping.PeerMapping;
import org.mercury_im.messenger.data.model.PeerModel;
import org.mercury_im.messenger.data.util.Optional;
import org.mercury_im.messenger.entity.Account;
import org.mercury_im.messenger.entity.contact.IPeer;
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.reactivex.ReactiveEntityStore;

public class XmppPeerRepository
        extends RequeryRepository
        implements PeerRepository {

    @Inject
    AccountRepository accountRepository;

    @Inject
    PeerMapping peerMapping;

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

    @Override
    public Single<Peer> insertPeer(Peer peer) {
        return data().insert(peerMapping.entityToModel(peer, new PeerModel()))
                .map(model -> peerMapping.modelToEntity(model, peer))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<Peer>> observePeer(long peerId) {
        return data().select(PeerModel.class)
                .where(PeerModel.ID.eq(peerId))
                .get().observableResult()
                .map(result -> new Optional<>(peerMapping.modelToEntity(result.firstOrNull(), new IPeer())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<Peer> getPeer(long peerId) {
        return data().select(PeerModel.class)
                .where(PeerModel.ID.eq(peerId))
                .get().maybe()
                .map(model -> peerMapping.modelToEntity(model, new IPeer()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<Peer>> observePeerByAddress(long accountId, String address) {
        return data().select(PeerModel.class)
                .where(PeerModel.ACCOUNT_ID.eq(accountId))
                .and(PeerModel.ADDRESS.eq(address))
                .get().observableResult()
                .map(result -> new Optional<>(peerMapping.modelToEntity(result.firstOrNull(), new IPeer())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Peer> getOrCreatePeer(long accountId, String address) {
        return accountRepository.getAccount(accountId).toSingle()
                .flatMap(account -> getOrCreatePeer(account, address));
    }

    @Override
    public Maybe<Peer> getPeerByAddress(long accountId, String address) {
        return data().select(PeerModel.class)
                .where(PeerModel.ACCOUNT_ID.eq(accountId))
                .and(PeerModel.ADDRESS.eq(address))
                .get().maybe()
                .map(model -> peerMapping.modelToEntity(model, new IPeer()))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Peer> getOrCreatePeer(Account _account, String _address) {
        return getPeerByAddress(_account, _address)
                .switchIfEmpty(insertPeer(new IPeer(){
                    {
                        setAccount(_account);
                        setAddress(_address);
                    }
                }))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<List<Peer>> observeAllPeers() {
        return data().select(PeerModel.class)
                .get().observableResult()
                .map(result -> {
                    List<PeerModel> peerModels = result.toList();
                    List<Peer> peerEntities = new ArrayList<>(peerModels.size());
                    for (PeerModel model : peerModels) {
                        peerEntities.add(peerMapping.modelToEntity(model, new IPeer()));
                    }
                    return peerEntities;
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Peer> updatePeer(Peer peer) {
        // In order to update, we fetch the model, update it and write it back.
        return data().select(PeerModel.class)
                .where(PeerModel.ID.eq(peer.getId()))
                .get().maybe().toSingle()
                .flatMap(model -> {
                    // write changes into model
                    model = peerMapping.entityToModel(peer, model);
                    // write model back to db
                    return data().update(model);
                })
                .map(model -> peerMapping.modelToEntity(model, peer))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Peer> upsertPeer(Peer peer) {
        return data().select(PeerModel.class)
                .where(PeerModel.ID.eq(peer.getId()))
                .get().maybe()
                // if not exists, create
                .switchIfEmpty(data().insert(peerMapping.entityToModel(peer, new PeerModel())))
                .flatMap(model -> {
                    // write changes into fetched model
                    model = peerMapping.entityToModel(peer, model);
                    // write changed model back to db
                    return data().update(model);
                })
                .map(model -> peerMapping.modelToEntity(model, peer))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Completable deletePeer(Peer peer) {
        return data().delete(PeerModel.class)
                .where(PeerModel.ID.eq(peer.getId()))
                .get().single().ignoreElement()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }
}