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

package org.mercury_im.messenger.data.repository;

import org.mercury_im.messenger.data.mapping.EntityCapsMapping;
import org.mercury_im.messenger.data.model.EntityCapsModel;
import org.mercury_im.messenger.entity.caps.EntityCapsRecord;
import org.mercury_im.messenger.util.ThreadUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

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.requery.Persistable;
import io.requery.query.Expression;
import io.requery.query.ResultDelegate;
import io.requery.reactivex.ReactiveEntityStore;
import io.requery.reactivex.ReactiveResult;

public class XmppEntityCapsRepository extends RequeryRepository implements EntityCapsRepository {

    private final EntityCapsMapping entityCapsMapping;

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

    @Override
    public Observable<Map<String, EntityCapsRecord>> observeAllEntityCapsRecords() {
        return data().select(EntityCapsModel.class).get()
                .observableResult()
                .map(result -> result.toMap(EntityCapsModel.NODE_VER, new ConcurrentHashMap<>()))
                .map(this::mapModelsToEntities)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    private Map<String, EntityCapsRecord> mapModelsToEntities(Map<String, EntityCapsModel> models) {
        Map<String, EntityCapsRecord> entities = new ConcurrentHashMap<>();
        for (String key : models.keySet()) {
            entities.put(key, entityCapsMapping.toEntity(models.get(key)));
        }
        return entities;
    }

    private Map<String, EntityCapsModel> mapEntitiesToModels(Map<String, EntityCapsRecord> entities) {
        Map<String, EntityCapsModel> models = new ConcurrentHashMap<>();
        for (String key : entities.keySet()) {
            models.put(key, entityCapsMapping.toModel(entities.get(key)));
        }
        return models;
    }

    @Override
    public Observable<EntityCapsRecord> observeEntityCapsRecords() {
        return data().select(EntityCapsModel.class)
                .get().observableResult()
                .flatMap(ReactiveResult::observable)
                .map(entityCapsMapping::toEntity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Maybe<EntityCapsRecord> maybeGetEntityCapsRecord(String nodeVer) {
        return data().select(EntityCapsModel.class)
                .where(EntityCapsModel.NODE_VER.eq(nodeVer))
                .get().maybe()
                .map(entityCapsMapping::toEntity)
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }

    @Override
    public Completable insertEntityCapsRecord(EntityCapsRecord entityCapsRecord) {
        return data().upsert(entityCapsMapping.toModel(entityCapsRecord))
                .ignoreElement()
                .subscribeOn(subscriberScheduler())
                .observeOn(observerScheduler());
    }
}