D-Net

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;

	}

}