D-Net

package eu.dnetlib.efg.workflows.nodes.thumbnail;

import java.io.IOException;

import java.io.InputStream;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.Callable;

import eu.dnetlib.data.objectstore.ObjectStoreRecord;

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

import eu.dnetlib.rmi.data.ObjectStoreFile;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

/**

 * Created by sandro on 4/8/16.

 */

public class WorkerMap implements Callable<Boolean> {

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

	private final ObjectStore thumb256;

	private final ObjectStore thumb96;

	private final ObjectStore nativeObjectStore;

	private final BlockingQueue<ObjectStoreFile> queue;

	private final ThumbnailGenerator thumbnailGenerator;

	public WorkerMap(final ObjectStore nativeObjectStore, final ObjectStore thumb256, final ObjectStore thumb96, final BlockingQueue<ObjectStoreFile> queue)

			throws IOException {

		this.thumb96 = thumb96;

		this.thumb256 = thumb256;

		this.queue = queue;

		this.nativeObjectStore = nativeObjectStore;

		this.thumbnailGenerator = new ThumbnailGenerator();

	}

	@Override

	public Boolean call() throws Exception {

		ObjectStoreFile currentFile = queue.take();

		while (currentFile != GenerateThumbnailJobNode.END_QUEUE) {

			try {

				final ObjectStoreRecord record = new ObjectStoreRecord();

				if (!thumb96.alreadyExist(currentFile.getObjectID())) {

					final InputStream convert = thumbnailGenerator.convert(nativeObjectStore.deliverStream(currentFile.getObjectID()), 150, 96, "96", true);

					record.setFileMetadata(currentFile);

					record.setInputStream(convert);

					thumb96.feedObjectRecord(record);

				}

				if (!thumb256.alreadyExist(currentFile.getObjectID())) {

					record.setInputStream(thumbnailGenerator.convert(nativeObjectStore.deliverStream(currentFile.getObjectID()), 600, 250, "250", false));

					thumb256.feedObjectRecord(record);

				}

			} catch (Throwable e) {

				log.error("Error on generating thumbnail", e);

			} finally {

                currentFile = queue.take();

            }

		}

		if (currentFile == GenerateThumbnailJobNode.END_QUEUE)

			queue.put(currentFile);

		return true;

	}

}