D-Net
/
AriadnePlus
4
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

set KeepAlive true on used socket; replaced SimpleJobNode with AsyncJobNode

This commit is contained in:
Enrico Ottonello 2020-10-13 15:34:43 +02:00
parent 263c1beb7e
commit 25de647504
1 changed files with 9 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
dnet-ariadneplus/src/main/java/eu/dnetlib/ariadneplus/workflows/nodes/IndexOnESJobNode.java
View File

@ -1,32 +1,22 @@
package eu.dnetlib.ariadneplus.workflows.nodes; package eu.dnetlib.ariadneplus.workflows.nodes;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import com.google.gson.Gson;
import eu.dnetlib.enabling.resultset.ResultSetInfo;
import eu.dnetlib.enabling.resultset.client.ResultSetClient;
import eu.dnetlib.msro.workflows.graph.Arc; import eu.dnetlib.msro.workflows.graph.Arc;
import eu.dnetlib.msro.workflows.nodes.AsyncJobNode; import eu.dnetlib.msro.workflows.nodes.AsyncJobNode;
import eu.dnetlib.msro.workflows.nodes.SimpleJobNode;
import eu.dnetlib.msro.workflows.procs.Env; import eu.dnetlib.msro.workflows.procs.Env;
import eu.dnetlib.msro.workflows.procs.Token;
import eu.dnetlib.msro.workflows.util.ResultsetProgressProvider;
import eu.dnetlib.msro.workflows.util.WorkflowsConstants; import eu.dnetlib.msro.workflows.util.WorkflowsConstants;
import eu.dnetlib.rmi.common.ResultSet;
import eu.dnetlib.rmi.manager.MSROException; import eu.dnetlib.rmi.manager.MSROException;
import org.apache.commons.io.IOUtils; import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import org.apache.http.NameValuePair; import org.apache.http.NameValuePair;
import org.apache.http.client.HttpClient;
import org.apache.http.client.entity.UrlEncodedFormEntity; import org.apache.http.client.entity.UrlEncodedFormEntity;
import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPost; import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity; import org.apache.http.config.SocketConfig;
import org.apache.http.impl.client.CloseableHttpClient; import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients; import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.message.BasicNameValuePair; import org.apache.http.message.BasicNameValuePair;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
@ -34,7 +24,7 @@ import java.net.ConnectException;
import java.util.List; import java.util.List;
public class IndexOnESJobNode extends SimpleJobNode { public class IndexOnESJobNode extends AsyncJobNode {
private static final Log log = LogFactory.getLog(IndexOnESJobNode.class); private static final Log log = LogFactory.getLog(IndexOnESJobNode.class);
@ -42,20 +32,15 @@ public class IndexOnESJobNode extends SimpleJobNode {
private String datasourceInterface; private String datasourceInterface;
private String datasource; private String datasource;
//for parallel requests to the publisher endpoint
// private int nThreads = 5;
@Override @Override
protected String execute(final Env env) throws Exception { protected String execute(final Env env) throws Exception {
int statusCode = -1; int statusCode = -1;
String indexOnESResult = "noResult"; String indexOnESResult = "noResult";
log.info("Publisher endpoint: " + getPublisherEndpoint()); log.info("Publisher endpoint: " + getPublisherEndpoint());
// PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(); SocketConfig socketConfig = SocketConfig.custom().setSoKeepAlive(true).build();
// cm.setMaxTotal(nThreads); CloseableHttpClient client = HttpClientBuilder.create()
// CloseableHttpClient client = HttpClients.custom().setConnectionManager(cm).build(); .setDefaultSocketConfig(socketConfig).build();
CloseableHttpClient client = HttpClients.createDefault();
log.info("IndexOnES endpoint: " + getIndexOnESEndpoint()); log.info("IndexOnES endpoint: " + getIndexOnESEndpoint());
CloseableHttpResponse responsePOST = null; CloseableHttpResponse responsePOST = null;
try { try {
@ -73,7 +58,9 @@ public class IndexOnESJobNode extends SimpleJobNode {
statusCode = responsePOST.getStatusLine().getStatusCode(); statusCode = responsePOST.getStatusLine().getStatusCode();
try(InputStream responseBody = responsePOST.getEntity().getContent()) { try(InputStream responseBody = responsePOST.getEntity().getContent()) {
indexOnESResult = IOUtils.toString(responseBody, "UTF-8"); indexOnESResult = IOUtils.toString(responseBody, "UTF-8");
} catch (Exception e) {} } catch (Exception e) {
log.error(e);
}
switch (statusCode) { switch (statusCode) {
case 200: case 200:
log.info("index on ES completed"); log.info("index on ES completed");