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

package org.mercury_im.messenger.data.repository;

import org.mercury_im.messenger.data.model.AccountModel;
import org.mercury_im.messenger.entity.Account;
import org.mercury_im.messenger.data.util.Optional;
import org.mercury_im.messenger.util.ThreadUtils;
import org.mercury_im.messenger.data.mapping.AccountMapping;

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 XmppAccountRepository
        extends RequeryRepository
        implements AccountRepository {

    @Inject
    AccountMapping mapping;

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

    @Override
    public Single<Account> insertAccount(Account account) {
        return data().insert(mapping.entityToData(account))
                .map(mapping::dataToEntity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<Account>> observeAccount(long accountId) {
        return data().select(AccountModel.class)
                .where(AccountModel.ID.eq(accountId))
                .get().observableResult()
                .map(result -> new Optional<>(mapping.dataToEntity(result.firstOrNull())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<Account> getAccount(long accountId) {
        return data().select(AccountModel.class)
                .where(AccountModel.ID.eq(accountId))
                .get().maybe()
                .map(mapping::dataToEntity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<Optional<Account>> observeAccountByAddress(String address) {
        return data().select(AccountModel.class)
                .where(AccountModel.ADDRESS.eq(address))
                .get().observableResult()
                .map(result -> new Optional<>(mapping.dataToEntity(result.firstOrNull())))
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<Account> getAccountByAddress(String address) {
        return data().select(AccountModel.class)
                .where(AccountModel.ADDRESS.eq(address))
                .get().maybe()
                .map(mapping::dataToEntity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Observable<List<Account>> observeAllAccounts() {
        return data().select(AccountModel.class)
                .get().observableResult()
                .map(ResultDelegate::toList)
                .map(list -> {
                    List<Account> entities = new ArrayList<>(list.size());
                    for (AccountModel model : list) {
                        entities.add(mapping.dataToEntity(model));
                    }
                    return entities;
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Account> updateAccount(Account account) {
        // Since we cannot access setId() of AccountModel, we have to query the model by ID and update it manually.
        // https://github.com/requery/requery/issues/616#issuecomment-315685460

        // fetch model
        return data().select(AccountModel.class)
                .where(AccountModel.ID.eq(account.getId()))
                .get().maybe().toSingle() // to single
                .map(model -> {
                    // update it
                    model.setEnabled(account.isEnabled());
                    model.setPassword(account.getAuthentication().getPassword());
                    model.setAddress(account.getAddress());
                    // write the updated model back
                    model = data().update(model).blockingGet();
                    return mapping.dataToEntity(model);
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Single<Account> upsertAccount(Account account) {
        // Try to fetch model
        return data().select(AccountModel.class)
                .where(AccountModel.ID.eq(account.getId()))
                .get().maybe()
                // If it does not exist, create a new model from the entity
                .switchIfEmpty(data().insert(mapping.entityToData(account)))
                // finally
                .map(model -> {
                    // update the model
                    model.setEnabled(account.isEnabled());
                    model.setPassword(account.getAuthentication().getPassword());
                    model.setAddress(account.getAddress());
                    // write the updated model back
                    model = data().update(model).blockingGet();
                    return mapping.dataToEntity(model);
                })
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Completable deleteAccount(Account account) {
        return data().delete(AccountModel.class)
                .where(AccountModel.ID.eq(account.getId()))
                .get().single().ignoreElement() // to completable
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }
}