D-Net

package eu.dnetlib.parthenos.workflows.nodes;

import java.util.List;

import java.util.Map;

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

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

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

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 SimpleJobNode{

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

	private String inputEprParam;

	@Autowired

	private ResultSetClient resultSetClient;

	private String publisherEndpoint;

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

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

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

		//let's start sequentially

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

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

			HttpClient client = HttpClients.createDefault();

			HttpResponse responsePOST = client.execute(post);

			countAll++;

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

			switch (statusCode) {

			case 200:

				countOk++;

				break;

			default:

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

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

			}

		}

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

		if (!errors.isEmpty()) {

			throw new MSROException("Problems in publishing on "+getTarget());

		} else 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;

	}

}