Skip to content

GitLab

Commit b1efd7ac authored by Andre Freyssinet's avatar Andre Freyssinet
Browse files

Clean use of the 2 Rest/JMS API.

parent c56826b6
Hide whitespace changes
Inline Side-by-side
joram/joram/mom/extensions/restbridge/src/main/java/com/scalagent/joram/mom/dest/rest/RESTAcquisition.java
View file @ b1efd7ac
......@@ -28,7 +28,6 @@ import java.io.ObjectOutputStream;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Constructor;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.HashMap;
......@@ -42,6 +41,7 @@ import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.Invocation.Builder;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Form;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
......@@ -53,6 +53,7 @@ import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.excepts.MessageValueException;
import org.objectweb.joram.shared.messages.ConversionHelper;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.tools.rest.jms.JmsService;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
......@@ -69,17 +70,23 @@ public class RESTAcquisition implements AcquisitionHandler {
private String hostName = "localhost";
private int port = 8989;
// If true, use the API with authentication parameters in URL.
private boolean useOldAPI = false;
private Client client;
private WebTarget target;
private String userName = null;
private String password = null;
private long timeout = 0; //timeout = 0 => NoWait
private boolean persistent = false;
private int nbMaxMsg = 100;
private boolean mediaTypeJson = true;//default true "application/json"
private String destName;
private URI uriCreateConsumer;
private URI uriConsume;
private URI uriCloseConsumer;
......@@ -107,12 +114,19 @@ public class RESTAcquisition implements AcquisitionHandler {
+ " could not be parsed properly, use default value.", nfe);
}
}
if (properties.containsKey(DestinationConstants.REST_USE_OLD_API)) {
useOldAPI = Boolean.parseBoolean(properties.getProperty(DestinationConstants.REST_USE_OLD_API));
logger.log(BasicLevel.INFO, DestinationConstants.REST_USE_OLD_API + "=" + useOldAPI);
}
if (properties.containsKey(DestinationConstants.REST_USERNAME_PROP)) {
userName = properties.getProperty(DestinationConstants.REST_USERNAME_PROP);
}
if (properties.containsKey(DestinationConstants.REST_PASSWORD_PROP)) {
password = properties.getProperty(DestinationConstants.REST_PASSWORD_PROP);
}
if (properties.containsKey(DestinationConstants.TIMEOUT_PROP)) {
try {
timeout = Long.parseLong(properties.getProperty(DestinationConstants.TIMEOUT_PROP));
......@@ -133,48 +147,72 @@ public class RESTAcquisition implements AcquisitionHandler {
if (logger.isLoggable(BasicLevel.DEBUG)) {
logger.log(BasicLevel.DEBUG, "RESTAcquisition.init Target : " + target.getUri());
}
// create the consumer
if (uriCreateConsumer != null) {
client.target(uriCloseConsumer).request().accept(MediaType.TEXT_PLAIN).delete();
uriCreateConsumer = null;
}
if (properties.containsKey(DestinationConstants.MEDIA_TYPE_JSON_PROP)) {
mediaTypeJson = Boolean.parseBoolean(properties.getProperty(DestinationConstants.MEDIA_TYPE_JSON_PROP));
}
createConsumer();
}
public void createConsumer() {
// Close the consumer if it is already alive
close();
URI base = UriBuilder.fromUri("http://" + hostName + ":" + port + "/joram/").build();
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestAcquisitionAsync.createConsumer(), use base URI " + base);
Builder builder = target.path("jndi").path(destName).request();
Response response = builder.accept(MediaType.TEXT_PLAIN).head();
// if (logger.isLoggable(BasicLevel.DEBUG))
// logger.log(BasicLevel.DEBUG, "RESTAcquisition.createConsumer: response = " + response);
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RESTAcquisition.createConsumer: response = " + response);
if (201 != response.getStatus()) {
return;
}
uriCreateConsumer = response.getLink("create-consumer").getUri();
try {
// TODO (AF): to remove (use FormParam alternate URI)
uriCreateConsumer = new URI(uriCreateConsumer.toString() + "-fp");
} catch (URISyntaxException exc) {
logger.log(BasicLevel.ERROR, "RESTAcquisition.createConsumer: use FormParam alternate URI", exc);
uriCreateConsumer = null;
WebTarget target;
/* This code corresponds to the last version of the Rest/JMS API URI. Currently it can not be
* used for compatibility reason.
*
* uriCreateConsumer = response.getLink("create-consumer-fp").getUri();
*/
if (useOldAPI) {
URI uriCreateConsumer = response.getLink("create-consumer").getUri();
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RESTAcquisition.createConsumer(): create-producer = " + uriCreateConsumer);
target = client.target(uriCreateConsumer);
} else {
target = client.target(base).path("jndi").path(destName).path(JmsService.JMS_CREATE_CONS_FP);
}
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RESTAcquisition.createConsumer: uriCreateConsumer = " + uriCreateConsumer);
WebTarget wTarget = client.target(uriCreateConsumer)
.queryParam("name", "cons-" + destName)
.queryParam("client-id", "id-" + destName);
if (userName != null)
wTarget = wTarget.queryParam("user", userName);
if (password != null)
wTarget = wTarget.queryParam("password", password);
logger.log(BasicLevel.DEBUG, "RestAcquisition.createConsumer(): create-consumer = " + target.getUri());
// TODO (AF): We should fix name and client-id to be unique.
target = target.queryParam("name", "cons-" + destName).queryParam("client-id", "id-" + destName);
response = wTarget.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity("", MediaType.APPLICATION_FORM_URLENCODED));
// if (logger.isLoggable(BasicLevel.DEBUG))
// logger.log(BasicLevel.DEBUG, "RESTAcquisition.createConsumer: response = " + response);
Form auth = new Form();
if (useOldAPI) {
if (userName != null) target = target.queryParam("user", userName);
if (password != null) target = target.queryParam("password", password);
} else {
if (userName != null) auth.param("user", userName);
if (password != null) auth.param("password", password);
}
if (useOldAPI) {
response = target.request().accept(MediaType.TEXT_PLAIN).post(null);
} else {
response = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(auth, MediaType.APPLICATION_FORM_URLENCODED));
}
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RESTAcquisition.createConsumer: response = " + response);
if (201 == response.getStatus()) {
uriCloseConsumer = response.getLink("close-context").getUri();
uriConsume = response.getLink("receive-message").getUri();
......@@ -340,7 +378,6 @@ public class RESTAcquisition implements AcquisitionHandler {
message = new Message();
message.type = Message.TEXT;
message.setText(msg);
} else {
//JSON
builder = client.target(uriConsume)
......@@ -476,6 +513,8 @@ public class RESTAcquisition implements AcquisitionHandler {
}
}
// TODO (AF): Get the new uriConsume URI ?
if (message != null)
messages.add(message);
}
......@@ -510,9 +549,9 @@ public class RESTAcquisition implements AcquisitionHandler {
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "Close JMSAcquisition.");
if (uriCloseConsumer == null) {
if (uriCloseConsumer == null)
return;
}
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RESTAcquisition:: close-consumer = " + uriCloseConsumer);
client.target(uriCloseConsumer).request().accept(MediaType.TEXT_PLAIN).delete();
......
joram/joram/mom/extensions/restbridge/src/main/java/com/scalagent/joram/mom/dest/rest/RESTDistribution.java
View file @ b1efd7ac
......@@ -40,7 +40,9 @@ import javax.jms.MessageFormatException;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;import javax.ws.rs.core.MediaType;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Form;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
......@@ -49,6 +51,7 @@ import org.glassfish.jersey.client.ClientProperties;
import org.objectweb.joram.mom.dest.DistributionHandler;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.tools.rest.jms.JmsService;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
......@@ -66,6 +69,9 @@ public class RESTDistribution implements DistributionHandler {
private String hostName = "localhost";
private int port = 8989;
// If true, use the API with authentication parameters in URL.
private boolean useOldAPI = false;
private int connectTimeout = 10000;
private int readTimeout = 10000;
......@@ -85,6 +91,7 @@ public class RESTDistribution implements DistributionHandler {
private URI uriCloseProducer = null;
public void init(Properties properties, boolean firstTime) {
// Get URI informations: host, port, etc.
if (properties.containsKey(DestinationConstants.REST_HOST_PROP)) {
hostName = properties.getProperty(DestinationConstants.REST_HOST_PROP);
} else {
......@@ -103,6 +110,12 @@ public class RESTDistribution implements DistributionHandler {
"Missing property " + DestinationConstants.REST_PORT_PROP + ", use default value: " + port);
}
if (properties.containsKey(DestinationConstants.REST_USE_OLD_API)) {
useOldAPI = Boolean.parseBoolean(properties.getProperty(DestinationConstants.REST_USE_OLD_API));
logger.log(BasicLevel.INFO, DestinationConstants.REST_USE_OLD_API + "=" + useOldAPI);
}
// Get authentication informations: user, password.
if (properties.containsKey(DestinationConstants.REST_USERNAME_PROP)) {
userName = properties.getProperty(DestinationConstants.REST_USERNAME_PROP);
} else {
......@@ -122,6 +135,7 @@ public class RESTDistribution implements DistributionHandler {
} catch (NumberFormatException exc) { }
}
// Get Destination
destName = properties.getProperty(DestinationConstants.DESTINATION_NAME_PROP);
if (destName == null) {
logger.log(BasicLevel.ERROR,
......@@ -184,22 +198,52 @@ public class RESTDistribution implements DistributionHandler {
}
try {
URI uriCreateProducer = response.getLink("create-producer").getUri();
// TODO (AF): to remove (use FormParam alternate URI)
uriCreateProducer = new URI(uriCreateProducer.toString() + "-fp");
WebTarget target;
/* This code corresponds to the last version of the Rest/JMS API URI. Currently it can not be
* used for compatibility reason.
*
* uriCreateProducer = response.getLink("create-producer-fp").getUri();
* if (logger.isLoggable(BasicLevel.DEBUG))
* logger.log(BasicLevel.DEBUG, "RestDistribution.createProducer(): create-producer = " + uriCreateProducer);
* // TODO (AF): We should fix name and client-id to be unique. target =
* client.target(uriCreateProducer);
*/
if (useOldAPI) {
URI uriCreateProducer = response.getLink("create-producer").getUri();
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestDistribution.createProducer(): create-producer = " + uriCreateProducer);
// TODO (AF): We should fix name and client-id to be unique.
WebTarget target = client.target(uriCreateProducer);
target = client.target(uriCreateProducer);
} else {
target = client.target(base).path("jndi").path(destName).path(JmsService.JMS_CREATE_PROD_FP);
}
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RestDistribution.createProducer(): create-consumer = " + target.getUri());
logger.log(BasicLevel.WARN, "RestDistribution.createProducer(): create-consumer = " + target.getUri());
if (prodName != null) target = target.queryParam("name", prodName);
if (clientId != null) target = target.queryParam("client-id", clientId);
if (userName != null) target = target.queryParam("user", userName);
if (password != null) target = target.queryParam("password", password);
Form auth = new Form();
if (useOldAPI) {
if (userName != null) target = target.queryParam("user", userName);
if (password != null) target = target.queryParam("password", password);
} else {
if (userName != null) auth.param("user", userName);
if (password != null) auth.param("password", password);
}
if (idleTimeout != null) target = target.queryParam("idle-timeout", idleTimeout);
response = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity("", MediaType.APPLICATION_FORM_URLENCODED));
if (useOldAPI) {
response = target.request().accept(MediaType.TEXT_PLAIN).post(null);
} else {
response = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(auth, MediaType.APPLICATION_FORM_URLENCODED));
}
} catch (Exception exc) {
if (logger.isLoggable(BasicLevel.ERROR))
logger.log(BasicLevel.ERROR,
......
joram/joram/mom/extensions/restbridge/src/main/java/com/scalagent/joram/mom/dest/rest/RestAcquisitionAsync.java
View file @ b1efd7ac
......@@ -34,10 +34,12 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import javax.jms.JMSContext;
import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Form;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriBuilder;
......@@ -48,6 +50,7 @@ import org.objectweb.joram.mom.dest.AcquisitionDaemon;
import org.objectweb.joram.mom.dest.ReliableTransmitter;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.tools.rest.jms.JmsService;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;
......@@ -69,6 +72,9 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
private String hostName = "localhost";
private int port = 8989;
// If true, use the API with authentication parameters in URL.
private boolean useOldAPI = false;
private final int connectTimeout = 5000; // Value for jersey.config.client.connectTimeout.
private final int readTimeout = 5000; // Base value for jersey.config.client.readTimeout (timeout is added).
......@@ -129,6 +135,11 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
"Missing property " + DestinationConstants.REST_PORT_PROP + ", use default value: " + port);
}
if (properties.containsKey(DestinationConstants.REST_USE_OLD_API)) {
useOldAPI = Boolean.parseBoolean(properties.getProperty(DestinationConstants.REST_USE_OLD_API));
logger.log(BasicLevel.INFO, DestinationConstants.REST_USE_OLD_API + "=" + useOldAPI);
}
if (properties.containsKey(DestinationConstants.REST_USERNAME_PROP)) {
userName = properties.getProperty(DestinationConstants.REST_USERNAME_PROP);
} else {
......@@ -240,26 +251,48 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
}
try {
// Get URI to create a consumer in ClientAcknowledge mode.
URI uriCreateConsumer = response.getLink("create-consumer-client-ack").getUri();
// TODO (AF): Be careful, unlike the other bridge components the RestAcquisitionAsync always use the
// old API with authentication fields in the URI.
WebTarget target;
/* This code corresponds to the last version of the Rest/JMS API URI. Currently it can not be
* used for compatibility reason.
*
* URI uriCreateConsumer = response.getLink("create-consumer-fp-client-ack").getUri();
*/
if (useOldAPI) {
URI uriCreateConsumer = response.getLink("create-consumer-client-ack").getUri();
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RESTAcquisition.createConsumer(): create-consumer = " + uriCreateConsumer);
target = client.target(uriCreateConsumer);
} else {
target = client.target(base).path("jndi").path(destName).path(JmsService.JMS_CREATE_CONS_FP)
.queryParam("session-mode", JMSContext.CLIENT_ACKNOWLEDGE);
}
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG, "RestAcquisitionAsync.createConsumer(): create-consumer = " + uriCreateConsumer);
logger.log(BasicLevel.DEBUG, "RestAcquisitionAsync.createConsumer(): create-consumer = " + target.getUri());
// TODO (AF): We should fix name and client-id to be unique.
WebTarget target = client.target(uriCreateConsumer);
if (consName != null) target = target.queryParam("name", consName);
if (clientId != null) target = target.queryParam("client-id", clientId);
// Normally each consumer resource need to be explicitly closed, this parameter allows to set the idle time
// in seconds in which the consumer context will be closed if idle.
if (idleTimeout != null) target = target.queryParam("idle-timeout", idleTimeout);
if (userName != null) target = target.queryParam("user", userName);
if (password != null) target = target.queryParam("password", password);
// TODO (AF): to remove (use FormParam alternate URI)
// response = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity("", MediaType.APPLICATION_FORM_URLENCODED));
response = target.request().accept(MediaType.TEXT_PLAIN).post(null);
Form auth = new Form();
if (useOldAPI) {
if (userName != null) target = target.queryParam("user", userName);
if (password != null) target = target.queryParam("password", password);
} else {
if (userName != null) auth.param("user", userName);
if (password != null) auth.param("password", password);
}
if (useOldAPI) {
response = target.request().accept(MediaType.TEXT_PLAIN).post(null);
} else {
response = target.request().accept(MediaType.TEXT_PLAIN).post(Entity.entity(auth, MediaType.APPLICATION_FORM_URLENCODED));
}
} catch (Exception exc) {
if (logger.isLoggable(BasicLevel.ERROR))
logger.log(BasicLevel.ERROR,
......@@ -300,8 +333,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
.accept(MediaType.APPLICATION_JSON)
.get();
if (response.getStatus() != 200) {
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestAcquisitionAsync.receiveNextMessage: cannot receive next message ->" + response.getStatusInfo());
return null;
......@@ -309,8 +342,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
String json = response.readEntity(String.class);
if ((json == null) || json.isEmpty() || "null".equals(json)){
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestAcquisitionAsync.receiveNextMessage: receive empty message.");
return null;
......@@ -326,8 +359,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
try {
msg = createJSonMessage(map);
} catch (Exception exc) {
if (logger.isLoggable(BasicLevel.ERROR))
logger.log(BasicLevel.ERROR,
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
"RestAcquisitionAsync.receiveNextMessage: cannot create message", exc);
return null;
......@@ -342,8 +375,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
try {
msg = createTextMessage(text);
} catch (Exception exc) {
if (logger.isLoggable(BasicLevel.ERROR))
logger.log(BasicLevel.ERROR,
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
"RestAcquisitionAsync.receiveNextMessage: cannot create message", exc);
return null;
......@@ -363,8 +396,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
.accept(MediaType.TEXT_PLAIN)
.get();
if (response.getStatus() != 200) {
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestAcquisitionAsync.receiveNextMessage: cannot receive next message ->" + response.getStatusInfo());
return null;
......@@ -372,8 +405,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
String text = response.readEntity(String.class);
if ((text == null) || text.isEmpty()) {
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
if (logger.isLoggable(BasicLevel.DEBUG))
logger.log(BasicLevel.DEBUG,
"RestAcquisitionAsync.receiveNextMessage: receive empty message.");
return null;
......@@ -382,8 +415,8 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
try {
msg = createTextMessage(text);
} catch (Exception exc) {
if (logger.isLoggable(BasicLevel.ERROR))
logger.log(BasicLevel.ERROR,
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.WARN,
"RestAcquisitionAsync.receiveNextMessage: cannot create message", exc);
return null;
......@@ -492,8 +525,6 @@ public class RestAcquisitionAsync implements AcquisitionDaemon {
if (header.containsKey("CorrelationIDAsBytes")) {
msg.setJMSCorrelationIDAsBytes((byte[]) header.get("CorrelationIDAsBytes"));
if (logger.isLoggable(BasicLevel.WARN))
logger.log(BasicLevel.DEBUG, "-- CorrelationIDAsBytes = " + header.get("CorrelationIDAsBytes"));
}
// TODO (AF): Is it correct? The destination should correspond to the current destination.
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment