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;

public class PublishVirtuosoJobNode extends SimpleJobNode {

	private static final Log log = LogFactory.getLog(PublishVirtuosoJobNode.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", "VIRTUOSO"));

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

			default:

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

		log.debug("countOK: "+countOk);

		return Arc.DEFAULT_ARC;

	}

	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;

	}

}