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=90818a572d0524c87a2bd7241560313af0204aa4;hb=56f9a29fc3d208bb1a0e9476b9d67dc372098e70;hp=7aff0ab20ab3315f21569251d667a82d85a1d042;hpb=11a9d0e8d3e9f610d0546463cfaf07ae884cb249;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 7aff0ab20..90818a572 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,6 +2,8 @@ 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.Destination;
@@ -12,18 +14,35 @@ import javax.jms.MessageListener;
 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.msg.ExecutionAnswer;
+import org.argeo.slc.msg.ReferenceList;
+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.core.MessagePostProcessor;
 
 /** JMS based implementation of SLC Agent. */
-public class JmsAgent implements SlcAgent, MessageListener, InitializingBean {
+public class JmsAgent extends AbstractAgent implements SlcAgent,
+		InitializingBean, DisposableBean, MessageListener {
+	public final static String PROPERTY_QUERY = "query";
+	public final static String PROPERTY_SLC_AGENT_ID = "slc_agentId";
+
+	public final static String QUERY_PING_ALL = "pingAll";
+
 	private final static Log log = LogFactory.getLog(JmsAgent.class);
 
 	private final SlcAgentDescriptor agentDescriptor;
 	private JmsTemplate jmsTemplate;
 	private Destination agentRegister;
+	private Destination agentUnregister;
+
+	private Destination responseDestination;
 
 	public JmsAgent() {
 		try {
@@ -41,21 +60,150 @@ public class JmsAgent implements SlcAgent, MessageListener, InitializingBean {
 				+ agentRegister);
 	}
 
-	public void onMessage(Message message) {
+	public void destroy() throws Exception {
+		jmsTemplate.convertAndSend(agentUnregister, agentDescriptor);
+		log.info("Agent #" + agentDescriptor.getUuid() + " unregistered from "
+				+ agentUnregister);
+	}
+
+	public void setAgentRegister(Destination agentRegister) {
+		this.agentRegister = agentRegister;
+	}
+
+	public void setAgentUnregister(Destination agentUnregister) {
+		this.agentUnregister = agentUnregister;
+	}
+
+	public String getMessageSelector() {
+		String messageSelector = "slc_agentId='" + agentDescriptor.getUuid()
+				+ "'";
+		// 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 boolean ping() {
+		return true;
+	}
+
+	public void onMessage(final Message message) {
+		final String query;
+		final String correlationId;
+		try {
+			query = message.getStringProperty(PROPERTY_QUERY);
+			correlationId = message.getJMSCorrelationID();
+		} catch (JMSException e1) {
+			throw new SlcException("Cannot analyze incoming message " + message);
+		}
+
+		final Object response;
+		final Destination destinationSend;
+		if (QUERY_PING_ALL.equals(query)) {
+			ReferenceList refList = (ReferenceList) convertFrom(message);
+			if (!refList.getReferences().contains(agentDescriptor.getUuid())) {
+				response = agentDescriptor;
+				destinationSend = agentRegister;
+				log.info("Agent #" + agentDescriptor.getUuid()
+						+ " registering to " + agentRegister
+						+ " in reply to a " + QUERY_PING_ALL + " query");
+			} else {
+				return;
+			}
+		} else {
+			response = process(query, message);
+			destinationSend = responseDestination;
+		}
+
+		new Thread() {
+			public void run() {
+				// Send response
+				jmsTemplate.convertAndSend(destinationSend, response,
+						new MessagePostProcessor() {
+							public Message postProcessMessage(
+									Message messageToSend) throws JMSException {
+								messageToSend.setStringProperty(PROPERTY_QUERY,
+										query);
+								messageToSend.setStringProperty(
+										PROPERTY_SLC_AGENT_ID, agentDescriptor
+												.getUuid());
+								messageToSend
+										.setJMSCorrelationID(correlationId);
+								return messageToSend;
+							}
+						});
+				if (log.isDebugEnabled())
+					log.debug("Sent response to query '" + query
+							+ "' with correlationId " + correlationId);
+			}
+		}.start();
+
+	}
+
+	/** @return response */
+	public Object process(String query, Message message) {
 		try {
-			log.info("Received message " + message.getJMSMessageID());
+			if ("getExecutionModuleDescriptor".equals(query)) {
+				String moduleName = message.getStringProperty("moduleName");
+				String version = message.getStringProperty("version");
+				return getExecutionModuleDescriptor(moduleName, version);
+			} else if ("listExecutionModuleDescriptors".equals(query)) {
+
+				List<ExecutionModuleDescriptor> lst = listExecutionModuleDescriptors();
+				SlcAgentDescriptor agentDescriptorToSend = new SlcAgentDescriptor(
+						agentDescriptor);
+				agentDescriptorToSend.setModuleDescriptors(lst);
+				return agentDescriptorToSend;
+			} else if ("runSlcExecution".equals(query)) {
+				SlcExecution slcExecution = (SlcExecution) convertFrom(message);
+				runSlcExecution(slcExecution);
+				return ExecutionAnswer.ok("Execution started on agent "
+						+ agentDescriptor.getUuid());
+			} else if ("ping".equals(query)) {
+				return ExecutionAnswer.ok("Agent " + agentDescriptor.getUuid()
+						+ " is alive.");
+			} else {
+				throw new SlcException("Unsupported query " + query);
+			}
+		} catch (Exception e) {
+			log.error("Processing of query " + query + " failed", e);
+			return ExecutionAnswer.error(e);
+		}
+	}
+
+	protected Object convertFrom(Message message) {
+		try {
+			return jmsTemplate.getMessageConverter().fromMessage(message);
 		} catch (JMSException e) {
-			e.printStackTrace();
+			throw new SlcException("Cannot convert message", e);
 		}
+	}
 
+	public void setResponseDestination(Destination responseDestination) {
+		this.responseDestination = responseDestination;
 	}
 
 	public void setJmsTemplate(JmsTemplate jmsTemplate) {
 		this.jmsTemplate = jmsTemplate;
 	}
 
-	public void setAgentRegister(Destination agentRegister) {
-		this.agentRegister = agentRegister;
-	}
-
 }