D-Net

package eu.dnetlib.openaire.directindex.api;

import java.util.Map;

import java.util.concurrent.Executors;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import eu.dnetlib.data.index.CloudIndexClient;

import eu.dnetlib.data.index.CloudIndexClientException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.annotation.Value;

public class ResultSubmitterService {

    private static final Log log = LogFactory.getLog(ResultSubmitterService.class);

    @Resource(name = "resultSubmitterClientMap")

    private IndexClientMap clientMap;

    /**

     * Autocommit feature activation flag

     */

    @Value(value = "${openaire.api.directindex.autocommit.active}")

    private boolean autocommitactive;

    /**

     * Autocommit frequency (Seconds)

     */

    @Value(value = "${openaire.api.directindex.autocommit.frequency}")

    private long commitfrquency = 60;

    private ScheduledExecutorService executor;

    public ResultSubmitterService() {

        executor = Executors.newSingleThreadScheduledExecutor();

        updateCommitSchedule();

    }

    private void updateCommitSchedule() {

        log.info("updating commit schedule");

        executor.scheduleAtFixedRate(() -> {

            if (isAutocommitactive()) {

                try {

                    for (Map.Entry<IndexDsInfo, CloudIndexClient> entry : clientMap.getClients().entrySet()) {

                        final IndexDsInfo i = entry.getKey();

                        final CloudIndexClient client = entry.getValue();

                        log.info("performing commit on " + i.getColl());

                        try {

                            client.commit();

                        } catch (CloudIndexClientException e) {

                            log.error("error performing commit on " + i.getColl(), e);

                        }

                    }

                } catch (Throwable e) {

                    log.error("unable to perform commit", e);

                }

            }

        }, 0, getCommitfrquency(), TimeUnit.SECONDS);

    }

    public boolean isAutocommitactive() {

        return autocommitactive;

    }

    public synchronized void setAutocommitactive(boolean autocommitactive) {

        this.autocommitactive = autocommitactive;

    }

    public long getCommitfrquency() {

        return commitfrquency;

    }

    public synchronized void setCommitfrquency(long commitfrquency) {

        this.commitfrquency = commitfrquency;

    }

}