X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;f=runtime%2Forg.argeo.slc.support.activemq%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fjms%2FJmsAgentProxy.java;h=2230cb876583e58cf6d8a53125205b8c9e0895f7;hb=d601802d5a50e4e30eb639a08eff5270b63fe599;hp=8837a8155993cdc4c9a04954f45e228c848e3961;hpb=a9bd656db864f3f648d375deef6833c2f53f29ba;p=gpl%2Fargeo-slc.git diff --git a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgentProxy.java b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgentProxy.java index 8837a8155..2230cb876 100644 --- a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgentProxy.java +++ b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgentProxy.java @@ -1,3 +1,19 @@ +/* + * Copyright (C) 2010 Mathieu Baudier + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package org.argeo.slc.jms; import java.util.List; @@ -14,12 +30,12 @@ import org.apache.commons.logging.LogFactory; import org.argeo.slc.SlcException; import org.argeo.slc.execution.ExecutionModuleDescriptor; import org.argeo.slc.msg.ExecutionAnswer; +import org.argeo.slc.msg.MsgConstants; import org.argeo.slc.process.SlcExecution; import org.argeo.slc.runtime.SlcAgent; import org.argeo.slc.runtime.SlcAgentDescriptor; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.core.MessageCreator; -import org.springframework.jms.support.converter.MessageConverter; public class JmsAgentProxy implements SlcAgent { private final static Log log = LogFactory.getLog(JmsAgentProxy.class); @@ -28,16 +44,17 @@ public class JmsAgentProxy implements SlcAgent { private final Destination requestDestination; private final Destination responseDestination; private final JmsTemplate jmsTemplate; - private final MessageConverter messageConverter; public JmsAgentProxy(String agentUuid, Destination requestDestination, - Destination responseDestination, JmsTemplate jmsTemplate, - MessageConverter messageConverter) { + Destination responseDestination, JmsTemplate jmsTemplate) { this.agentUuid = agentUuid; this.requestDestination = requestDestination; this.responseDestination = responseDestination; this.jmsTemplate = jmsTemplate; - this.messageConverter = messageConverter; + } + + public String getAgentUuid() { + return agentUuid; } public ExecutionModuleDescriptor getExecutionModuleDescriptor( @@ -71,7 +88,18 @@ public class JmsAgentProxy implements SlcAgent { } protected Object sendReceive(AgentMC messageCreator) { - return sendReceive(messageCreator, true); + long begin = System.currentTimeMillis(); + Object res; + try { + res = sendReceive(messageCreator, true); + } finally { + if (log.isTraceEnabled()) + log.trace("To agent #" + agentUuid + " in " + + (System.currentTimeMillis() - begin) + " ms, query '" + + messageCreator.getQuery() + "', correlationId " + + messageCreator.getCorrelationId()); + } + return res; } /** @@ -103,11 +131,6 @@ public class JmsAgentProxy implements SlcAgent { protected void send(AgentMC messageCreator) { jmsTemplate.send(requestDestination, messageCreator); - if (log.isDebugEnabled()) - log.debug("Sent query '" + messageCreator.getQuery() - + "' with correlationId " - + messageCreator.getCorrelationId() + " to agent " - + agentUuid); } protected Object processResponse(AgentMC messageCreator, @@ -132,13 +155,9 @@ public class JmsAgentProxy implements SlcAgent { else return null; } - if (log.isDebugEnabled()) - log.debug("Received response for query '" + query - + "' with correlationId " + correlationId + " from agent " - + agentUuid); try { - return messageConverter.fromMessage(responseMsg); + return fromMessage(responseMsg); } catch (Exception e) { throw new SlcException("Could not convert response from agent " + agentUuid + " with correlationId " + correlationId @@ -146,6 +165,15 @@ public class JmsAgentProxy implements SlcAgent { } } + protected Object fromMessage(Message message) throws JMSException { + return jmsTemplate.getMessageConverter().fromMessage(message); + } + + protected Message toMessage(Object obj, Session session) + throws JMSException { + return jmsTemplate.getMessageConverter().toMessage(obj, session); + } + protected class AgentMC implements MessageCreator { private final String query; private Object body = null; @@ -169,11 +197,22 @@ public class JmsAgentProxy implements SlcAgent { if (body == null) msg = session.createTextMessage(); else - msg = messageConverter.toMessage(body, session); - msg.setStringProperty(JmsAgent.PROPERTY_SLC_AGENT_ID, agentUuid); + msg = toMessage(body, session); + msg + .setStringProperty(MsgConstants.PROPERTY_SLC_AGENT_ID, + agentUuid); msg.setStringProperty(JmsAgent.PROPERTY_QUERY, query); msg.setJMSCorrelationID(correlationId); setArguments(msg); + if (msg instanceof TextMessage) { + TextMessage textMessage = (TextMessage) msg; + if (textMessage.getText() == null) { + // TODO: remove workaround when upgrading to ActiveMQ 5.3 + // Workaround for + // https://issues.apache.org/activemq/browse/AMQ-2046 + textMessage.setText(""); + } + } return msg; }