package eu.dnetlib.ariadneplus.workflows.nodes; import com.google.common.collect.Lists; 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.manager.MSROException; import org.apache.commons.io.IOUtils; 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.HttpGet; import org.apache.http.client.methods.HttpPost; import org.apache.http.client.utils.URIBuilder; import org.apache.http.config.SocketConfig; import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.HttpClientBuilder; import org.apache.http.impl.client.HttpClients; import org.apache.http.message.BasicNameValuePair; import org.apache.http.util.EntityUtils; import java.io.IOException; import java.io.InputStream; import java.net.ConnectException; import java.net.URI; import java.util.List; public class IndexOnESJobNode extends AsyncJobNode { private static final Log log = LogFactory.getLog(IndexOnESJobNode.class); private String publisherEndpoint; private String datasourceInterface; private String datasource; @Override protected String execute(final Env env) throws Exception { int statusCode = -1; String response = ""; // String indexOnESResult = "noResult"; // log.info("Publisher endpoint: " + getPublisherEndpoint()); // SocketConfig socketConfig = SocketConfig.custom().setSoKeepAlive(true).build(); // CloseableHttpClient client = HttpClientBuilder.create() // .setDefaultSocketConfig(socketConfig).build(); log.info("IndexOnES endpoint: " + getIndexOnESEndpoint()); // CloseableHttpResponse responsePOST = null; HttpClient client = null; try { // HttpPost post = new HttpPost(getIndexOnESEndpoint()); // List params = Lists.newArrayList(); String[] splits = getDatasourceInterface().split("::"); String datasource = splits[2]; String collectionId = splits[3]; // params.add(new BasicNameValuePair("datasource", datasource)); // params.add(new BasicNameValuePair("collectionId", collectionId)); // UrlEncodedFormEntity ent = new UrlEncodedFormEntity(params, "UTF-8"); // post.setEntity(ent); // log.info("Calling IndexOnES endpoint with params: "+getDatasource()+" "+getDatasourceInterface()); // responsePOST = client.execute(post); URI getURI = new URIBuilder(getIndexOnESEndpoint()) .addParameter("datasource", datasource) .addParameter("collectionId", collectionId) .build(); client = HttpClients.createDefault(); HttpResponse res = client.execute(new HttpGet(getURI)); response = EntityUtils.toString(res.getEntity()); if (res.getStatusLine()!=null) { statusCode = res.getStatusLine().getStatusCode(); } // statusCode = responsePOST.getStatusLine().getStatusCode(); // try(InputStream responseBody = responsePOST.getEntity().getContent()) { // indexOnESResult = IOUtils.toString(responseBody, "UTF-8"); // } catch (Exception e) { // log.error(e); // } // switch (statusCode) { // case 200: // log.info("index on ES completed"); // break; // default: // log.error("error indexing on ES " + responsePOST.getStatusLine().getStatusCode() + ": " + responsePOST.getStatusLine().getReasonPhrase()); // break; // } } // catch (ConnectException ce) { // log.error(ce); // throw new MSROException("Unable to connect to Publisher endpoint" + getIndexOnESEndpoint()); // } // catch (IOException e) { // log.error(e); // throw new MSROException("IO Error" + getIndexOnESEndpoint()); // } catch (Throwable t) { log.error(t); throw new MSROException("Indexing on Elastic Search: " + t.getMessage()); } finally{ // if(responsePOST != null) responsePOST.close(); // client.close(); // cm.shutdown(); } env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "statusCode", Integer.toString(statusCode)); // env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "indexResult", indexOnESResult); env.setAttribute(WorkflowsConstants.MAIN_LOG_PREFIX + "response", response); if (statusCode!=200) { throw new MSROException("Error from Publisher endpoint [ status code: " + statusCode + " ]"); } return Arc.DEFAULT_ARC; } public String getPublisherEndpoint() { return publisherEndpoint; } private String getIndexOnESEndpoint() { return publisherEndpoint.concat("/indexOnES"); } public void setPublisherEndpoint(final String publisherEndpoint) { this.publisherEndpoint = publisherEndpoint; } public String getDatasourceInterface() { return datasourceInterface; } public void setDatasourceInterface(String datasourceInterface) { this.datasourceInterface = datasourceInterface; } public String getDatasource() { return datasource; } public void setDatasource(String datasource) { this.datasource = datasource; } }