D-Net

		final Function<String, DownloadItem> applyDowunloadPlugin = input -> {

			if (input == null) {

				log.error("Input is null");

				return null;

			}

			if (input.equals(DownloadIntoObjectStoreAction.END_QUEUE_STRING)) return DownloadIntoObjectStoreAction.END_QUEUE;

			DownloadItem di = null;

			try {

				di = DownloadItem.newObjectfromJSON(input);

				if (downloadPlugin.retrieveUrl(di) == null) {

					di.setUrl(null);

					di.setOriginalUrl(null);

				return di;

			} catch (Throwable e) {

				reportException(pluginReport, di, e);

				log.error("Exception on transform item :" + input, e);

				return null;

			}

		};

		for (int i = 0; i < numberOfThreads; i++) {

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

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.*;

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

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

import eu.dnetlib.enabling.resultset.client.ResultSetClient;

import eu.dnetlib.enabling.resultset.listener.ResultSetListener;

import eu.dnetlib.msro.workflows.graph.Arc;

import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;

import eu.dnetlib.msro.workflows.procs.Env;

import eu.dnetlib.msro.workflows.util.ProgressProvider;

import eu.dnetlib.rmi.data.ObjectStoreFile;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

/**

 * Created by sandro on 4/8/16.

 */

public class GenerateThumbnailJobNode extends SimpleJobNode implements ProgressProvider {

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

	public static ObjectStoreFile END_QUEUE = new ObjectStoreFile();

	@Autowired

	ResultSetClient resultSetClient;

	private String nativeObjectStore;

	private String storeThumb250;

	private String storeThumb96;

	private int numberOfThreads = 4;

	@Autowired

	private ObjectStoreDao objectStoreDao;

	private int counter = -1;

	private int total = -1;

	@Override

	protected String execute(final Env env) throws Exception {

		final ObjectStore objectStore = objectStoreDao.getObjectStore(nativeObjectStore);

		total = objectStore.getSize();

		final ObjectStore objectStore250 = objectStoreDao.getObjectStore(env.getAttribute(storeThumb250, String.class));

		final ObjectStore objectStore96 = objectStoreDao.getObjectStore(env.getAttribute(storeThumb96, String.class));

		final ResultSetListener<ObjectStoreFile> deliver = objectStore.deliver(0L, System.currentTimeMillis());

		final BlockingQueue<ObjectStoreFile> queue = new ArrayBlockingQueue<>(10);

		final ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);

		final List<Future<Boolean>> responses = new ArrayList<>();

		responses.add(executor.submit(new WorkerMap(objectStore, objectStore250, objectStore96, queue)));

		counter = 0;

		while (deliver.hasNext()) {

			counter++;

			queue.put(deliver.next());

		}

		queue.put(END_QUEUE);

		for (final Future<Boolean> currentResponse : responses) {

			if (currentResponse.get() == false) {

				log.error("Some response fail");

			}

		}

		return Arc.DEFAULT_ARC;

	}

	public String getNativeObjectStore() {

		return nativeObjectStore;

	}

	public void setNativeObjectStore(final String nativeObjectStore) {

		this.nativeObjectStore = nativeObjectStore;

	}

	public int getNumberOfThreads() {

		return numberOfThreads;

	}

	public void setNumberOfThreads(final int numberOfThreads) {

		this.numberOfThreads = numberOfThreads;

	}

	public String getStoreThumb250() {

		return storeThumb250;

	}

	public void setStoreThumb250(final String storeThumb250) {

		this.storeThumb250 = storeThumb250;

	}

	public String getStoreThumb96() {

		return storeThumb96;

	}

	public void setStoreThumb96(final String storeThumb96) {

		this.storeThumb96 = storeThumb96;

	}

	@Override

	public String asString() {

		if (counter < 0) {

			return "-";

		}

		return String.format("%d / %d", counter, total);

	}

}

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);

				}

				currentFile = queue.take();

			} catch (Throwable e) {

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

			}

		}

		if (currentFile == GenerateThumbnailJobNode.END_QUEUE)

			queue.put(currentFile);

		return true;

	}

}

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

import java.io.File;

import java.io.FileNotFoundException;

import java.io.IOException;

import java.io.InputStream;

import java.nio.file.FileSystems;

import java.nio.file.Files;

import java.nio.file.Path;

import java.nio.file.StandardCopyOption;

import java.util.UUID;

import org.apache.tika.Tika;

import org.im4java.core.ConvertCmd;

import org.im4java.core.IMOperation;

/**

 * Created by sandro on 4/8/16.

 */

public class ThumbnailGenerator {

	private Path baseDirPath;

	private String dirName;

	public ThumbnailGenerator() throws IOException {

		dirName = UUID.randomUUID().toString();

		baseDirPath = Files.createTempDirectory(dirName);

	}

	public static boolean deleteRecursive(File path) throws FileNotFoundException {

		if (!path.exists()) throw new FileNotFoundException(path.getAbsolutePath());

		boolean ret = true;

		if (path.isDirectory()) {

			for (File f : path.listFiles()) {

				ret = ret && deleteRecursive(f);

			}

		}

		return ret && path.delete();

	}

	public InputStream convert(final InputStream input, int width, int height, final String name, boolean deleteBeforeConver) {

		try {

			destroyFolder();

			baseDirPath = Files.createTempDirectory(dirName);

			final Path tempFile = baseDirPath.resolve(name + "_orig");

			Files.copy(input, tempFile, StandardCopyOption.REPLACE_EXISTING);

			final String inputType = guessMimeType(tempFile.toAbsolutePath().toString());

			String inputPath = tempFile.toAbsolutePath().toString();

			if (inputType != null && inputType.contains("pdf")) {

				inputPath += "[0]";

			}

			ConvertCmd cmd = new ConvertCmd();

			IMOperation op = new IMOperation();

			op.addImage(inputPath);

			op.define(String.format("jpeg:size=%dx%d", width * 2, height * 2));

			op.thumbnail(width, height);

			op.unsharp(0.5);

			op.background("transparent");

			op.gravity("center");

			op.extent(width, height);

			op.colors(256);

			op.addImage(tempFile.toAbsolutePath().toString().replace("orig", "conv.png"));

			cmd.run(op);

			final Path path = FileSystems.getDefault().getPath(tempFile.toAbsolutePath().toString().replace("orig", "conv.png"));

			return Files.newInputStream(path);

		} catch (Throwable e) {

			System.out.println("Error on convert file" + e.toString());

			return null;

		}

	}

	private String guessMimeType(final String inputPath) throws IOException {

		Tika tika = new Tika();

		return tika.detect(new File(inputPath));

	}

	public boolean destroyFolder() throws FileNotFoundException {

		return deleteRecursive(baseDirPath.toFile());

	}

}

	<bean id="wfNodeGenerateThumbnail"

	      class="eu.dnetlib.efg.workflows.nodes.thumbnail.GenerateThumbnailJobNode"

	      scope="prototype"/>

						<ARC to="generateThumbnail"/>

					</ARCS>

				</NODE>

				<NODE name="generateThumbnail" type="GenerateThumbnail">

					<DESCRIPTION>Generate Thumbnail for EFG</DESCRIPTION>

					<PARAMETERS>

						<PARAM name="nativeObjectStore" ref="nativeObjectStoreId"/>

						<PARAM name="storeThumb250" value="thumb256_id"/>

						<PARAM name="storeThumb96" value="thumb96_id"/>

					</PARAMETERS>

					<ARCS>

			<groupId>org.apache.commons</groupId>

			<artifactId>commons-compress</artifactId>

			<version>1.11</version>

		</dependency>

		<dependency>

		<dependency>

			<groupId>org.im4java</groupId>

			<artifactId>im4java</artifactId>

			<version>1.4.0</version>

		</dependency>

		<dependency>

			<groupId>org.apache.tika</groupId>

			<artifactId>tika-core</artifactId>

			<version>1.12</version>

		</dependency>