X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;f=runtime%2Forg.argeo.slc.support.activemq%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fslc%2Fjms%2FJmsAgent.java;h=e8d3cd83c33796b615733a2eeaf9b2b32dd9c8f4;hb=693cc53b426d088e49d746585bf7c5e197fc3998;hp=7623e79f0651767c8295dd03b9edc14f9c60800c;hpb=2401fec9d1374a1e7b9ebeb85dfd4d967a7a3e68;p=gpl%2Fargeo-slc.git diff --git a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgent.java b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgent.java index 7623e79f0..e8d3cd83c 100644 --- a/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgent.java +++ b/runtime/org.argeo.slc.support.activemq/src/main/java/org/argeo/slc/jms/JmsAgent.java @@ -4,7 +4,11 @@ import java.net.InetAddress; import java.net.UnknownHostException; import java.util.UUID; +import javax.jms.ConnectionFactory; import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -16,17 +20,21 @@ import org.argeo.slc.runtime.SlcAgentDescriptor; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.support.converter.MessageConverter; /** JMS based implementation of SLC Agent. */ public class JmsAgent extends AbstractAgent implements SlcAgent, - InitializingBean, DisposableBean { + InitializingBean, DisposableBean, MessageListener { private final static Log log = LogFactory.getLog(JmsAgent.class); private final SlcAgentDescriptor agentDescriptor; + private ConnectionFactory connectionFactory; private JmsTemplate jmsTemplate; private Destination agentRegister; private Destination agentUnregister; + private MessageConverter messageConverter; + public JmsAgent() { try { agentDescriptor = new SlcAgentDescriptor(); @@ -38,6 +46,10 @@ public class JmsAgent extends AbstractAgent implements SlcAgent, } public void afterPropertiesSet() throws Exception { + // Initialize JMS Template + jmsTemplate = new JmsTemplate(connectionFactory); + jmsTemplate.setMessageConverter(messageConverter); + jmsTemplate.convertAndSend(agentRegister, agentDescriptor); log.info("Agent #" + agentDescriptor.getUuid() + " registered to " + agentRegister); @@ -49,15 +61,6 @@ public class JmsAgent extends AbstractAgent implements SlcAgent, + agentUnregister); } - public void newExecution(SlcExecution slcExecution) { - log.info("Execute SlcExecution :" + slcExecution); - runSlcExecution(slcExecution); - } - - public void setJmsTemplate(JmsTemplate jmsTemplate) { - this.jmsTemplate = jmsTemplate; - } - public void setAgentRegister(Destination agentRegister) { this.agentRegister = agentRegister; } @@ -66,7 +69,36 @@ public class JmsAgent extends AbstractAgent implements SlcAgent, this.agentUnregister = agentUnregister; } + public void onMessage(Message message) { + // FIXME: we filter the messages on the client side, + // because of a weird problem with selector since moving to OSGi + try { + if (message.getStringProperty("slc-agentId").equals( + agentDescriptor.getUuid())) { + runSlcExecution((SlcExecution) messageConverter + .fromMessage(message)); + } + } catch (JMSException e) { + throw new SlcException("Cannot convert message " + message, e); + } + + } + public String getMessageSelector() { - return "slc-agentId=" + agentDescriptor.getUuid(); + String messageSelector = "slc-agentId=" + agentDescriptor.getUuid() + + ""; + // String messageSelector = "slc-agentId LIKE '%'"; + if (log.isDebugEnabled()) + log.debug("Message selector: '" + messageSelector + "'"); + return messageSelector; + } + + public void setMessageConverter(MessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; } + }