runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgentProxy.java

   1 package org.argeo.slc.jms;
   2
   3 import java.util.List;
   4 import java.util.UUID;
   5
   6 import javax.jms.Destination;
   7 import javax.jms.JMSException;
   8 import javax.jms.Message;
   9 import javax.jms.Session;
  10 import javax.jms.TextMessage;
  11
  12 import org.apache.commons.logging.Log;
  13 import org.apache.commons.logging.LogFactory;
  14 import org.argeo.slc.SlcException;
  15 import org.argeo.slc.execution.ExecutionModuleDescriptor;
  16 import org.argeo.slc.runtime.SlcAgent;
  17 import org.argeo.slc.runtime.SlcAgentDescriptor;
  18 import org.springframework.jms.core.JmsTemplate;
  19 import org.springframework.jms.core.MessageCreator;
  20 import org.springframework.jms.support.converter.MessageConverter;
  21
  22 public class JmsAgentProxy implements SlcAgent {
  23         private final static Log log = LogFactory.getLog(JmsAgentProxy.class);
  24
  25         private final String agentUuid;
  26         private final Destination requestDestination;
  27         private final Destination responseDestination;
  28         private final JmsTemplate jmsTemplate;
  29         private final MessageConverter messageConverter;
  30
  31         public JmsAgentProxy(String agentUuid, Destination requestDestination,
  32                         Destination responseDestination, JmsTemplate jmsTemplate,
  33                         MessageConverter messageConverter) {
  34                 this.agentUuid = agentUuid;
  35                 this.requestDestination = requestDestination;
  36                 this.responseDestination = responseDestination;
  37                 this.jmsTemplate = jmsTemplate;
  38                 this.messageConverter = messageConverter;
  39         }
  40
  41         public ExecutionModuleDescriptor getExecutionModuleDescriptor(
  42                         final String moduleName, final String version) {
  43                 return (ExecutionModuleDescriptor) sendReceive(new AgentProxyMessageCreator(
  44                                 "getExecutionModuleDescriptor") {
  45                         public void setArguments(Message message) throws JMSException {
  46                                 message.setStringProperty("moduleName", moduleName);
  47                                 message.setStringProperty("version", version);
  48                         }
  49                 });
  50         }
  51
  52         public List<ExecutionModuleDescriptor> listExecutionModuleDescriptors() {
  53                 return ((SlcAgentDescriptor) sendReceive(new AgentProxyMessageCreator(
  54                                 "listExecutionModuleDescriptors"))).getModuleDescriptors();
  55         }
  56
  57         protected Object sendReceive(AgentProxyMessageCreator messageCreator) {
  58                 String correlationId = UUID.randomUUID().toString();
  59                 messageCreator.setCorrelationId(correlationId);
  60                 send(messageCreator);
  61                 return processResponse(correlationId);
  62         }
  63
  64         protected void send(AgentProxyMessageCreator messageCreator) {
  65                 jmsTemplate.send(requestDestination, messageCreator);
  66                 if (log.isDebugEnabled())
  67                         log.debug("Sent request" + messageCreator.getQuery() + " to agent "
  68                                         + agentUuid + " with correlationId "
  69                                         + messageCreator.getCorrelationId());
  70         }
  71
  72         protected Object processResponse(String correlationId) {
  73                 try {
  74                         Message responseMsg = jmsTemplate.receiveSelected(
  75                                         responseDestination, "JMSCorrelationID='" + correlationId
  76                                                         + "'");
  77                         if (log.isDebugEnabled())
  78                                 log.debug("Received response with correlationId "
  79                                                 + correlationId);
  80                         return messageConverter.fromMessage(responseMsg);
  81                 } catch (Exception e) {
  82                         throw new SlcException("Could not process response from agent "
  83                                         + agentUuid + " with correlationId " + correlationId, e);
  84                 }
  85         }
  86
  87         protected class AgentProxyMessageCreator implements MessageCreator {
  88                 private final String query;
  89                 private String correlationId;
  90
  91                 public AgentProxyMessageCreator(String query) {
  92                         this.query = query;
  93                 }
  94
  95                 public final Message createMessage(Session session) throws JMSException {
  96                         if (agentUuid == null)
  97                                 throw new SlcException("Agent UUID not set");
  98                         if (correlationId == null)
  99                                 throw new SlcException("JMSCorrelationID not set");
 100                         TextMessage msg = session.createTextMessage();
 101                         msg.setStringProperty(JmsAgent.PROPERTY_SLC_AGENT_ID, agentUuid);
 102                         msg.setStringProperty(JmsAgent.PROPERTY_QUERY, query);
 103                         msg.setJMSCorrelationID(correlationId);
 104                         setArguments(msg);
 105                         return msg;
 106                 }
 107
 108                 protected void setArguments(Message message) throws JMSException {
 109                 }
 110
 111                 public String getQuery() {
 112                         return query;
 113                 }
 114
 115                 public String getCorrelationId() {
 116                         return correlationId;
 117                 }
 118
 119                 public void setCorrelationId(String correlationId) {
 120                         this.correlationId = correlationId;
 121                 }
 122
 123         }
 124 }