QPID-2858: Implement FilterManager for 0-10 subscriptions

Implementation of review changes.

Patch from <sorins@gmail.com>


git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@998891 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 13 insertions, 14 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
index a7f49d0566..dac517150a 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/filter/FilterManagerFactory.java
@@ -20,15 +20,18 @@
  */
 package org.apache.qpid.server.filter;
 
+import java.util.Map;
+
 import org.apache.qpid.AMQException;
 import org.apache.qpid.common.AMQPFilterTypes;
 import org.apache.qpid.framing.FieldTable;
+import org.apache.log4j.Logger;
 
 
 public class FilterManagerFactory
 {
-    //private final static Logger _logger = LoggerFactory.getLogger(FilterManagerFactory.class);
-    private final static org.apache.log4j.Logger _logger = org.apache.log4j.Logger.getLogger(FilterManagerFactory.class);
+ 
+    private final static Logger _logger = Logger.getLogger(FilterManagerFactory.class);
 
     //fixme move to a common class so it can be refered to from client code.
 
@@ -64,4 +67,9 @@ public class FilterManagerFactory
         return manager;
 
     }
+    
+    public static FilterManager createManager(Map<String,Object> map) throws AMQException
+    {
+        return createManager(FieldTable.convertToFieldTable(map));
+    }
 }
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
index d57424dd32..53dad0fcfe 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/transport/ServerSessionDelegate.java
@@ -230,20 +230,11 @@ public class ServerSessionDelegate extends SessionDelegate
                     }
 
                     FlowCreditManager_0_10 creditManager = new WindowCreditManager(0L,0L);
-
-                    FieldTable filters = new FieldTable();
-                    Map<String,Object> fields = method.getFields();
-                    for (String key: fields.keySet()) 
-                    {
-                        if (key.equals(AMQPFilterTypes.JMS_SELECTOR.getValue().asString()))
-                        { 
-                            filters.setObject(key, fields.get(key));
-                        }
-                    }
+                    
                     FilterManager filterManager = null;
                     try 
                     {
-                        filterManager = FilterManagerFactory.createManager(filters);
+                        filterManager = FilterManagerFactory.createManager(method.getArguments());
                     } 
                     catch (AMQException amqe) 
                     {
@@ -256,7 +247,7 @@ public class ServerSessionDelegate extends SessionDelegate
                                                                   method.getAcceptMode(),
                                                                   method.getAcquireMode(),
                                                                   MessageFlowMode.WINDOW,
-                                                                  creditManager, null);
+                                                                  creditManager, filterManager);
 
                     ((ServerSession)session).register(destination, sub);
                     try