Author: cwiklik
Date: Tue Oct 16 16:30:51 2018
New Revision: 1844027
URL: http://svn.apache.org/viewvc?rev=1844027&view=rev
Log:
UIMA-5894 modified camel route to use JmsTemplate to publish JMS pings to test
broker viability. The JmsTemplate creates new broker connection for every ping.
Modified:
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/config/WebServerConfiguration.java
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/self/message/WebServerStateProcessor.java
Modified:
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/config/WebServerConfiguration.java
URL:
http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/config/WebServerConfiguration.java?rev=1844027&r1=1844026&r2=1844027&view=diff
==============================================================================
---
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/config/WebServerConfiguration.java
(original)
+++
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/config/WebServerConfiguration.java
Tue Oct 16 16:30:51 2018
@@ -29,9 +29,12 @@ import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.jms.JmsMessage;
+import org.apache.uima.ducc.common.authentication.BrokerCredentials;
+import
org.apache.uima.ducc.common.authentication.BrokerCredentials.Credentials;
import org.apache.uima.ducc.common.config.CommonConfiguration;
import org.apache.uima.ducc.common.config.DuccBlastGuardPredicate;
import org.apache.uima.ducc.common.utils.DuccLogger;
+import org.apache.uima.ducc.common.utils.Utils;
import org.apache.uima.ducc.common.utils.id.DuccId;
import org.apache.uima.ducc.transport.DuccTransportConfiguration;
import org.apache.uima.ducc.transport.dispatcher.DuccEventDispatcher;
@@ -62,7 +65,9 @@ public class WebServerConfiguration {
private DuccId jobid = null;
private AtomicBoolean singleton = new AtomicBoolean(false);
+ private Credentials brokerCredentials;
+
/**
* Instantiate {@link WebServerEventListener} which will handle
incoming messages.
*
@@ -109,13 +114,17 @@ public class WebServerConfiguration {
*/
private RouteBuilder routeBuilderForWebServerStatePost(final String
targetEndpointToReceiveWebServerStateUpdate, final int statePublishRate) throws
Exception {
final WebServerStateProcessor wssp = // an object responsible
for generating the state
- new WebServerStateProcessor();
+ new WebServerStateProcessor(common.brokerUrl,
+
brokerCredentials.getUsername(),
+ brokerCredentials.getPassword(),
+
targetEndpointToReceiveWebServerStateUpdate
+ );
return new RouteBuilder() {
public void configure() {
final Predicate blastFilter = new
DuccBlastGuardPredicate(duccLogger);
-
+ onException(Exception.class).handled(true).process(new
ErrorProcessor());
from("timer:webserverStateDumpTimer?fixedRate=true&period=" + statePublishRate)
// This route uses a filter to prevent sudden
bursts of messages which
// may flood DUCC daemons causing chaos. The
filter disposes any event
@@ -124,12 +133,19 @@ public class WebServerConfiguration {
//.process(xmStart)
.process(wssp)
//.process(xmEnded)
-
.to(targetEndpointToReceiveWebServerStateUpdate)
+
//.to(targetEndpointToReceiveWebServerStateUpdate)
;
}
};
}
-
+ public class ErrorProcessor implements Processor {
+
+ public void process(Exchange exchange) throws Exception {
+ // the caused by exception is stored in a property on the
exchange
+ Throwable causedBy =
exchange.getProperty(Exchange.EXCEPTION_CAUGHT, Throwable.class);
+ duccLogger.error("ErrorProcessor.process", jobid,
causedBy);
+ }
+ }
/**
* Creates and initializes {@link WebServerComponent} instance. @Bean
annotation identifies {@link WebServerComponent}
* as a Spring framework Bean which will be managed by Spring
container.
@@ -143,6 +159,19 @@ public class WebServerConfiguration {
String methodName = "webServer";
WebServerComponent ws = null;
try {
+ String brokerCredentialsFile =
+
System.getProperty("ducc.broker.credentials.file");
+ duccLogger.info("webServer", jobid, "Credentials
File:"+brokerCredentialsFile);
+ String path =
+
Utils.resolvePlaceholderIfExists(brokerCredentialsFile, System.getProperties());
+ duccLogger.info("webServer", jobid, "Path:"+path);
+ brokerCredentials = BrokerCredentials.get(path);
+
+ } catch( Throwable e) {
+ duccLogger.error("webServer", jobid, e);
+
+ }
+ try {
if(singleton.getAndSet(true)) {
try {
throw new RuntimeException("singleton
already present!");
Modified:
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/self/message/WebServerStateProcessor.java
URL:
http://svn.apache.org/viewvc/uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/self/message/WebServerStateProcessor.java?rev=1844027&r1=1844026&r2=1844027&view=diff
==============================================================================
---
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/self/message/WebServerStateProcessor.java
(original)
+++
uima/uima-ducc/trunk/uima-ducc-web/src/main/java/org/apache/uima/ducc/ws/self/message/WebServerStateProcessor.java
Tue Oct 16 16:30:51 2018
@@ -18,20 +18,75 @@
*/
package org.apache.uima.ducc.ws.self.message;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQTopic;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.uima.ducc.common.utils.DuccLogger;
+import org.springframework.jms.core.JmsTemplate;
+import org.springframework.jms.core.MessageCreator;
/**
* Processor of received webserver messages to self via broker
*/
public class WebServerStateProcessor implements Processor {
-
+ private final String brokerUrl;
+ private final String username;
+ private final String password;
+ private final String endpoint;
+ private DuccLogger duccLogger =
DuccLogger.getLogger(WebServerStateProcessor.class);
+ public WebServerStateProcessor(String brokerUrl, String username,
String password, String endpoint) {
+ this.brokerUrl = brokerUrl;
+ this.username = username;
+ this.password = password;
+ this.endpoint = endpoint;
+ }
@Override
public void process(Exchange exchange) throws Exception {
- WebServerStateDuccEvent wse = new WebServerStateDuccEvent();
- WebServerState wss = new WebServerState();
- wse.setState(wss);
- exchange.getIn().setBody(wse);
+ try {
+ ActiveMQConnectionFactory amqcf =
+ new ActiveMQConnectionFactory(username,
password, brokerUrl);
+ amqcf.setTrustAllPackages(true);
+ // this dispatcher will create a new connection for
every ping.
+ // No cleanup is needed as this also closes session and
connection
+ // after each ping.
+ JmsTemplate jmsDispatcher =
+ new JmsTemplate(amqcf);
+ // The endpoint is configured for Camel use and has the
following
+ // syntax: activemq:topic:<name>
+ // Strip the activemq:topic: part to just get the name
of the jms
+ // topic which will be used to create ActiveMQTopic
+ int pos = endpoint.indexOf("topic:")+"topic:".length();
+
+ String topicName = endpoint.substring(pos);
+ ActiveMQTopic topic = new ActiveMQTopic(topicName);
+ jmsDispatcher.send(topic, new MessageCreator() {
+
+ @Override
+ public Message createMessage(Session session)
throws JMSException {
+ // create a ping event message
+ ObjectMessage body =
session.createObjectMessage();
+ WebServerStateDuccEvent wse = new
WebServerStateDuccEvent();
+ WebServerState wss = new
WebServerState();
+ wse.setState(wss);
+ body.setObject(wse);
+ return body;
+ }
+
+ });
+
+ } catch( Throwable t ) {
+ duccLogger.error("process",null, t);
+ }
+// WebServerStateDuccEvent wse = new WebServerStateDuccEvent();
+// WebServerState wss = new WebServerState();
+// wse.setState(wss);
+// exchange.getIn().setBody(wse);
}
}
