[jboss-svn-commits] JBL Code SVN: r8472 - labs/jbossesb/trunk/product/core/listeners/src/org/jboss/soa/esb/listeners/message

[jboss-svn-commits at lists.jboss.org]

Author: estebanschifman
Date: 2006-12-20 19:25:53 -0500 (Wed, 20 Dec 2006)
New Revision: 8472

Added:
   labs/jbossesb/trunk/product/core/listeners/src/org/jboss/soa/esb/listeners/message/CbrListener.java
Log:
New class CbrListener - transport independent CBR service

Added: labs/jbossesb/trunk/product/core/listeners/src/org/jboss/soa/esb/listeners/message/CbrListener.java
===================================================================
--- labs/jbossesb/trunk/product/core/listeners/src/org/jboss/soa/esb/listeners/message/CbrListener.java	2006-12-21 00:08:09 UTC (rev 8471)
+++ labs/jbossesb/trunk/product/core/listeners/src/org/jboss/soa/esb/listeners/message/CbrListener.java	2006-12-21 00:25:53 UTC (rev 8472)
@@ -0,0 +1,141 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2006, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.soa.esb.listeners.message;
+
+import java.net.URISyntaxException;
+import java.util.Collection;
+
+import org.apache.log4j.Logger;
+import org.apache.log4j.Priority;
+import org.jboss.soa.esb.ConfigurationException;
+import org.jboss.soa.esb.addressing.EPR;
+import org.jboss.soa.esb.couriers.CourierException;
+import org.jboss.soa.esb.couriers.CourierFactory;
+import org.jboss.soa.esb.couriers.CourierTimeoutException;
+import org.jboss.soa.esb.helpers.ConfigTree;
+import org.jboss.soa.esb.listeners.ListenerManager;
+import org.jboss.soa.esb.listeners.ListenerTagNames;
+import org.jboss.soa.esb.message.Message;
+import org.jboss.soa.esb.services.routing.MessageRouter;
+import org.jboss.soa.esb.services.routing.MessageRouterException;
+import org.jboss.soa.esb.services.routing.cbr.ContentBasedRouter;
+import org.jboss.soa.esb.services.routing.cbr.ContentBasedRouterFactory;
+
+/**
+ * Esb Message aware Content Based Router listener.
+ * @author <a href="mailto:kurt.stam at redhat.com">Kurt Stam</a>
+ * @since Version 4.0
+ */
+
+public class CbrListener extends MessageAwareListener 
+{
+	protected static transient Logger _logger = Logger.getLogger(CbrListener.class);
+	protected ContentBasedRouter _cbr;
+	protected String _ruleSet;
+	protected String _ruleLanguage;
+	
+	public CbrListener(ListenerManager controller, ConfigTree config) throws ConfigurationException 
+	{
+		super(controller, config);
+	}
+	
+    protected void checkMyParms() throws ConfigurationException 
+    {
+    	super.checkMyParms();
+    	if (_config.getFirstChild(ListenerTagNames.CBR_TAG)==null) {
+    		_logger.error("Required child element " + ListenerTagNames.CBR_TAG + " not found in " + _config.getName() + ".");
+    		throw new ConfigurationException("Required child element " + ListenerTagNames.CBR_TAG + " not found.");
+    	} else {
+    		_ruleSet =_config.getFirstChild(ListenerTagNames.CBR_TAG).getAttribute(ListenerTagNames.RULE_SET_TAG);
+    		if (_ruleSet==null) {
+    			throw new ConfigurationException("Required attribute " + ListenerTagNames.RULE_SET_TAG + " not found.");
+    		}
+    		_ruleLanguage =_config.getFirstChild(ListenerTagNames.CBR_TAG).getAttribute(ListenerTagNames.RULE_LANGUAGE_TAG);
+    	}
+
+    } // ________________________________
+	
+	@Override
+	public boolean initializeRun() 
+	{
+		if (!super.initializeRun())
+			return false;
+		
+    	try
+    	{
+    		_cbr =ContentBasedRouterFactory.getRouter();
+    	}
+    	catch (MessageRouterException e) 
+    	{
+    		return false;
+    	}
+		return true;
+	} //________________________________
+
+	@Override
+	public void waitForEventAndProcess(long maxWaitMillis) 
+	{
+		Message message = null;
+		try { message = (maxWaitMillis > 0 ) ? _pickUpCourier.pickup(maxWaitMillis) : null; }
+		catch (CourierTimeoutException e) { return; }
+		catch (CourierException e) 
+		{
+			_logger.error(e);
+			return;
+		}
+
+    	if (null!=message)
+    	{	
+    		try	
+    		{ 
+	        	if (null!=message) {
+        			_logger.log(Priority.DEBUG, "Sending message to the Content Based Router.");
+        			Collection<String> destinationServices = _cbr.route(_ruleSet, _ruleLanguage, message);
+        			if (Boolean.FALSE.equals(message.getProperties().getProperty(MessageRouter.DELIVER_MESSAGES))) {
+        				message.getProperties().setProperty(MessageRouter.ROUTING_DESTINATION_SERVICE_LIST, destinationServices);
+        				try  { 
+        					_logger.debug("Replying to caller");
+        					if (message.getHeader().getCall()!=null && message.getHeader().getCall().getReplyTo()!=null) {
+        						EPR replyTo = message.getHeader().getCall().getReplyTo();
+        						CourierFactory.getCourier(replyTo).deliver(message);
+        					} else {
+        						_logger.log(Priority.ERROR, "Unable to reply to caller. Could not optain replyTo EPR.");
+        					}
+        				} catch (URISyntaxException use) {
+        					_logger.log(Priority.ERROR, "Unable to reply to caller. "+ use.getLocalizedMessage(), use);
+        				} catch (CourierException ce) {
+        					_logger.log(Priority.ERROR, "Unable to reply to caller. "+ ce.getLocalizedMessage(), ce);
+        				}
+        			}
+        		}
+    		}
+    		catch (Exception e)	
+    		{	
+    			_logger.error(e); 	
+    			return; 
+    		}
+    	}
+		
+	} //________________________________
+
+}