D-Net

package eu.dnetlib.enabling.resultset.client.utils;

import java.io.StringReader;

import javax.xml.transform.stream.StreamSource;

import javax.xml.ws.EndpointReference;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

/**

 * 

 * @author claudio

 *

 */

public class EPRUtils {

	/**

	 * builds an epr from its string representation

	 * 

	 * @param epr

	 * 			String epr

	 * @return 

	 * 			W3CEndpointReference epr

	 * 		

	 */

	public W3CEndpointReference getEpr(String epr) {

		return (W3CEndpointReference) EndpointReference.readFrom(new StreamSource(new StringReader(epr)));

	}

}

package eu.dnetlib.enabling.resultset.client.utils;

public class ResultSetRuntimeException extends RuntimeException {

	/**

	 * 

	 */

	private static final long serialVersionUID = -5131499590327995897L;

	public ResultSetRuntimeException(String message) {

		super(message);

	}

	public ResultSetRuntimeException(Throwable e) {

		super(e);

	}

}

package eu.dnetlib.enabling.resultset.client.utils;

public class ResultSetTimeoutException extends RuntimeException {

	private static final long serialVersionUID = -3713991101055085620L;

	public ResultSetTimeoutException(String message) {

		super(message);

	}

}

package eu.dnetlib.enabling.resultset.client;

import java.util.Iterator;

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

/**

 * 

 * @author claudio

 *

 */

public class IterableResultSetClient implements Iterable<String> {

	/**

	 * reference to resultset service.

	 */

	private ResultSetService resultSet;

	/**

	 * resultset id

	 */

	private String rsId;

	/**

	 * page size.

	 */

	private int pageSize;

	/**

	 * timeout

	 */

	private long timeout;

	public IterableResultSetClient(ResultSetService resultSet, String rsId, int pageSize) {

		this.resultSet = resultSet;

		this.rsId = rsId;

		this.pageSize = pageSize;

		this.timeout = 0;

	} 

	public IterableResultSetClient(ResultSetService resultSet, String rsId, int pageSize, long timeout) {

		this(resultSet, rsId, pageSize);

		this.timeout = timeout;

	} 

	@Override

	public Iterator<String> iterator() {

		if (timeout == 0)

			return new ResultSetClientIterator(resultSet, rsId, pageSize);

		return new ResultSetClientIterator(resultSet, rsId, pageSize, timeout);

	}

	public int getPageSize() {

		return pageSize;

	}

	public void setPageSize(int pageSize) {

		this.pageSize = pageSize;

	}

}

package eu.dnetlib.enabling.resultset.client;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import eu.dnetlib.enabling.resultset.client.utils.EPRUtils;

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

import eu.dnetlib.enabling.tools.ServiceResolver;

/**

 * 

 * @author claudio

 * 

 */

public class ResultSetClientFactory implements ResultSetClient {

	/**

	 * logger

	 */

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

	/**

	 * used to resolve the epr references to the service endpoint

	 */

	private ServiceResolver serviceResolver;

	/**

	 * utility object

	 */

	private EPRUtils eprUtils;

	/**

	 * actual page size

	 */

	private int pageSize;

	/**

	 * actual timeout

	 */

	private long timeout;

	public ResultSetClientFactory() {

		log.info("creating new ResultSetClientIterableFactory with default pageSize and timeout");

	}

	/**

	 * 

	 * @param pageSize

	 * @param timeout

	 * @throws IllegalArgumentException

	 */

	public ResultSetClientFactory(int pageSize, long timeout) throws IllegalArgumentException {

		if (pageSize <= 0 || timeout <= 0)

			throw new IllegalArgumentException("parameters pageSize and timeout must be greater than zero");

		this.pageSize = pageSize;

		this.timeout = timeout;

	}

	/**

	 * 

	 * @param epr

	 * @param pageSize

	 * @return

	 */

	@Override

	public IterableResultSetClient getClient(W3CEndpointReference epr, int pageSize) {

		final ResultSetService resultSet = serviceResolver.getService(ResultSetService.class, epr);

		final String rsId = serviceResolver.getResourceIdentifier(epr);

		//using given pageSize and default timeout

		return new IterableResultSetClient(resultSet, rsId, pageSize, timeout);

	}

	/**

	 * 

	 * @param epr

	 * @return

	 */

	@Override

	public IterableResultSetClient getClient(W3CEndpointReference epr) {

		final ResultSetService resultSet = serviceResolver.getService(ResultSetService.class, epr);

		final String rsId = serviceResolver.getResourceIdentifier(epr);

		//using default pageSize and timeout

		return new IterableResultSetClient(resultSet, rsId, pageSize, timeout);

	}

	/**

	 * 

	 * @param stringEpr

	 * @param pageSize

	 * @return

	 */

	@Override

	public IterableResultSetClient getClient(String stringEpr, int pageSize) {

		return getClient(eprUtils.getEpr(stringEpr), pageSize);

	}

	/**

	 * 

	 * @param stringEpr

	 * @return

	 */

	@Override

	public IterableResultSetClient getClient(String stringEpr) {

		return getClient(eprUtils.getEpr(stringEpr));

	}

	@Required

	public void setServiceResolver(ServiceResolver serviceResolver) {

		this.serviceResolver = serviceResolver;

	}

	@Required

	public void setEprUtils(EPRUtils eprUtils) {

		this.eprUtils = eprUtils;

	}

	public int getPageSize() {

		return pageSize;

	}

	@Required

	public void setPageSize(int pageSize) {

		this.pageSize = pageSize;

	}

	public long getTimeout() {

		return timeout;

	}

	@Required

	public void setTimeout(long timeout) {

		this.timeout = timeout;

	}

}

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(ResultSetService resultSet, 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(ResultSetService resultSet, String rsId, 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.info(" - getting result from " + fromPosition + " to " + toPosition + ", numberOfElements: " + numberOfElements + ", availableElements: "+ availableElements);

				//System.out.println(" - 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))

			if (RSStatus.equals(RS_CLOSED) && availableElements == 0)

				return null;

			else

				stopAndWait(++delayCount);

		} while (true);

	}

	/**

	 * 

	 * @param delayCount

	 * @throws ResultSetTimeoutException

	 */

	private void stopAndWait(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(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(long maxWaitTime) {

		if (maxWaitTime <= 0)

			throw new IllegalArgumentException("parameter 'maxWaitTime' must be grater than zero");

		this.maxWaitTime = maxWaitTime;

	}

}

package eu.dnetlib.enabling.resultset.client;

import java.util.Iterator;

import java.util.LinkedList;

import java.util.List;

import java.util.NoSuchElementException;

import java.util.Queue;

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 ResultSetClientIterator implements Iterator<String> {

	/**

	 * logger

	 */

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

	/**

	 * the page source 

	 */

	private ResultSetPageProvider pageProvider;

	/**

	 * buffer used to provide a single element for the next() method

	 */

	private Queue<String> buffer;

	public ResultSetClientIterator(ResultSetService resultSet, String rsId) 

		throws ResultSetRuntimeException {

		pageProvider = new ResultSetPageProvider(resultSet, rsId);

		buffer = new LinkedList<String>();

	}

	public ResultSetClientIterator(ResultSetService resultSet, String rsId, int pageSize) {

		this(resultSet, rsId);

		pageProvider.setPageSize(pageSize);

	}

	public ResultSetClientIterator(ResultSetService resultSet, String rsId, int pageSize, long timeout) {

		this(resultSet, rsId, pageSize);

		pageProvider.setMaxWaitTime(timeout);

	}

	/**

	 * Tries to refill the buffer with a nextPage()

	 * @return true if the buffer was filled successfully, false otherwise

	 * @throws ResultSetTimeoutException in case of timeout

	 * @throws ResultSetException 

	 * @throws ResultSetException 

	 */

	private boolean refillBuffer() throws ResultSetTimeoutException, ResultSetRuntimeException {

		List<String> page = pageProvider.nextPage();

		if (page != null && !page.isEmpty()) {

			buffer.addAll(page);

			return true;

		}

		return false;

	}

	@Override

	public boolean hasNext() {

		if (!buffer.isEmpty())

			return true;

		return refillBuffer();

	}

	@Override

	public String next() {

		if (!hasNext()) {

			log.info("NoSuchElementException");

			throw new NoSuchElementException();

		}

		return buffer.poll();

	}

	@Override

	public void remove() {

		throw new UnsupportedOperationException();

	}

}

package eu.dnetlib.enabling.resultset.client;

import javax.xml.ws.wsaddressing.W3CEndpointReference;

/**

 * 

 * @author claudio

 *

 */

public interface ResultSetClient {

	/**

	 * 

	 * @param epr

	 * @return

	 */

	public IterableResultSetClient getClient(W3CEndpointReference epr);

	/**

	 * 

	 * @param epr

	 * @return

	 */

	public IterableResultSetClient getClient(String epr);

	/**

	 * 

	 * @param epr

	 * @param pageSize

	 * @return

	 */

	public IterableResultSetClient getClient(W3CEndpointReference epr, int pageSize);

	/**

	 * 

	 * @param epr

	 * @param pageSize

	 * @return

	 */

	public IterableResultSetClient getClient(String epr, int pageSize);

}