D-Net

package eu.dnetlib.enabling.database.resultset;

import java.util.Iterator;

import java.util.NoSuchElementException;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.TimeUnit;

import eu.dnetlib.enabling.database.rmi.DatabaseException;

import eu.dnetlib.enabling.database.utils.DatabaseUtils;

import eu.dnetlib.enabling.resultset.SizedIterable;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.dom4j.Document;

public class IterableRowSet implements SizedIterable<String> {

	private String db;

	private String sql;

	private String sqlForCount;

	private DatabaseUtils dbUtils;

	private Integer size = null;

	private static final Log log = LogFactory.getLog(IterableRowSet.class); // NOPMD by marko on 11/24/08 5:02 PM

	public IterableRowSet(final String db, final String sql, final String sqlForCount, final DatabaseUtils dbUtils) {

		super();

		this.db = db;

		this.sql = sql;

		this.sqlForCount = sqlForCount;

		this.dbUtils = dbUtils;

	}

	@Override

	@SuppressWarnings("unchecked")

	public Iterator<String> iterator() {

		try {

			final BlockingQueue<Document> queue = dbUtils.executeSql(db, sql, BlockingQueue.class);

			return new QueueIterator(queue);

		} catch (Exception e) {

			throw new RuntimeException("Error creating iterator for query: " + sql, e);

		}

	}

	@Override

	public int getNumberOfElements() {

		if (size != null) { return size; }

		String query = (sqlForCount == null) || sqlForCount.isEmpty() ? "SELECT count(*) FROM ( " + sql + " ) AS TABLELISTENER" : sqlForCount;

		try {

			log.debug("Calculating size using query: " + query);

			this.size = dbUtils.executeSql(db, query, Integer.class);

			return size;

		} catch (DatabaseException e) {

			log.error("Error in getSize, query: " + query, e);

			throw new IllegalStateException("Error in getSize, query: " + query, e);

		}

	}

	private class QueueIterator implements Iterator<String> {

		private Document curr;

		private BlockingQueue<Document> queue;

		public QueueIterator(final BlockingQueue<Document> queue) throws InterruptedException {

			super();

			this.queue = queue;

			this.curr = queue.poll(DatabaseUtils.BLOCKING_QUEUE_TIMEOUT, TimeUnit.SECONDS);

		}

		@Override

		public String next() {

			log.debug("Reading Next Element from queue");

			try {

				if ((curr == null) || !hasContent(curr)) { throw new NoSuchElementException(

						"Value from queue is null or empty, probably the producer doesn't produce"); }

				String res = curr.asXML();

				curr = queue.poll(DatabaseUtils.BLOCKING_QUEUE_TIMEOUT, TimeUnit.SECONDS);

				return res;

			} catch (Exception e) {

				throw new NoSuchElementException("Error navigating rowset for query: " + sql + "\n" + e.getMessage());

			}

		}

		private boolean hasContent(final Document curr) {

			return curr != null && Integer.parseInt(curr.valueOf("count(/ROW/*)")) > 0;

		}

		@Override

		public boolean hasNext() {

			// An empty Map indicates the end of the Queue

			return (curr != null) && hasContent(curr);

		}

		@Override

		public void remove() {}

	}

}