gCubeSystem
/
event-publisher-library
16
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Event name now is the end part of the endpoint URL, removed Accept header to avoid Conductor issue and client identy (to construct authorization header) now is optional

This commit is contained in:
Mauro Mugnaini 2020-06-29 17:00:57 +02:00
parent 6b7326636e
commit 86afd4c162
1 changed files with 29 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/main/java/org/gcube/event/publisher/HTTPEventSender.java
View File

@ -16,13 +16,13 @@ public class HTTPEventSender implements EventSender {
protected static final Logger log = LoggerFactory.getLogger(HTTPEventSender.class); protected static final Logger log = LoggerFactory.getLogger(HTTPEventSender.class);
private URL endpoint; private URL baseEnndpointURL;
private String clientId; private String clientId;
private String clientSecret; private String clientSecret;
private URL tokenURL; private URL tokenURL;
public HTTPEventSender(URL endpointURL, String clientId, String clientSecret, URL tokenURL) { public HTTPEventSender(URL baseEnndpointURL, String clientId, String clientSecret, URL tokenURL) {
this.endpoint = endpointURL; this.baseEnndpointURL = baseEnndpointURL;
this.clientId = clientId; this.clientId = clientId;
this.clientSecret = clientSecret; this.clientSecret = clientSecret;
this.tokenURL = tokenURL; this.tokenURL = tokenURL;
@ -31,14 +31,14 @@ public class HTTPEventSender implements EventSender {
@Override @Override
public void send(Event event) { public void send(Event event) {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("Starting HTTP POST thread to: " + endpoint); log.debug("Starting HTTP POST thread to: " + baseEnndpointURL);
} }
try { try {
URL eventEndpoint = new URL(endpoint, event.getName()); URL eventEndpoint = new URL(baseEnndpointURL, event.getName());
new Thread(new HTTPost(eventEndpoint, event.toJSONString())).start(); new Thread(new HTTPost(eventEndpoint, event.toJSONString())).start();
} catch (MalformedURLException e) { } catch (MalformedURLException e) {
log.error("Cannot compute event endpoint URL. Event name: " + event.getName() + ", base endpoint: " log.error("Cannot compute event endpoint URL. Event name: " + event.getName() + ", base endpoint: "
+ endpoint, e); + baseEnndpointURL, e);
} }
} }
@ -61,11 +61,18 @@ public class HTTPEventSender implements EventSender {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("Getting OIDC token for client: " + clientId); log.debug("Getting OIDC token for client: " + clientId);
} }
JWTToken token = OpenIdConnectRESTHelper.queryClientToken(clientId, clientSecret, tokenURL); JWTToken token = null;
if (clientId != null && clientSecret != null && tokenURL != null) {
if (log.isDebugEnabled()) {
log.debug("Getting OIDC token for clientId '" + clientId + "' from: " + tokenURL);
}
token = OpenIdConnectRESTHelper.queryClientToken(clientId, clientSecret, tokenURL);
}
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("Performing HTTP POST to: " + endpoint); log.debug("Performing HTTP POST to: " + endpoint);
} }
HttpURLConnection connection = (HttpURLConnection) endpoint.openConnection(); HttpURLConnection connection = (HttpURLConnection) endpoint.openConnection();
connection.setRequestMethod("POST");
connection.setConnectTimeout(CONNECTION_TIMEOUT); connection.setConnectTimeout(CONNECTION_TIMEOUT);
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
log.trace("HTTP connection timeout set to: " + connection.getConnectTimeout()); log.trace("HTTP connection timeout set to: " + connection.getConnectTimeout());
@ -74,17 +81,26 @@ public class HTTPEventSender implements EventSender {
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
log.trace("HTTP connection Read timeout set to: " + connection.getReadTimeout()); log.trace("HTTP connection Read timeout set to: " + connection.getReadTimeout());
} }
connection.setDoOutput(true);
connection.setDoInput(true);
connection.setRequestProperty("Content-Type", "application/json"); connection.setRequestProperty("Content-Type", "application/json");
connection.setRequestProperty("Accept", "application/json"); // Commented out as per the Conductor issue: https://github.com/Netflix/conductor/issues/376
connection.setRequestProperty("Method", "POST"); // connection.setRequestProperty("Accept", "application/json");
connection.setRequestProperty("Authorization", token.getAsBearer()); connection.setDoOutput(true);
if (token != null) {
if (log.isDebugEnabled()) {
log.debug("Setting authorization header as: " + token.getAsBearer());
}
connection.setRequestProperty("Authorization", token.getAsBearer());
} else {
if (log.isDebugEnabled()) {
log.debug("Sending request without authorization header");
}
}
OutputStream os = connection.getOutputStream(); OutputStream os = connection.getOutputStream();
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
log.trace("Sending event JSON: " + jsonString); log.trace("Sending event JSON: " + jsonString);
} }
os.write(jsonString.getBytes("UTF-8")); os.write(jsonString.getBytes("UTF-8"));
os.flush();
os.close(); os.close();
StringBuilder sb = new StringBuilder(); StringBuilder sb = new StringBuilder();
@ -109,6 +125,7 @@ public class HTTPEventSender implements EventSender {
sb.append(line + "\n"); sb.append(line + "\n");
} }
br.close(); br.close();
isr.close();
if (ok) { if (ok) {
if (log.isDebugEnabled()) { if (log.isDebugEnabled()) {
log.debug("[" + httpResultCode + "] Event publish OK. Results: " + sb.toString()); log.debug("[" + httpResultCode + "] Event publish OK. Results: " + sb.toString());