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=e70af24a849263f550944b4822ef5b4c849de4eb;hb=e1384618cb6d1c06ec570d7798b783fa04a3d807;hp=66cf3920e7799e8415606c3fbf67a9a2601ea1bc;hpb=7a2f320afe9a0d3d7590365b26f3f5b0e8d9fd3b;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 66cf3920e..e70af24a8 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 @@ -7,17 +7,17 @@ import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.Session; -import javax.jms.TextMessage; import org.apache.commons.logging.Log; 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.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); @@ -26,21 +26,18 @@ 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 ExecutionModuleDescriptor getExecutionModuleDescriptor( final String moduleName, final String version) { - return (ExecutionModuleDescriptor) sendReceive(new AgentProxyMessageCreator( + return (ExecutionModuleDescriptor) sendReceive(new AgentMC( "getExecutionModuleDescriptor") { public void setArguments(Message message) throws JMSException { message.setStringProperty("moduleName", moduleName); @@ -50,46 +47,121 @@ public class JmsAgentProxy implements SlcAgent { } public List listExecutionModuleDescriptors() { - return ((SlcAgentDescriptor) sendReceive(new AgentProxyMessageCreator( + return ((SlcAgentDescriptor) sendReceive(new AgentMC( "listExecutionModuleDescriptors"))).getModuleDescriptors(); } - protected Object sendReceive(AgentProxyMessageCreator messageCreator) { + public void runSlcExecution(SlcExecution slcExecution) { + sendReceive(new AgentMC("runSlcExecution", slcExecution)); + } + + public boolean ping() { + Object response = sendReceive(new AgentMC("ping"), false); + if (response == null) + return false; + else { + ExecutionAnswer answer = (ExecutionAnswer) response; + return ExecutionAnswer.OK.equals(answer.getStatus()); + } + } + + protected Object sendReceive(AgentMC messageCreator) { + return sendReceive(messageCreator, true); + } + + /** + * @param timeoutException + * if true throws an exception if reception timeouted, else + * return null + */ + protected Object sendReceive(AgentMC messageCreator, + boolean timeoutException) { String correlationId = UUID.randomUUID().toString(); messageCreator.setCorrelationId(correlationId); send(messageCreator); - return processResponse(correlationId); + + Object response = processResponse(messageCreator, timeoutException); + + if (response instanceof ExecutionAnswer) { + ExecutionAnswer answer = (ExecutionAnswer) response; + if (ExecutionAnswer.ERROR.equals(answer.getStatus())) + throw new SlcException("Execution of '" + + messageCreator.getQuery() + "' failed on the agent " + + agentUuid + ": " + answer.getMessage() + + " (correlationId=" + correlationId + ")"); + else + return answer; + } else { + return response; + } } - protected void send(AgentProxyMessageCreator messageCreator) { + protected void send(AgentMC messageCreator) { jmsTemplate.send(requestDestination, messageCreator); if (log.isDebugEnabled()) - log.debug("Sent request" + messageCreator.getQuery() + " to agent " - + agentUuid + " with correlationId " - + messageCreator.getCorrelationId()); + log.debug("Sent query '" + messageCreator.getQuery() + + "' with correlationId " + + messageCreator.getCorrelationId() + " to agent " + + agentUuid); } - protected Object processResponse(String correlationId) { + protected Object processResponse(AgentMC messageCreator, + boolean timeoutException) { + String correlationId = messageCreator.getCorrelationId(); + String query = messageCreator.getQuery(); + Message responseMsg = null; try { - Message responseMsg = jmsTemplate.receiveSelected( - responseDestination, "JMSCorrelationID='" + correlationId - + "'"); - if (log.isDebugEnabled()) - log.debug("Received response with correlationId " - + correlationId); - return messageConverter.fromMessage(responseMsg); + responseMsg = jmsTemplate.receiveSelected(responseDestination, + "JMSCorrelationID='" + correlationId + "'"); } catch (Exception e) { - throw new SlcException("Could not process response from agent " - + agentUuid + " with correlationId " + correlationId, e); + throw new SlcException("Could not receive response from agent " + + agentUuid + " with correlationId " + correlationId + + " (query '" + query + "')", e); } + + if (responseMsg == null) {// timeout + if (timeoutException) + throw new SlcException("TIMEOUT: Query '" + query + "'" + + " with correlationId " + correlationId + + " sent to agent " + agentUuid + " timed out."); + else + return null; + } + if (log.isDebugEnabled()) + log.debug("Received response for query '" + query + + "' with correlationId " + correlationId + " from agent " + + agentUuid); + + try { + return fromMessage(responseMsg); + } catch (Exception e) { + throw new SlcException("Could not convert response from agent " + + agentUuid + " with correlationId " + correlationId + + " (query '" + query + "')", e); + } + } + + protected Object fromMessage(Message message) throws JMSException { + return jmsTemplate.getMessageConverter().fromMessage(message); } - protected class AgentProxyMessageCreator implements MessageCreator { + 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; private String correlationId; - public AgentProxyMessageCreator(String query) { + public AgentMC(String query) { + this.query = query; + } + + public AgentMC(String query, Object body) { this.query = query; + this.body = body; } public final Message createMessage(Session session) throws JMSException { @@ -97,7 +169,11 @@ public class JmsAgentProxy implements SlcAgent { throw new SlcException("Agent UUID not set"); if (correlationId == null) throw new SlcException("JMSCorrelationID not set"); - TextMessage msg = session.createTextMessage(); + final Message msg; + if (body == null) + msg = session.createTextMessage(); + else + msg = toMessage(body, session); msg.setStringProperty(JmsAgent.PROPERTY_SLC_AGENT_ID, agentUuid); msg.setStringProperty(JmsAgent.PROPERTY_QUERY, query); msg.setJMSCorrelationID(correlationId);