D-Net

package eu.dnetlib.data.download.worker;

import java.io.IOException;

import java.net.HttpURLConnection;

import java.net.URL;

import java.net.URLConnection;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.Callable;

import java.util.concurrent.TimeUnit;

import com.google.common.base.Function;

import com.google.common.base.Joiner;

import eu.dnetlib.data.download.DownloadReportMap;

import eu.dnetlib.data.download.DownloadServiceImpl;

import eu.dnetlib.data.download.rmi.DownloadItem;

import eu.dnetlib.data.download.rmi.DownloadServiceFeeder;

import eu.dnetlib.data.objectstore.modular.ObjectStoreRecord;

import eu.dnetlib.data.objectstore.modular.connector.ObjectStore;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreFile;

import eu.dnetlib.data.objectstore.rmi.ObjectStoreServiceException;

import eu.dnetlib.data.objectstore.rmi.Protocols;

import org.apache.commons.lang3.StringUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * The Class DownloadWorker is a worker responsible to download the data into the object store.

 */

public class DownloadWorker implements Callable<DownloadReportMap> {

	/**

	 * The Constant log.

	 */

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

	private static final int MAX_NULLS = 5 ;

	/**

	 * The queue.

	 */

	private BlockingQueue<String> queue = null;

	/**

	 * The object store.

	 */

	private ObjectStore objectStore = null;

	/**

	 * The protocol.

	 */

	private Protocols protocol;

	/**

	 * The mime type.

	 */

	private String mimeType;

	private Function<String, DownloadItem> converter;

	private int connectTimeoutMs;

	private int readTimeoutMs;

	/**

	 * ms to wait between two subsequent download request

	 **/

	private int sleepTimeMs;

	private String userAgent = "Mozilla/5.0 (compatible; OAI; +http://www.openaire.eu)";

	/**

	 * Instantiates a new download worker.

	 *

	 * @param queue

	 *            the queue

	 * @param objectStore

	 *            the object store

	 * @param protocol

	 *            the protocol

	 * @param mimeType

	 *            the mime type

	 */

	public DownloadWorker(final BlockingQueue<String> queue,

			final ObjectStore objectStore,

			final Protocols protocol,

			final String mimeType,

			final int connectTimeoutMs,

			final int readTimeoutMs, final int sleepTimeMs,

			final Function<String, DownloadItem> converter) {

		this.setConverter(converter);

		this.setQueue(queue);

		this.setObjectStore(objectStore);

		this.setMimeType(mimeType);

		this.setProtocol(protocol);

		this.setConnectTimeoutMs(connectTimeoutMs);

		this.setReadTimeoutMs(readTimeoutMs);

		this.setSleepTimeMs(sleepTimeMs);

	}

	/*

	 * (non-Javadoc)

	 *

	 * @see java.lang.Runnable#run()

	 */

	@Override

	public DownloadReportMap call() throws Exception {

		final DownloadReportMap report = new DownloadReportMap();

		final long threadId = Thread.currentThread().getId();

		int nullCounter = 0;

		try {

			String takedObject = queue.poll(5, TimeUnit.SECONDS);

			while (!DownloadServiceImpl.END_QUEUE_STRING.equals(takedObject) && nullCounter < MAX_NULLS) {

				if (takedObject == null) {

					nullCounter++;

				}

				if (log.isDebugEnabled()) {

					log.debug(takedObject);

				}

				final DownloadItem di = getConverter().apply(takedObject);

				log.debug(threadId + ": Reading " + takedObject);

				if (di == null) {

					log.debug("the current download item is Null, skipping");

					DownloadServiceFeeder.reportException(report, null, new IllegalArgumentException("found null DownloadItem"));

				} else if (StringUtils.isNotBlank(di.getUrl()) && !checkIfExists(di)) {

					doDownload(threadId, report, di);

				}

				takedObject = queue.poll(5, TimeUnit.SECONDS);

				log.debug(String.format("%s: next object from queue %s, remaining items: %s", threadId, takedObject, queue.size()));

			}

		} catch (final Exception e) {

			log.debug("An error occured : " + Joiner.on("\tat ").join(e.getStackTrace()));

			DownloadServiceFeeder.reportException(report, null, e);

			report.setStatus(false);

			return report;

		} finally {

			log.info(String.format("%s: finalising queue, remaining items: %s, nulls: %s", threadId, queue.size(), nullCounter));

			final boolean res = queue.offer(DownloadServiceImpl.END_QUEUE_STRING, 5, TimeUnit.SECONDS);

			log.info("put terminator in queue: " + res);

		}

		log.info("CLOSED THREAD " + threadId);

		report.setStatus(true);

		return report;

	}

	public void doDownload(final long threadId, final DownloadReportMap report, final DownloadItem di) {

		try {

			if (getSleepTimeMs() > 0) {

				log.debug(threadId + ": I will sleep for " + getSleepTimeMs() + " ms, as requested...");

				Thread.sleep(getSleepTimeMs());

			}

			final URL toDownload = followURL(threadId, new URL(di.getUrl()), report, di);

			final ObjectStoreRecord record = new ObjectStoreRecord();

			final ObjectStoreFile metadata = new ObjectStoreFile();

			metadata.setObjectID(di.getFileName());

			metadata.setMetadataRelatedID(di.getIdItemMetadata());

			metadata.setAccessProtocol(this.protocol);

			metadata.setMimeType(this.mimeType);

			metadata.setDownloadedURL(di.getOriginalUrl());

			record.setFileMetadata(metadata);

			log.debug(threadId + ": opening connection " + toDownload);

			final URLConnection connection = toDownload.openConnection();

			// set user-agent

			connection.setRequestProperty("User-Agent", userAgent);

			connection.setConnectTimeout(getConnectTimeoutMs());

			connection.setReadTimeout(getReadTimeoutMs());

			connection.setRequestProperty("User-Agent", "OpenAIRE aggregation system/2.2.5 ; (OpenAccess aggregator; mailto:helpdesk@openaire.eu) ");

			log.debug(threadId + ": getting input stream from " + toDownload);

			record.setInputStream(connection.getInputStream());

			log.debug(threadId + ": feeding object from  " + toDownload + " into objectstore ...");

			objectStore.feedObjectRecord(record);

			report.addDowload();

			log.debug(threadId + ": saved object " + metadata.toJSON());

		} catch (final Throwable e) {

			log.debug(threadId + ": error downloading Item: " + di.toJSON(), e);

			DownloadServiceFeeder.reportException(report, di, e);

		}

	}

	private URL followURL(final long threadId, final URL inputURL, final DownloadReportMap report, final DownloadItem di) throws IOException {

		final String ptrcl = inputURL.getProtocol();

		if (ptrcl.startsWith("file")) {

			log.debug("the protocol is File, returning " + inputURL);

			return inputURL;

		}

		HttpURLConnection conn = (HttpURLConnection) inputURL.openConnection();

		conn.setInstanceFollowRedirects(true);  // you still need to handle redirect manually.

		conn.setReadTimeout(getReadTimeoutMs());

		conn.setConnectTimeout(getConnectTimeoutMs());

		// set user-agent

		conn.setRequestProperty("User-Agent", userAgent);

		String location = inputURL.toString();

		log.debug(threadId + " : followURL connecting  " + inputURL);

		conn.connect();

		log.debug(threadId + " : followURL connected  " + inputURL);

		int responseCode = conn.getResponseCode();

		log.debug(threadId + " : followURL " + inputURL + ", response code: " + responseCode);

		if ((responseCode >= 300) && (responseCode < 400)) {

			location = conn.getHeaderFields().get("Location").get(0);

			conn.disconnect();

			log.debug(threadId + " : followURL disconnected  " + inputURL);

		}

		if (!location.equals(inputURL.toString())) return new URL(location);

		return inputURL;

	}

	private boolean checkIfExists(final DownloadItem di) {

		try {

			return objectStore.deliverObject(di.getFileName()) != null;

		} catch (final ObjectStoreServiceException e) {

			log.debug(e.getMessage());

			return false;

		}

	}

	/**

	 * Sets the object store.

	 *

	 * @param objectStore

	 *            the objectStore to set

	 */

	public void setObjectStore(final ObjectStore objectStore) {

		this.objectStore = objectStore;

	}

	/**

	 * Sets the queue.

	 *

	 * @param queue

	 *            the queue to set

	 */

	public void setQueue(final BlockingQueue<String> queue) {

		this.queue = queue;

	}

	/**

	 * Sets the protocol.

	 *

	 * @param protocol

	 *            the protocol to set

	 */

	public void setProtocol(final Protocols protocol) {

		this.protocol = protocol;

	}

	/**

	 * Sets the mime type.

	 *

	 * @param mimeType

	 *            the mimeType to set

	 */

	public void setMimeType(final String mimeType) {

		this.mimeType = mimeType;

	}

	public Function<String, DownloadItem> getConverter() {

		return converter;

	}

	public void setConverter(final Function<String, DownloadItem> converter) {

		this.converter = converter;

	}

	public int getReadTimeoutMs() {

		return readTimeoutMs;

	}

	public void setReadTimeoutMs(final int readTimeoutMs) {

		this.readTimeoutMs = readTimeoutMs;

	}

	public int getConnectTimeoutMs() {

		return connectTimeoutMs;

	}

	public void setConnectTimeoutMs(final int connectTimeoutMs) {

		this.connectTimeoutMs = connectTimeoutMs;

	}

	public int getSleepTimeMs() {

		return sleepTimeMs;

	}

	public void setSleepTimeMs(final int sleepTimeMs) {

		this.sleepTimeMs = sleepTimeMs;

	}

}