dnet-core/dnet-core-components/src/main/java/eu/dnetlib/enabling/resultset/client/ResultSetPageProvider.java

package eu.dnetlib.enabling.resultset.client;

import java.util.List;
import java.util.NoSuchElementException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import eu.dnetlib.enabling.resultset.client.utils.ResultSetRuntimeException;
import eu.dnetlib.enabling.resultset.client.utils.ResultSetTimeoutException;
import eu.dnetlib.enabling.resultset.rmi.ResultSetException;
import eu.dnetlib.enabling.resultset.rmi.ResultSetService;

/**
 * 
 * @author claudio
 * 
 */
public class ResultSetPageProvider {

	/**
	 * logger
	 */
	private static final Log log = LogFactory.getLog(ResultSetPageProvider.class);

	/**
	 * the resultset service
	 */
	private final ResultSetService resultSet;

	/**
	 * the resultset id
	 */
	private final String rsId;

	/**
	 * positional value of the getResult requests
	 */
	private int fromPosition;

	/**
	 * positional value of the getResult requests
	 */
	private int toPosition;

	/**
	 * actual page size
	 */
	private int pageSize;

	/**
	 * default page size
	 */
	private static int DEFAULT_PAGE_SIZE = 10;

	/**
	 * default max waiting time
	 */
	private static long DEFAULT_MAX_WAIT_TIME = 30000;

	/**
	 * actual max waiting time
	 */
	private long maxWaitTime;

	/**
	 * current wait time
	 */
	private long waitTime;

	/**
	 * request counter used to calculate the waitTime
	 */
	private int delayCount;

	/**
	 * resultset status
	 */
	private String RSStatus;

	/**
	 * number of elements in the resultset
	 */
	private int numberOfElements;

	private final static String RS_CLOSED = "closed";

	private final static String RS_OPEN = "open";

	/**
	 * 
	 * @param resultSet
	 * @param rsId
	 * @throws ResultSetException
	 */
	public ResultSetPageProvider(final ResultSetService resultSet, final String rsId) throws ResultSetRuntimeException {

		this.resultSet = resultSet;
		this.rsId = rsId;
		this.pageSize = DEFAULT_PAGE_SIZE;
		this.maxWaitTime = DEFAULT_MAX_WAIT_TIME;
		fromPosition = toPosition = 0;
		delayCount = 0;
		waitTime = 0;
		updateResultSetStatus();
	}

	public ResultSetPageProvider(final ResultSetService resultSet, final String rsId, final int pageSize) throws ResultSetRuntimeException {

		this(resultSet, rsId);
		this.pageSize = pageSize;
	}

	/**
	 * 
	 * @return
	 * @throws ResultSetTimeoutException
	 * @throws ResultSetException
	 */
	public List<String> nextPage() throws ResultSetTimeoutException, ResultSetRuntimeException {
		do {
			updateResultSetStatus();
			int availableElements = numberOfElements - toPosition;
			log.debug("availableElements: " + availableElements);
			if (availableElements > 0) {
				fromPosition = toPosition + 1;
				if (availableElements < pageSize) {
					toPosition = (fromPosition + availableElements) - 1;
				} else {
					toPosition = (fromPosition + pageSize) - 1;
					delayCount = 0;
				}
				log.debug(" - getting result from " + fromPosition + " to " + toPosition + ", numberOfElements: " + numberOfElements + ", availableElements: "
						+ availableElements);
				try {
					return resultSet.getResult(rsId, fromPosition, toPosition, "waiting");
				} catch (ResultSetException e) {
					log.info(e);
					throw new NoSuchElementException(e.getMessage());
				}
			}
			if (RSStatus.equals(RS_CLOSED) && (availableElements == 0)) return null;
			else {
				stopAndWait(++delayCount);
			}
		} while (true);
	}

	/**
	 * 
	 * @param delayCount
	 * @throws ResultSetTimeoutException
	 */
	private void stopAndWait(final int delayCount) throws ResultSetTimeoutException {
		try {

			waitTime = (long) ((Math.pow(1.2, 10 + delayCount) * 10L)) + 200;
			// waitTime = (long) Math.exp(delayCount);
			if (waitTime > maxWaitTime) {
				log.warn("Timeout getting elements from resultset: " + waitTime + ". next poll would wait more than " + maxWaitTime + " ms");
				throw new ResultSetTimeoutException("Timeout getting elements from resultset: next poll would wait more than " + maxWaitTime + " ms");
			}
			log.debug("resultset client is going to sleep for: " + waitTime);
			// System.out.println("resultset client is going to sleep for: " + waitTime);

			Thread.sleep(waitTime);
		} catch (InterruptedException e) {
			log.error("resultSetClient got InterruptedException", e);
		}
	}

	/**
	 * updates the
	 * 
	 * @throws ResultSetException
	 */
	private void updateResultSetStatus() throws ResultSetRuntimeException {

		try {
			RSStatus = resultSet.getRSStatus(rsId);
			numberOfElements = resultSet.getNumberOfElements(rsId);
			// System.out.println("updateResultSetStatus: size is " + numberOfElements + " and status is " + RSStatus);
		} catch (ResultSetException e) {
			log.warn(e);
			throw new ResultSetRuntimeException(e);
		}
	}

	public int getPageSize() {
		return pageSize;
	}

	public void setPageSize(final int pageSize) {
		if (pageSize <= 0) throw new IllegalArgumentException("parameter 'pageSize' must be grater than zero");
		this.pageSize = pageSize;
	}

	@Deprecated
	public boolean hasElements() throws ResultSetRuntimeException {
		updateResultSetStatus();
		if (RSStatus.equals(RS_OPEN)) return true;
		if (RSStatus.equals(RS_CLOSED) && (numberOfElements == 0)) return false;
		return true;
	}

	public void setMaxWaitTime(final long maxWaitTime) {
		if (maxWaitTime <= 0) throw new IllegalArgumentException("parameter 'maxWaitTime' must be grater than zero");
		this.maxWaitTime = maxWaitTime;
	}

}