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