D-Net

package eu.dnetlib.msro.workflows.resultset;

import java.util.List;

import eu.dnetlib.enabling.resultset.client.ResultSetClient;

import org.apache.commons.lang.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import com.google.common.collect.Lists;

import com.googlecode.sarasvati.GraphProcess;

import eu.dnetlib.enabling.resultset.ResultSet;

import eu.dnetlib.enabling.resultset.ResultSetAware;

import eu.dnetlib.enabling.resultset.ResultSetListener;

import eu.dnetlib.enabling.resultset.rmi.ResultSetException;

import eu.dnetlib.enabling.resultset.rmi.ResultSetService;

public class ProcessCountingResultSetListener implements ResultSetListener, ResultSetAware {

	private int count;

	private GraphProcess process;

	private ResultSetService inputService;

	private String inputRsId;

	private ResultSet outputResulset;

	private int inputSize;

	private boolean inaccurate = false;

	private final static String RS_CLOSED = "closed";

	/**

	 * Logger.

	 */

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

	public ProcessCountingResultSetListener(final GraphProcess process, final ResultSetService inputService, final String inputRsId) throws ResultSetException {

		super();

		this.process = process;

		this.inputService = inputService;

		this.inputRsId = inputRsId;

		this.inputSize = inputService.getNumberOfElements(inputRsId);

		this.count = 0;

	}

	@Override

	public List<String> getResult(final int from, final int to) {

		if (process.isCanceled()) {

			log.warn("Process canceled, closing result set");

			this.outputResulset.close();

			return Lists.newArrayList();

		}

		try {

			this.count = to;

			if (count >= this.inputSize) {

				closeRSIfNeeded();

			}

			return inputService.getResult(inputRsId, from, to, "WAITING");

		} catch (ResultSetException e) {

			log.error("Error fetching records from resultset: " + inputRsId);

			throw new RuntimeException(e);

		}

	}

	@Override

	public int getSize() {

		if (process.isCanceled()) {

			this.outputResulset.close();

			return count;

		}

		try {

			int size = inputService.getNumberOfElements(inputRsId);

			if (this.inputSize != size) {

				this.inputSize = size;

				this.inaccurate = true;

			}

			this.inputSize = size;

			if (count >= this.inputSize) {

				this.inaccurate = false;

				closeRSIfNeeded();

			}

			return this.inputSize;

		} catch (ResultSetException e) {

			log.error("Error fetching records from resultset: " + inputRsId);

			throw new RuntimeException(e);

		}

	}

	private void closeRSIfNeeded() throws ResultSetException {

		String inputClassName = inputService.getProperty(inputRsId, "classSimpleName");

		if (StringUtils.isNotBlank(inputClassName) && inputClassName.equalsIgnoreCase("TransientPushResultSetImpl")) {

			// Push result set has inaccurate progress

			this.inaccurate = true;

			log.info("Not closing the counting resultset unless the underlying push result is already closed ");

			if (inputService.getRSStatus(inputRsId).equals(RS_CLOSED)) {

				outputResulset.close();

			}

		} else {

			outputResulset.close();

		}

	}

	@Override

	public void setResultSet(final ResultSet outputResulset) {

		this.outputResulset = outputResulset;

	}

	public int getCount() {

		return count;

	}

	public void setCount(final int count) {

		this.count = count;

	}

	public boolean isInaccurate() {

		return inaccurate;

	}

}