D-Net

package eu.dnetlib.parthenos.workflows.nodes;

import java.io.IOException;

import java.util.List;

import java.util.Map;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

import com.google.common.collect.Lists;

import com.google.common.collect.Maps;

import com.google.gson.Gson;

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

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

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

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

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

import eu.dnetlib.rmi.common.ResultSet;

import eu.dnetlib.rmi.manager.MSROException;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

import org.apache.http.HttpResponse;

import org.apache.http.NameValuePair;

import org.apache.http.client.HttpClient;

import org.apache.http.client.entity.UrlEncodedFormEntity;

import org.apache.http.client.methods.CloseableHttpResponse;

import org.apache.http.client.methods.HttpPost;

import org.apache.http.impl.client.CloseableHttpClient;

import org.apache.http.impl.client.HttpClients;

import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;

import org.apache.http.message.BasicNameValuePair;

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

/**

 * Created by Alessia Bardi on 09/10/2017.

 *

 * @author Alessia Bardi

 */

public abstract class PublishAbstractJobNode extends AsyncJobNode {

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

	private String inputEprParam;

	@Autowired

	private ResultSetClient resultSetClient;

	private String publisherEndpoint;

	//for parallel requests to the publisher endpoint

	private int nThreads = 5;

	private int nTasks = 150;

	private ExecutorService executorService = Executors.newFixedThreadPool(nThreads);

	private List<Future<Integer>> resList = Lists.newArrayList();

	@Override

	protected String execute(final Env env) throws Exception {

		final ResultSet<?> rsIn = env.getAttribute(getInputEprParam(), ResultSet.class);

		if ((rsIn == null)) { throw new MSROException("InputEprParam (" + getInputEprParam() + ") not found in ENV"); }

		int countAll = 0;

		int countOk = 0;

		int partial = 0;

		Map<Integer, Integer> errors = Maps.newHashMap();

		log.info("Publisher endpoint: " + getPublisherEndpoint());

		PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();

		cm.setMaxTotal(nThreads);

		CloseableHttpClient client = HttpClients.custom().setConnectionManager(cm).build();

		//need to slow down the producer to avoid OOM errors due to many tasks in the queue of the executor

		//see for example here: https://stackoverflow.com/questions/42108351/executorservice-giving-out-of-memory-error

		//let's stop and wait after submission of nLatch tasks

		for (String record : getResultSetClient().iter(rsIn, String.class)) {

			countAll++;

			if(partial == nTasks) {

				log.debug("Waiting for tasks to complete before resubmitting to executor (countAll = "+countAll+") . . . ");

				log.debug("Getting replies");

				long startWait = System.currentTimeMillis();

				for(Future<Integer> res : resList){

					if(res.get() == 200) countOk++;

				}

				resList.clear();

				partial = 0;

				log.debug(". . . Ready to submit again after "+(System.currentTimeMillis() - startWait)+" ms" );

			}

			partial++;

			Future<Integer> res = executorService.submit( () -> {

				CloseableHttpResponse responsePOST = null;

				try {

					HttpPost post = new HttpPost(getPublisherEndpoint());

					List<NameValuePair> params = Lists.newArrayList();

					params.add(new BasicNameValuePair("record", record));

					params.add(new BasicNameValuePair("parthenosTarget", getTarget()));

					UrlEncodedFormEntity ent = new UrlEncodedFormEntity(params, "UTF-8");

					post.setEntity(ent);

					responsePOST = client.execute(post);

					int statusCode = responsePOST.getStatusLine().getStatusCode();

					switch (statusCode) {

						case 200:

							return statusCode;

						default:

							log.error(responsePOST.getStatusLine().getStatusCode() + ": " + responsePOST.getStatusLine().getReasonPhrase());

							log.debug("Source record causing error: " + record);

							errors.merge(statusCode, 1, Integer::sum);

							return statusCode;

					}

				} catch (Exception e) {

					log.error(e.getMessage());

					errors.merge(-1, 1, Integer::sum);

					return -1;

				}

				finally{

					if(responsePOST != null) responsePOST.close();

				}

			});

			resList.add(res);

		}

		executorService.shutdown();

		//now let's wait for the results. We can block ourselves here: we have nothing else to do

		log.info("Waiting for responses");

		for(Future<Integer> res : resList){

			if(res.get() == 200) countOk++;

		}

		client.close();

		cm.shutdown();

		log.info(String.format("Got all responses. Ok %s/%s", countOk, countAll));

		env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "countOk", countOk);

		env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "countAll", countAll);

		env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "errorsMap", new Gson().toJson(errors));

		log.info("publishing completed");

		if (!errors.isEmpty()) {

			log.warn("Problems in publishing on "+getTarget()+": "+countOk+"/"+countAll+" see error maps for details");

		}

		if(countAll == 0) log.warn("0 resources to publish");

		return Arc.DEFAULT_ARC;

	}

	public abstract String getTarget();

	public String getInputEprParam() {

		return this.inputEprParam;

	}

	public void setInputEprParam(final String inputEprParam) {

		this.inputEprParam = inputEprParam;

	}

	public String getPublisherEndpoint() {

		return publisherEndpoint;

	}

	public void setPublisherEndpoint(final String publisherEndpoint) {

		this.publisherEndpoint = publisherEndpoint;

	}

	public ResultSetClient getResultSetClient() {

		return resultSetClient;

	}

	public void setResultSetClient(final ResultSetClient resultSetClient) {

		this.resultSetClient = resultSetClient;

	}

}