dnet-core/dnet-data-services/src/main/java/eu/dnetlib/data/mdstore/modular/MDStoreFeeder.java

package eu.dnetlib.data.mdstore.modular;

import com.google.common.collect.Maps;
import eu.dnetlib.data.mdstore.MDStoreServiceException;
import eu.dnetlib.data.mdstore.modular.action.DoneCallback;
import eu.dnetlib.data.mdstore.modular.action.FailedCallback;
import eu.dnetlib.data.mdstore.modular.connector.MDStore;
import eu.dnetlib.data.mdstore.modular.connector.MDStoreDao;
import eu.dnetlib.enabling.is.registry.rmi.ISRegistryService;
import eu.dnetlib.enabling.locators.UniqueServiceLocator;
import eu.dnetlib.enabling.resultset.client.ResultSetClientFactory;
import eu.dnetlib.miscutils.datetime.DateUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Required;

import java.util.List;
import java.util.Map;

public class MDStoreFeeder {

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

	private MDStoreDao dao;

	private ResultSetClientFactory resultSetClientFactory;

	private UniqueServiceLocator serviceLocator;

	private boolean syncFeed = true;

	public void feed(final String mdId,
                     final String rsEpr,
                     final String storingType,
                     final boolean sync,
                     final List<MDFormatDescription> mdformats,
                     final DoneCallback doneCallback,
                     final FailedCallback failCallback) throws MDStoreServiceException {
		log.info("Start feeding mdstore " + mdId);
		log.debug("Start feeding mdstore " + mdId + " with epr " + rsEpr);

		String transactionId = null;

		try {
			final boolean refresh = "REFRESH".equals(storingType);

			final MDStore mdstore = dao.startTransaction(mdId, refresh);
			transactionId = mdstore.getId();

			final Iterable<String> records = resultSetClientFactory.getClient(rsEpr);

			if (refresh) {
				mdstore.truncate();
			}
            int writeOps;

            if (mdformats == null) {
                writeOps = mdstore.feed(records, refresh);
            } else {
                writeOps = mdstore.feed(records, refresh, mdformats);
            }

			dao.commit(mdstore.getId(), mdId);

			int size = dao.refreshSize(mdId);

			touch(mdId, size);

			log.info("Finished feeding mdstore " + mdId + " - new size: " + size);

			doneCallback.call(buildParams(size, writeOps));
		} catch (Throwable e) {
			if (transactionId != null) {
				dao.invalidTransaction(transactionId, mdId);
			}
			log.error("Error feeding mdstore: " + mdId);
			failCallback.call(e);
		}
	}

	private Map<String, String> buildParams(final int size, final int storeCount) {
		Map<String, String> params = Maps.newHashMap();
		params.put("mdstoreSize", String.valueOf(size));
		params.put("writeOps", String.valueOf(storeCount));
		return params;
	}

	/**
	 * Sets the last modified date in the profile.
	 *
	 * @param mdId
	 */
	public void touch(final String mdId, final int size) {
		try {
			final String now = DateUtils.now_ISO8601();

			final String mdstoreXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + mdId + "']"
					+ "return update value $x//LAST_STORAGE_DATE with '" + now + "'";

			serviceLocator.getService(ISRegistryService.class).executeXUpdate(mdstoreXUpdate);

			touchSize(mdId, size);
		} catch (final Exception e) {
			throw new IllegalStateException(e);
		}
	}

	public void touchSize(final String mdId, final int size) {
		try {
			final String mdstoreNumberXUpdate = "for $x in //RESOURCE_PROFILE[.//RESOURCE_IDENTIFIER/@value = '" + mdId + "']"
					+ "return update value $x//NUMBER_OF_RECORDS with '" + size + "'";

			serviceLocator.getService(ISRegistryService.class).executeXUpdate(mdstoreNumberXUpdate);
		} catch (final Exception e) {
			throw new IllegalStateException(e);
		}
	}

	public MDStoreDao getDao() {
		return dao;
	}

	@Required
	public void setDao(final MDStoreDao dao) {
		this.dao = dao;
	}

	public ResultSetClientFactory getResultSetClientFactory() {
		return resultSetClientFactory;
	}

	@Required
	public void setResultSetClientFactory(final ResultSetClientFactory resultSetClientFactory) {
		this.resultSetClientFactory = resultSetClientFactory;
	}

	public boolean isSyncFeed() {
		return syncFeed;
	}

	public void setSyncFeed(final boolean syncFeed) {
		this.syncFeed = syncFeed;
	}

	public UniqueServiceLocator getServiceLocator() {
		return serviceLocator;
	}

	@Required
	public void setServiceLocator(final UniqueServiceLocator serviceLocator) {
		this.serviceLocator = serviceLocator;
	}

}