<commit_before>package com.playmoweb.store2store.store;

import com.playmoweb.store2store.utils.Filter;
import com.playmoweb.store2store.utils.SortingMode;

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

import io.reactivex.Flowable;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Function;

/**
 * Abstract StoreService
 *
 * @author <NAME>
 * @by Playmoweb
 * @date 07/09/2017
 */
public abstract class StoreService<T> extends StoreDao<T> {

    /**
     * Class used by the Store
     */
    private final Class<T> clazz;

    /**
     * Default dao used
     */
    private final StoreDao<T> dao;

    /**
     * If true, this Service will be used as a cache (first call).
     * It's a read-only feature (getAll, getById, getOne, ...).
     *
     * @warning If this service is used as a cache and its access is slow, it will reduce drastically the performances of syncing
     */
    private final boolean isCache;

    /**
     * List of store synced with this one
     */
    private StoreService<T> syncedStore;

    /**
     * A local composite disposable to handle local subscriptions
     */
    protected final CompositeDisposable compositeDisposable = new CompositeDisposable();

    /**
     * Create a typed Store
     */
    public StoreService(Class<T> clazz, StoreDao<T> dao) {
        this(clazz, dao, false);
    }

    /**
     * Create a typed store with cache information
     */
    public StoreService(Class<T> clazz, StoreDao<T> dao, boolean isCache) {
        this.clazz = clazz;
        this.dao = dao;
        this.isCache = isCache;
    }

    /**
     * Add a Store to be synced to
     */
    public StoreService<T> syncWith(StoreService<T> otherStore) {
        syncedStore = otherStore;
        return this;
    }

    /**
     * Get the dao used by this Store
     *
     * @return
     */
    public StoreDao<T> getDao() {
        return dao;
    }

    /**
     * Clear all remaining disposables
     */
    public void clearAllDisposables() {
        compositeDisposable.clear();
    }

    /**
     * This Store is Synced with another
     *
     * @return
     */
    public boolean hasSyncedStore() {
        return syncedStore != null;
    }

    public boolean isCache() {
        return isCache;
    }

    @Override
    public Flowable<Optional<List<T>>> getAll(final Filter filter, final SortingMode sortingMode) {
        List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        Flowable<Optional<List<T>>> flowStorage = dao.getAll(filter, sortingMode);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                            final List<T> copy = new ArrayList<>(items.get());
                            if (filter == null) {
                                // full replacement, we clean up the Store dao
                                return syncedStore.deleteAll().map(new Function<Integer, Optional<List<T>>>() {
                                    @Override
                                    public Optional<List<T>> apply(Integer integer) throws Exception {
                                        return Optional.wrap(copy);
                                    }
                                });
                            }
                            return Flowable.just(Optional.wrap(copy));
                        }
                    })
                    .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                            return syncedStore.insertOrUpdate(items.get());
                        }
                    });

            if (syncedStore.isCache()) {
                return flowStorage.startWith(syncedStore.getAll(filter, sortingMode));
            }

            flowables.add(syncedStore.getAll(filter, sortingMode));
        }

        flowables.add(flowStorage);
        return Flowable.concat(flowables);
    }

    @Override
    public Flowable<Optional<List<T>>> getAll(final List<T> items) {
        List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        Flowable<Optional<List<T>>> flowStorage = dao.getAll(items);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                            return syncedStore.insertOrUpdate(items.get());
                        }
                    });

            if (syncedStore.isCache()) {
                return flowStorage.startWith(syncedStore.getAll(items));
            }

            flowables.add(syncedStore.getAll(items));
        }

        flowables.add(flowStorage);
        return Flowable.concat(flowables);
    }

    public final Flowable<Optional<List<T>>> getAll(final Filter filter) {
        return getAll(filter, SortingMode.DEFAULT);
    }

    public final Flowable<Optional<List<T>>> getAll() {
        return getAll(Filter.NONE);
    }

    @Override
    public Flowable<Optional<T>> getOne(final Filter filter, final SortingMode sortingMode) {
        List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        Flowable<Optional<T>> flowStorage = dao.getOne(filter, sortingMode);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
                        @Override
                        public Flowable<Optional<T>> apply(Optional<T> item) throws Exception {
                            return syncedStore.insertOrUpdate(item.get());
                        }
                    });

            if (syncedStore.isCache()) {
                return flowStorage.startWith(syncedStore.getOne(filter, sortingMode));
            }

            flowables.add(syncedStore.getOne(filter, sortingMode));
        }

        flowables.add(flowStorage);
        return Flowable.concat(flowables);
    }

    @Override
    public Flowable<Optional<T>> getOne(final T item) {
        List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        Flowable<Optional<T>> flowStorage = dao.getOne(item);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
                        @Override
                        public Flowable<Optional<T>> apply(Optional<T> item) throws Exception {
                            return syncedStore.insertOrUpdate(item.get());
                        }
                    });
            if (syncedStore.isCache()) {
                return flowStorage.startWith(syncedStore.getOne(item));
            }
            flowables.add(syncedStore.getOne(item));
        }

        flowables.add(flowStorage);
        return Flowable.concat(flowables);
    }

    public Flowable<Optional<T>> getOne(final Filter filter) {
        return getOne(filter, null);
    }

    public Flowable<Optional<T>> getOne(final SortingMode sortingMode) {
        return getOne(Filter.NONE, sortingMode);
    }

    public Flowable<Optional<T>> getOne() {
        return getOne(Filter.NONE, null);
    }

    @Override
    public Flowable<Optional<T>> getById(final int id) {
        List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        Flowable<Optional<T>> flowStorage = dao.getById(id);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
                        @Override
                        public Flowable<Optional<T>> apply(final Optional<T> item) throws Exception {
                            return syncedStore.insertOrUpdate(item.get());
                        }
                    });
            if (syncedStore.isCache()) {
                return flowStorage.startWith(syncedStore.getById(id));
            }
            flowables.add(syncedStore.getById(id));
        }

        flowables.add(flowStorage);
        return Flowable.concat(flowables);
    }

    @Override
    public Flowable<Optional<List<T>>> insert(final List<T> items) {
        List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        Flowable<Optional<List<T>>> flowStorage = dao.insert(items);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .onErrorResumeNext(new Function<Throwable, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(final Throwable throwable) throws Exception {
                            return syncedStore.delete(items).flatMap(new Function<Object, Flowable<Optional<List<T>>>>() {
                                @Override
                                public Flowable<Optional<List<T>>> apply(Object o) throws Exception {
                                    return Flowable.error(throwable);
                                }
                            });
                        }
                    })
                    .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(Optional<List<T>> it) throws Exception {<commit_msg>Improve cache error dispatching with delayError<commit_after>package com.playmoweb.store2store.store;

import com.playmoweb.store2store.utils.Filter;
import com.playmoweb.store2store.utils.SortingMode;

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

import io.reactivex.Flowable;
import io.reactivex.disposables.CompositeDisposable;
import io.reactivex.functions.Function;

/**
 * Abstract StoreService
 *
 * @author <NAME>
 * @by Playmoweb
 * @date 07/09/2017
 */
public abstract class StoreService<T> extends StoreDao<T> {

    /**
     * Class used by the Store
     */
    private final Class<T> clazz;

    /**
     * Default dao used
     */
    private final StoreDao<T> dao;

    /**
     * If true, this Service will be used as a cache (first call).
     * It's a read-only feature (getAll, getById, getOne, ...).
     *
     * @warning If this service is used as a cache and its access is slow, it will reduce drastically the performances of syncing
     */
    private final boolean isCache;

    /**
     * List of store synced with this one
     */
    private StoreService<T> syncedStore;

    /**
     * A local composite disposable to handle local subscriptions
     */
    protected final CompositeDisposable compositeDisposable = new CompositeDisposable();

    /**
     * Create a typed Store
     */
    public StoreService(Class<T> clazz, StoreDao<T> dao) {
        this(clazz, dao, false);
    }

    /**
     * Create a typed store with cache information
     */
    public StoreService(Class<T> clazz, StoreDao<T> dao, boolean isCache) {
        this.clazz = clazz;
        this.dao = dao;
        this.isCache = isCache;
    }

    /**
     * Add a Store to be synced to
     */
    public StoreService<T> syncWith(StoreService<T> otherStore) {
        syncedStore = otherStore;
        return this;
    }

    /**
     * Get the dao used by this Store
     *
     * @return
     */
    public StoreDao<T> getDao() {
        return dao;
    }

    /**
     * Clear all remaining disposables
     */
    public void clearAllDisposables() {
        compositeDisposable.clear();
    }

    /**
     * This Store is Synced with another
     *
     * @return
     */
    public boolean hasSyncedStore() {
        return syncedStore != null;
    }

    public boolean isCache() {
        return isCache;
    }

    @Override
    public Flowable<Optional<List<T>>> getAll(final Filter filter, final SortingMode sortingMode) {
        Flowable<Optional<List<T>>> flowStorage = dao.getAll(filter, sortingMode);

        if (!hasSyncedStore()) {
            return flowStorage;
        }

        final List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        flowStorage = flowStorage
                .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                    @Override
                    public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                        final List<T> copy = new ArrayList<>(items.get());
                        if (filter == null) {
                            // full replacement, we clean up the Store dao
                            return syncedStore.deleteAll().map(new Function<Integer, Optional<List<T>>>() {
                                @Override
                                public Optional<List<T>> apply(Integer integer) throws Exception {
                                    return Optional.wrap(copy);
                                }
                            });
                        }
                        return Flowable.just(Optional.wrap(copy));
                    }
                })
                .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                    @Override
                    public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                        return syncedStore.insertOrUpdate(items.get());
                    }
                });

        flowables.add(syncedStore.getAll(filter, sortingMode));
        flowables.add(flowStorage);

        if (syncedStore.isCache()) {
            return Flowable.concatDelayError(flowables);
        } else {
            return Flowable.concat(flowables);
        }
    }

    @Override
    public Flowable<Optional<List<T>>> getAll(final List<T> items) {
        Flowable<Optional<List<T>>> flowStorage = dao.getAll(items);

        if (!hasSyncedStore()) {
            return flowStorage;
        }

        final List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        flowStorage = flowStorage.flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
            @Override
            public Flowable<Optional<List<T>>> apply(Optional<List<T>> items) throws Exception {
                return syncedStore.insertOrUpdate(items.get());
            }
        });
        flowables.add(syncedStore.getAll(items));
        flowables.add(flowStorage);

        if (syncedStore.isCache()) {
            return Flowable.concatDelayError(flowables);
        } else {
            return Flowable.concat(flowables);
        }
    }

    public final Flowable<Optional<List<T>>> getAll(final Filter filter) {
        return getAll(filter, SortingMode.DEFAULT);
    }

    public final Flowable<Optional<List<T>>> getAll() {
        return getAll(Filter.NONE);
    }

    @Override
    public Flowable<Optional<T>> getOne(final Filter filter, final SortingMode sortingMode) {
        Flowable<Optional<T>> flowStorage = dao.getOne(filter, sortingMode);

        if (!hasSyncedStore()) {
            return flowStorage;
        }

        final List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        flowStorage = flowStorage.flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
            @Override
            public Flowable<Optional<T>> apply(Optional<T> item) throws Exception {
                return syncedStore.insertOrUpdate(item.get());
            }
        });
        flowables.add(syncedStore.getOne(filter, sortingMode));
        flowables.add(flowStorage);

        if (syncedStore.isCache()) {
            return Flowable.concatDelayError(flowables);
        } else {
            return Flowable.concat(flowables);
        }
    }

    @Override
    public Flowable<Optional<T>> getOne(final T item) {
        Flowable<Optional<T>> flowStorage = dao.getOne(item);

        if (!hasSyncedStore()) {
            return flowStorage;
        }

        final List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        flowStorage = flowStorage.flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
            @Override
            public Flowable<Optional<T>> apply(Optional<T> item) throws Exception {
                return syncedStore.insertOrUpdate(item.get());
            }
        });
        flowables.add(syncedStore.getOne(item));
        flowables.add(flowStorage);

        if (syncedStore.isCache()) {
            return Flowable.concatDelayError(flowables);
        } else {
            return Flowable.concat(flowables);
        }
    }

    public Flowable<Optional<T>> getOne(final Filter filter) {
        return getOne(filter, null);
    }

    public Flowable<Optional<T>> getOne(final SortingMode sortingMode) {
        return getOne(Filter.NONE, sortingMode);
    }

    public Flowable<Optional<T>> getOne() {
        return getOne(Filter.NONE, null);
    }

    @Override
    public Flowable<Optional<T>> getById(final int id) {
        Flowable<Optional<T>> flowStorage = dao.getById(id);
        if (!hasSyncedStore()) {
            return flowStorage;
        }

        final List<Flowable<Optional<T>>> flowables = new ArrayList<>();
        flowStorage = flowStorage.flatMap(new Function<Optional<T>, Flowable<Optional<T>>>() {
            @Override
            public Flowable<Optional<T>> apply(final Optional<T> item) throws Exception {
                return syncedStore.insertOrUpdate(item.get());
            }
        });
        flowables.add(syncedStore.getById(id));
        flowables.add(flowStorage);

        if (syncedStore.isCache()) {
            return Flowable.concatDelayError(flowables);
        } else {
            return Flowable.concat(flowables);
        }
    }

    @Override
    public Flowable<Optional<List<T>>> insert(final List<T> items) {
        List<Flowable<Optional<List<T>>>> flowables = new ArrayList<>();
        Flowable<Optional<List<T>>> flowStorage = dao.insert(items);

        if (hasSyncedStore()) {
            flowStorage = flowStorage
                    .onErrorResumeNext(new Function<Throwable, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(final Throwable throwable) throws Exception {
                            return syncedStore.delete(items).flatMap(new Function<Object, Flowable<Optional<List<T>>>>() {
                                @Override
                                public Flowable<Optional<List<T>>> apply(Object o) throws Exception {
                                    return Flowable.error(throwable);
                                }
                            });
                        }
                    })
                    .flatMap(new Function<Optional<List<T>>, Flowable<Optional<List<T>>>>() {
                        @Override
                        public Flowable<Optional<List<T>>> apply(Optional<List<T>> it) throws Exception {
{'fraction_non_alphanumeric': 0.11140487299118715, 'fraction_numerical': 0.0011404872991187143, 'mean_word_length': 1.9009022556390978, 'pattern_counts': {'":': 0, '<': 343, '<?xml version=': 0, '>': 343, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5530006', 'n_tokens_mistral': 5249, 'n_tokens_neox': 5014, 'n_words': 1075}