D-Net

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

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

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.procs.Token;

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

import eu.dnetlib.rmi.data.ObjectStoreFile;

/**

 * 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 = this.objectStoreDao.getObjectStore(this.nativeObjectStore);

		this.total = objectStore.getSize();

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

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

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

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

		final ExecutorService executor = Executors.newFixedThreadPool(this.numberOfThreads);

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

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

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

        }

		this.counter = 0;

		while (deliver.hasNext()) {

			this.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;

	}

	@Override

	protected void beforeStart(final Token token) {

		token.setProgressProvider(this);

	}

	public String getNativeObjectStore() {

		return this.nativeObjectStore;

	}

	public void setNativeObjectStore(final String nativeObjectStore) {

		this.nativeObjectStore = nativeObjectStore;

	}

	public int getNumberOfThreads() {

		return this.numberOfThreads;

	}

	public void setNumberOfThreads(final int numberOfThreads) {

		this.numberOfThreads = numberOfThreads;

	}

	public String getStoreThumb250() {

		return this.storeThumb250;

	}

	public void setStoreThumb250(final String storeThumb250) {

		this.storeThumb250 = storeThumb250;

	}

	public String getStoreThumb96() {

		return this.storeThumb96;

	}

	public void setStoreThumb96(final String storeThumb96) {

		this.storeThumb96 = storeThumb96;

	}

	@Override

	public String getProgressDescription() {

		return this.counter < 0 ? "-" : String.format("%d / %d", this.counter, this.total);

	}

}