347 lines
14 KiB
Java
347 lines
14 KiB
Java
package org.gcube.event.publisher;
|
|
|
|
import java.io.BufferedReader;
|
|
import java.io.IOException;
|
|
import java.io.InputStreamReader;
|
|
import java.io.OutputStream;
|
|
import java.net.HttpURLConnection;
|
|
import java.net.MalformedURLException;
|
|
import java.net.URL;
|
|
import java.util.Arrays;
|
|
|
|
import org.gcube.oidc.rest.JWTToken;
|
|
import org.gcube.oidc.rest.OpenIdConnectRESTHelperException;
|
|
import org.json.simple.JSONObject;
|
|
import org.json.simple.parser.JSONParser;
|
|
import org.json.simple.parser.ParseException;
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
public abstract class AbstractHTTPWithJWTTokenAuthEventSender implements EventSender {
|
|
|
|
protected static final Logger log = LoggerFactory.getLogger(HTTPWithOIDCAuthEventSender.class);
|
|
protected URL baseEndpointURL;
|
|
protected String clientId;
|
|
protected String clientSecret;
|
|
protected URL tokenURL;
|
|
protected int connectionTimeout;
|
|
protected int readTimeout;
|
|
protected HTTPPost lastHTTPPost;
|
|
protected HTTPGet lastHTTPGet;
|
|
|
|
public AbstractHTTPWithJWTTokenAuthEventSender(URL baseEndpointURL, String clientId, String clientSecret,
|
|
URL tokenURL) {
|
|
|
|
super();
|
|
this.baseEndpointURL = baseEndpointURL;
|
|
this.clientId = clientId;
|
|
this.clientSecret = clientSecret;
|
|
this.tokenURL = tokenURL;
|
|
this.connectionTimeout = getDefaultConnectionTimeout();
|
|
this.readTimeout = getDefaultReadTimeout();
|
|
}
|
|
|
|
protected int getDefaultReadTimeout() {
|
|
return HTTPVerb.DEFAULT_READ_TIMEOUT;
|
|
}
|
|
|
|
protected int getDefaultConnectionTimeout() {
|
|
return HTTPVerb.DEFAULT_CONNECTION_TIMEOUT;
|
|
}
|
|
|
|
public int getReadTimeout() {
|
|
return readTimeout;
|
|
}
|
|
|
|
public void setReadTimeout(int readTimeout) {
|
|
this.readTimeout = readTimeout;
|
|
}
|
|
|
|
public int getConnectionTimeout() {
|
|
return connectionTimeout;
|
|
}
|
|
|
|
public void setConnectionTimeout(int connectionTimeout) {
|
|
this.connectionTimeout = connectionTimeout;
|
|
}
|
|
|
|
@Override
|
|
public void send(Event event) {
|
|
log.debug("Starting HTTP POST thread with base URL: {}", baseEndpointURL);
|
|
lastHTTPPost = new HTTPPost(baseEndpointURL, event);
|
|
new Thread(lastHTTPPost).start();
|
|
}
|
|
|
|
@Override
|
|
public int getLastSendHTTPResponseCode() {
|
|
return lastHTTPPost != null ? lastHTTPPost.gethttpResponseCode() : -1;
|
|
}
|
|
|
|
@Override
|
|
public String sendAndGetResult(Event event) {
|
|
log.debug("Starting HTTP POST thread to: {}", baseEndpointURL);
|
|
lastHTTPPost = new HTTPPost(baseEndpointURL, event);
|
|
Thread postThread = new Thread(lastHTTPPost);
|
|
postThread.start();
|
|
try {
|
|
postThread.join();
|
|
return lastHTTPPost.getResult();
|
|
} catch (InterruptedException e) {
|
|
log.error("While waiting for HTTP Post thread termination", e);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
@Override
|
|
public JSONObject retrive(String id) {
|
|
lastHTTPGet = new HTTPGet(baseEndpointURL, id);
|
|
return lastHTTPGet.readJSON();
|
|
}
|
|
|
|
@Override
|
|
public int getLastRetrieveHTTPResponseCode() {
|
|
return lastHTTPGet != null ? lastHTTPGet.gethttpResponseCode() : -1;
|
|
}
|
|
|
|
protected URL getTokenURL() {
|
|
return tokenURL;
|
|
}
|
|
|
|
protected abstract JWTToken getAuthorizationToken() throws OpenIdConnectRESTHelperException;
|
|
|
|
public abstract class HTTPVerb {
|
|
|
|
protected static final int DEFAULT_CONNECTION_TIMEOUT = 10000;
|
|
protected static final int DEFAULT_READ_TIMEOUT = 5000;
|
|
|
|
protected URL baseEndpoint;
|
|
protected int httpResponseCode = -1;
|
|
protected String httpContentType;
|
|
|
|
public HTTPVerb(URL baseEndpoint) {
|
|
this.baseEndpoint = baseEndpoint;
|
|
}
|
|
|
|
public int gethttpResponseCode() {
|
|
return httpResponseCode;
|
|
}
|
|
|
|
}
|
|
|
|
public class HTTPPost extends HTTPVerb implements Runnable {
|
|
|
|
private static final int PAUSE_INCREMENT_FACTOR = 2;
|
|
private static final long MAX_RETRYINGS = 2;
|
|
|
|
private final int[] RETRY_CODES = { HttpURLConnection.HTTP_BAD_GATEWAY,
|
|
HttpURLConnection.HTTP_CLIENT_TIMEOUT, HttpURLConnection.HTTP_GATEWAY_TIMEOUT,
|
|
HttpURLConnection.HTTP_INTERNAL_ERROR };
|
|
|
|
private Event event;
|
|
private String result;
|
|
private long actualPause = 1;
|
|
private long retryings = 0;
|
|
|
|
public HTTPPost(URL baseEndpoint, Event event) {
|
|
super(baseEndpoint);
|
|
this.event = event;
|
|
}
|
|
|
|
@Override
|
|
public void run() {
|
|
try {
|
|
URL eventEndpoint = null;
|
|
if (baseEndpoint.toString().endsWith("/")) {
|
|
try {
|
|
log.debug("Removing trailing slash from base URL since the endpoint computation API changed");
|
|
eventEndpoint = new URL(baseEndpoint.toString().substring(0, baseEndpoint.toString().length() - 1));
|
|
} catch (MalformedURLException e) {
|
|
log.error("Cannot remove trailing slash from base endpoint URL", e);
|
|
return;
|
|
}
|
|
} else {
|
|
eventEndpoint = baseEndpoint;
|
|
}
|
|
boolean OK = false;
|
|
do {
|
|
try {
|
|
log.debug("Getting auth token for client '{}' if needed", clientId);
|
|
JWTToken token = getAuthorizationToken();
|
|
log.debug("Performing HTTP POST to: {}", eventEndpoint);
|
|
HttpURLConnection connection = (HttpURLConnection) eventEndpoint.openConnection();
|
|
connection.setRequestMethod("POST");
|
|
connection.setConnectTimeout(connectionTimeout);
|
|
log.trace("HTTP connection timeout set to: {}", connection.getConnectTimeout());
|
|
connection.setReadTimeout(readTimeout);
|
|
log.trace("HTTP connection Read timeout set to: {}", connection.getReadTimeout());
|
|
connection.setRequestProperty("Content-Type", "application/json");
|
|
connection.setRequestProperty("Accept", "text/plain");
|
|
connection.setDoOutput(true);
|
|
if (token != null) {
|
|
log.debug("Setting authorization header as: {}", token.getAccessTokenAsBearer());
|
|
connection.setRequestProperty("Authorization", token.getAccessTokenAsBearer());
|
|
} else {
|
|
log.debug("Sending request without authorization header");
|
|
}
|
|
OutputStream os = connection.getOutputStream();
|
|
String jsonString = event.toJSONString();
|
|
log.trace("Sending event JSON: {}", jsonString);
|
|
os.write(jsonString.getBytes("UTF-8"));
|
|
os.flush();
|
|
os.close();
|
|
|
|
StringBuilder sb = new StringBuilder();
|
|
httpResponseCode = connection.getResponseCode();
|
|
log.trace("HTTP Response code: {}", httpResponseCode);
|
|
httpContentType = connection.getContentType();
|
|
log.trace("HTTP Response content type is: {}", httpContentType);
|
|
|
|
log.trace("Reading response");
|
|
InputStreamReader isr = null;
|
|
if (httpResponseCode == HttpURLConnection.HTTP_OK) {
|
|
OK = true;
|
|
// From this point ahead every exception should not set OK to false
|
|
// since the server acknowledged with 200
|
|
isr = new InputStreamReader(connection.getInputStream(), "UTF-8");
|
|
} else {
|
|
isr = new InputStreamReader(connection.getErrorStream(), "UTF-8");
|
|
}
|
|
BufferedReader br = new BufferedReader(isr);
|
|
String line = null;
|
|
while ((line = br.readLine()) != null) {
|
|
sb.append("\n" + line);
|
|
}
|
|
br.close();
|
|
isr.close();
|
|
sb.deleteCharAt(0);
|
|
result = sb.toString();
|
|
if (OK) {
|
|
log.debug("[{}] Event publish for {} is OK", httpResponseCode, event.getName());
|
|
} else {
|
|
log.trace("Response message from server:\n{}", result);
|
|
if (shouldRetryWithCode(httpResponseCode)) {
|
|
if (retryings <= MAX_RETRYINGS) {
|
|
log.warn("[{}] Event publish ERROR, retrying in {} seconds", httpResponseCode,
|
|
actualPause);
|
|
|
|
Thread.sleep(actualPause * 1000);
|
|
log.debug("Start retrying event publish: {}", event.getName());
|
|
actualPause *= PAUSE_INCREMENT_FACTOR;
|
|
retryings += 1;
|
|
} else {
|
|
log.error("[{}] Event publish ERROR, exhausted tries after {} retryings",
|
|
httpResponseCode,
|
|
retryings);
|
|
|
|
break;
|
|
}
|
|
} else {
|
|
log.info("[{}] Event publish ERROR but should not retry with this HTTP code",
|
|
httpResponseCode);
|
|
|
|
break;
|
|
}
|
|
}
|
|
} catch (IOException | OpenIdConnectRESTHelperException e) {
|
|
log.error("POSTing JSON to: " + eventEndpoint, e);
|
|
break;
|
|
}
|
|
} while (!OK);
|
|
} catch (InterruptedException e) {
|
|
log.error("Sleeping before retry event send", e);
|
|
}
|
|
}
|
|
|
|
private boolean shouldRetryWithCode(int httpResultCode) {
|
|
return Arrays.binarySearch(RETRY_CODES, httpResultCode) > 0;
|
|
}
|
|
|
|
public String getResult() {
|
|
return result;
|
|
}
|
|
|
|
}
|
|
|
|
public class HTTPGet extends HTTPVerb {
|
|
|
|
private String id;
|
|
|
|
public HTTPGet(URL baseEndpoint, String id) {
|
|
super(baseEndpoint);
|
|
this.id = id;
|
|
}
|
|
|
|
public JSONObject readJSON() {
|
|
URL endpoint = null;
|
|
JSONObject results = null;
|
|
try {
|
|
String baseEndpointString = baseEndpoint.toString();
|
|
if (baseEndpointString.endsWith("/")) {
|
|
endpoint = new URL(baseEndpointString + id);
|
|
} else {
|
|
endpoint = new URL(baseEndpointString + "/" + id);
|
|
}
|
|
} catch (MalformedURLException e) {
|
|
log.error("Cannot compute retrieve endpoint URL. ID: " + id + ", base endpoint: " + baseEndpointURL, e);
|
|
return null;
|
|
}
|
|
try {
|
|
log.debug("Getting auth token for client '{}' if needed", clientId);
|
|
JWTToken token = getAuthorizationToken();
|
|
|
|
log.debug("Performing HTTP GET to: {}", endpoint);
|
|
HttpURLConnection connection = (HttpURLConnection) endpoint.openConnection();
|
|
connection.setRequestMethod("GET");
|
|
connection.setConnectTimeout(connectionTimeout);
|
|
log.trace("HTTP connection timeout set to: {}", connection.getConnectTimeout());
|
|
connection.setReadTimeout(readTimeout);
|
|
log.trace("HTTP connection Read timeout set to: {}", connection.getReadTimeout());
|
|
if (token != null) {
|
|
log.debug("Setting authorization header as: {}", token.getAccessTokenAsBearer());
|
|
connection.setRequestProperty("Authorization", token.getAccessTokenAsBearer());
|
|
} else {
|
|
log.debug("Sending request without authorization header");
|
|
}
|
|
|
|
StringBuilder sb = new StringBuilder();
|
|
httpResponseCode = connection.getResponseCode();
|
|
log.trace("HTTP Response code: {}", httpResponseCode);
|
|
httpContentType = connection.getContentType();
|
|
log.trace("HTTP Response content type is: {}", httpContentType);
|
|
|
|
log.trace("Reading response");
|
|
InputStreamReader isr = null;
|
|
if (httpResponseCode == HttpURLConnection.HTTP_OK) {
|
|
isr = new InputStreamReader(connection.getInputStream(), "UTF-8");
|
|
} else {
|
|
isr = new InputStreamReader(connection.getErrorStream(), "UTF-8");
|
|
}
|
|
BufferedReader br = new BufferedReader(isr);
|
|
String line = null;
|
|
while ((line = br.readLine()) != null) {
|
|
sb.append(line + "\n");
|
|
}
|
|
br.close();
|
|
isr.close();
|
|
if (httpResponseCode == HttpURLConnection.HTTP_OK) {
|
|
log.debug("[{}] Got results for {}", httpResponseCode, id);
|
|
if (httpContentType.equals("application/json")) {
|
|
try {
|
|
results = (JSONObject) new JSONParser().parse(sb.toString());
|
|
} catch (ParseException e) {
|
|
log.warn("Error parsing results string as JSON: {}", sb.toString());
|
|
}
|
|
} else {
|
|
log.warn("Received a non JSON reponse: {}", sb.toString());
|
|
}
|
|
} else {
|
|
log.warn("[{}] Error getting results for ID {}", httpResponseCode, id);
|
|
}
|
|
} catch (IOException | OpenIdConnectRESTHelperException e) {
|
|
log.error("Getting results from: " + endpoint, e);
|
|
}
|
|
|
|
return results;
|
|
}
|
|
}
|
|
} |