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=baddb1c604f075f4af5a5cc7667531331f48631a;hb=a5bc0826e17c1fae461129be871fd51d40e35b8b;hp=e8d3cd83c33796b615733a2eeaf9b2b32dd9c8f4;hpb=0f26a1ac2596c4f949973028af21bc0f951b7b7a;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 e8d3cd83c..baddb1c60 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
@@ -2,38 +2,50 @@ package org.argeo.slc.jms;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
 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 javax.jms.MessageProducer;
+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.core.runtime.AbstractAgent;
+import org.argeo.slc.execution.ExecutionModule;
+import org.argeo.slc.execution.ExecutionModuleDescriptor;
 import org.argeo.slc.process.SlcExecution;
 import org.argeo.slc.runtime.SlcAgent;
 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;
+import org.springframework.jms.listener.SessionAwareMessageListener;
+import org.springframework.jms.support.converter.MessageConversionException;
 
 /** JMS based implementation of SLC Agent. */
 public class JmsAgent extends AbstractAgent implements SlcAgent,
-		InitializingBean, DisposableBean, MessageListener {
+		InitializingBean, DisposableBean, SessionAwareMessageListener {
+	public final static String PROPERTY_QUERY = "query";
+	public final static String PROPERTY_SLC_AGENT_ID = "slc_agentId";
+
 	private final static Log log = LogFactory.getLog(JmsAgent.class);
 
 	private final SlcAgentDescriptor agentDescriptor;
-	private ConnectionFactory connectionFactory;
+	// private ConnectionFactory connectionFactory;
 	private JmsTemplate jmsTemplate;
 	private Destination agentRegister;
 	private Destination agentUnregister;
 
-	private MessageConverter messageConverter;
+	// private Destination requestDestination;
+	private Destination responseDestination;
+
+	// private MessageConverter messageConverter;
 
 	public JmsAgent() {
 		try {
@@ -47,8 +59,8 @@ public class JmsAgent extends AbstractAgent implements SlcAgent,
 
 	public void afterPropertiesSet() throws Exception {
 		// Initialize JMS Template
-		jmsTemplate = new JmsTemplate(connectionFactory);
-		jmsTemplate.setMessageConverter(messageConverter);
+		// jmsTemplate = new JmsTemplate(connectionFactory);
+		// jmsTemplate.setMessageConverter(messageConverter);
 
 		jmsTemplate.convertAndSend(agentRegister, agentDescriptor);
 		log.info("Agent #" + agentDescriptor.getUuid() + " registered to "
@@ -69,36 +81,96 @@ 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() {
+		String messageSelector = "slc_agentId='" + agentDescriptor.getUuid()
+				+ "'";
+		// if (log.isDebugEnabled())
+		// log.debug("Message selector: " + messageSelector);
+		return messageSelector;
+	}
+
+	public void onMessage(Message message, Session session) throws JMSException {
+		MessageProducer producer = session.createProducer(responseDestination);
+		String query = message.getStringProperty(PROPERTY_QUERY);
+		String correlationId = message.getJMSCorrelationID();
+		if (log.isDebugEnabled())
+			log.debug("Received query " + query + " with correlationId "
+					+ correlationId);
+
+		Message responseMsg = null;
+		if ("getExecutionModuleDescriptor".equals(query)) {
+			String moduleName = message.getStringProperty("moduleName");
+			String version = message.getStringProperty("version");
+			ExecutionModuleDescriptor emd = getExecutionModuleDescriptor(
+					moduleName, version);
+			responseMsg = jmsTemplate.getMessageConverter().toMessage(emd,
+					session);
+		} else if ("listExecutionModuleDescriptors".equals(query)) {
+
+			List<ExecutionModuleDescriptor> lst = listExecutionModuleDescriptors();
+			SlcAgentDescriptor agentDescriptorToSend = new SlcAgentDescriptor(
+					agentDescriptor);
+			agentDescriptorToSend.setModuleDescriptors(lst);
+			responseMsg = jmsTemplate.getMessageConverter().toMessage(
+					agentDescriptorToSend, session);
+		} else if ("newExecution".equals(query)) {
+
+			SlcExecution slcExecution = (SlcExecution) jmsTemplate
+					.getMessageConverter().fromMessage(message);
+			runSlcExecution(slcExecution);
+			return;
+		} else {
+			// try {
+			// // FIXME: generalize
+			// SlcExecution slcExecution = (SlcExecution) jmsTemplate
+			// .getMessageConverter().fromMessage(message);
+			// runSlcExecution(slcExecution);
+			// } catch (MessageConversionException e) {
+			// if (log.isDebugEnabled())
+			// log.debug("Unsupported query " + query, e);
+			// }
+			if (log.isDebugEnabled())
+				log.debug("Unsupported query " + query);
+			return;
+		}
+
+		if (responseMsg != null) {
+			responseMsg.setJMSCorrelationID(correlationId);
+			producer.send(responseMsg);
+			if (log.isDebugEnabled())
+				log.debug("Sent response to query " + query
+						+ " with correlationId " + correlationId + ": "
+						+ responseMsg);
 		}
 
 	}
 
-	public String getMessageSelector() {
-		String messageSelector = "slc-agentId=" + agentDescriptor.getUuid()
-				+ "";
-		// String messageSelector = "slc-agentId LIKE '%'";
-		if (log.isDebugEnabled())
-			log.debug("Message selector: '" + messageSelector + "'");
-		return messageSelector;
+	public ExecutionModuleDescriptor getExecutionModuleDescriptor(
+			String moduleName, String version) {
+		return getModulesManager().getExecutionModuleDescriptor(moduleName,
+				version);
+	}
+
+	public List<ExecutionModuleDescriptor> listExecutionModuleDescriptors() {
+		List<ExecutionModule> modules = getModulesManager()
+				.listExecutionModules();
+
+		List<ExecutionModuleDescriptor> descriptors = new ArrayList<ExecutionModuleDescriptor>();
+		for (ExecutionModule module : modules) {
+			ExecutionModuleDescriptor md = new ExecutionModuleDescriptor();
+			md.setName(module.getName());
+			md.setVersion(module.getVersion());
+			descriptors.add(md);
+		}
+		return descriptors;
 	}
 
-	public void setMessageConverter(MessageConverter messageConverter) {
-		this.messageConverter = messageConverter;
+	public void setResponseDestination(Destination responseDestination) {
+		this.responseDestination = responseDestination;
 	}
 
-	public void setConnectionFactory(ConnectionFactory connectionFactory) {
-		this.connectionFactory = connectionFactory;
+	public void setJmsTemplate(JmsTemplate jmsTemplate) {
+		this.jmsTemplate = jmsTemplate;
 	}
 
 }