[esb-java-dev] svn commit r24349 - branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers

[asankaa at wso2.com]

Author: asankaa
Date: Thu Nov 20 05:35:28 2008
New Revision: 24349
URL: http://wso2.org/svn/browse/wso2?view=rev&revision=24349

Log:
fix the subscription matching issues of the default sub manager , implement the property getters and setters


Modified:
   branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers/DefaultInMemorySubscriptionManager.java

Modified: branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers/DefaultInMemorySubscriptionManager.java
URL: http://wso2.org/svn/browse/wso2/branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers/DefaultInMemorySubscriptionManager.java?rev=24349&r1=24348&r2=24349&view=diff
==============================================================================
--- branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers/DefaultInMemorySubscriptionManager.java	(original)
+++ branches/synapse/1.2.wso2v1/modules/core/src/main/java/org/apache/synapse/eventing/managers/DefaultInMemorySubscriptionManager.java	Thu Nov 20 05:35:28 2008
@@ -20,9 +20,15 @@
 package org.apache.synapse.eventing.managers;
 
 import org.apache.synapse.MessageContext;
+import org.apache.synapse.SynapseException;
+import org.apache.synapse.util.xpath.SynapseXPath;
 import org.apache.synapse.eventing.EventFilter;
 import org.apache.synapse.eventing.Subscription;
 import org.apache.synapse.eventing.SubscriptionManager;
+import org.apache.synapse.eventing.filters.XPathBasedEventFilter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.jaxen.JaxenException;
 
 import java.util.LinkedList;
 import java.util.List;
@@ -34,7 +40,11 @@
  */
 public class DefaultInMemorySubscriptionManager extends SubscriptionManager {
 
-    Map<String, Subscription> store = new ConcurrentHashMap<String, Subscription>();
+    private Map<String, Subscription> store = new ConcurrentHashMap<String, Subscription>();
+    private String topicHeaderName;
+    private String topicHeaderNS;
+    private SynapseXPath topicXPath;
+    private static final Log log = LogFactory.getLog(DefaultInMemorySubscriptionManager.class);
 
     public String addSubscription(Subscription subs) {
 		if (subs.getId() == null) {
@@ -64,7 +74,8 @@
     public List<Subscription> getMatchingSubscribers(MessageContext mc) {
 		LinkedList<Subscription> list = new LinkedList<Subscription>();
         for (Map.Entry<String, Subscription> stringSubscriptionEntry : store.entrySet()) {
-            EventFilter filter = stringSubscriptionEntry.getValue().getFilter();
+            XPathBasedEventFilter filter =(XPathBasedEventFilter) stringSubscriptionEntry.getValue().getFilter();
+            filter.setSourceXpath(topicXPath);
             if (filter == null || filter.isSatisfied(mc)) {
                 list.add(stringSubscriptionEntry.getValue());
             }
@@ -77,6 +88,39 @@
 	}
 
     public void init() {
-        // nothing to do at the inmemory subscriber initialization
+        try {
+            topicXPath = new SynapseXPath("s11:Header/ns:" + topicHeaderName + " | s12:Header/ns:" + topicHeaderName);
+            topicXPath.addNamespace("s11", "http://schemas.xmlsoap.org/soap/envelope/");
+            topicXPath.addNamespace("s12", "http://www.w3.org/2003/05/soap-envelope");
+            topicXPath.addNamespace("ns", topicHeaderNS);
+        } catch (JaxenException e) {
+            handleException("Unable to create the topic header XPath", e);
+        }
+
+
+    }
+    public String getTopicHeaderName() {
+        return topicHeaderName;
+    }
+
+    public void setTopicHeaderName(String topicHeaderName) {
+        this.topicHeaderName = topicHeaderName;
+    }
+
+    public String getTopicHeaderNS() {
+        return topicHeaderNS;
+    }
+
+    public void setTopicHeaderNS(String topicHeaderNS) {
+        this.topicHeaderNS = topicHeaderNS;
+    }    
+    private void handleException(String message) {
+        log.error(message);
+        throw new SynapseException(message);
+    }
+
+    private void handleException(String message, Exception e) {
+        log.error(message, e);
+        throw new SynapseException(message, e);
     }
 }