executor, selector);
executor);
	public ConsumerDelegate createConsumerDelegate(JBossDestination jmsDestination,                                                  String selectorString,                                                  boolean noLocal,                                                  String subscriptionName,                                                  boolean isCC) throws JMSException   {      try      {         if (closed)         {            throw new IllegalStateException("Session is closed");         }                  if ("".equals(selectorString))         {            selectorString = null;         }                  log.debug("creating consumer for " + jmsDestination + ", selector " + selectorString + ", " + (noLocal ? "noLocal, " : "") + "subscription " + subscriptionName);            ManagedDestination mDest = dm.getDestination(jmsDestination.getName(), jmsDestination.isQueue());                  if (mDest == null)         {            throw new InvalidDestinationException("No such destination: " + jmsDestination);         }                  if (jmsDestination.isTemporary())         {            // Can only create a consumer for a temporary destination on the same connection            // that created it            if (!connectionEndpoint.hasTemporaryDestination(jmsDestination))            {               String msg = "Cannot create a message consumer on a different connection " +                            "to that which created the temporary destination";               throw new IllegalStateException(msg);            }         }                                int consumerID = connectionEndpoint.getServerPeer().getNextObjectID();                 Binding binding = null;                  //Always validate the selector first         Selector selector = null;         if (selectorString != null)         {            selector = new Selector(selectorString);         }            if (jmsDestination.isTopic())         {            if (subscriptionName == null)            {               // non-durable subscription               if (log.isTraceEnabled()) { log.trace("creating new non-durable subscription on " + jmsDestination); }                                    //Create the sub               QueuedExecutor executor = (QueuedExecutor)pool.get();               Queue q =                   new Queue(idm.getId(), ms, pm, true, false,                           mDest.getFullSize(),                           mDest.getPageSize(),                           mDest.getDownCacheSize(),                           executor, selector);                              //Make a binding for this queue               binding = topicPostOffice.bindQueue(new GUID().toString(), jmsDestination.getName(), q);                           }            else            {               if (jmsDestination.isTemporary())               {                  throw new InvalidDestinationException("Cannot create a durable subscription on a temporary topic");               }                              // we have a durable subscription, look it up               String clientID = connectionEndpoint.getClientID();               if (clientID == null)               {                  throw new JMSException("Cannot create durable subscriber without a valid client ID");               }                              //See if there any bindings with the same client_id.subscription_name name                              String name = MessageQueueNameHelper.createSubscriptionName(clientID, subscriptionName);                              binding = topicPostOffice.getBindingForQueueName(name);                                 if (binding == null)               {                  //Does not already exist                                    if (trace) { log.trace("creating new durable subscription on " + jmsDestination); }                                    QueuedExecutor executor = (QueuedExecutor)pool.get();                  Queue q =                      new Queue(idm.getId(), ms, pm, true, true,                              mDest.getFullSize(),                              mDest.getPageSize(),                              mDest.getDownCacheSize(),                              executor, selector);                                    //Make a binding for this queue                  binding = topicPostOffice.bindQueue(name, jmsDestination.getName(), q);                              }               else               {                  //Durable sub already exists                                    if (trace) { log.trace("subscription " + subscriptionName + " already exists"); }                                    // From javax.jms.Session Javadoc (and also JMS 1.1 6.11.1):                  // A client can change an existing durable subscription by creating a durable                  // TopicSubscriber with the same name and a new topic and/or message selector.                  // Changing a durable subscriber is equivalent to unsubscribing (deleting) the old                  // one and creating a new one.                     boolean selectorChanged =                     (selectorString == null && binding.getSelector() != null) ||                     (binding.getSelector() == null && selectorString != null) ||                     (binding.getSelector() != null && selectorString != null &&                     !binding.getSelector().equals(selectorString));                                    if (trace) { log.trace("selector " + (selectorChanged ? "has" : "has NOT") + " changed"); }                     boolean topicChanged = !binding.getCondition().equals(jmsDestination.getName());                                    if (log.isTraceEnabled()) { log.trace("topic " + (topicChanged ? "has" : "has NOT") + " changed"); }                                    if (selectorChanged || topicChanged)                  {                         if (trace) { log.trace("topic or selector changed so deleting old subscription"); }                        // Unbind the durable subscription                                          topicPostOffice.unbindQueue(name);                        // create a fresh new subscription                                          QueuedExecutor executor = (QueuedExecutor)pool.get();                     Queue q =                         new Queue(idm.getId(), ms, pm, true, true,                                 mDest.getFullSize(),                                 mDest.getPageSize(),                                 mDest.getDownCacheSize(),                                 executor, selector);                                          //Make a binding for this queue                     binding = topicPostOffice.bindQueue(name, jmsDestination.getName(), q);                    }                              }            }         }         else         {            //Consumer on a jms queue                        //Let's find the binding            binding = queuePostOffice.getBindingForQueueName(jmsDestination.getName());                        if (binding == null)            {               throw new IllegalStateException("Cannot find binding for jms queue: " + jmsDestination.getName());            }         }                  int prefetchSize = connectionEndpoint.getPrefetchSize();                  ServerConsumerEndpoint ep =            new ServerConsumerEndpoint(consumerID, binding.getQueue(), binding.getQueueName(),                                       this, selectorString, noLocal, jmsDestination, prefetchSize);                   JMSDispatcher.instance.registerTarget(new Integer(consumerID), new ConsumerAdvised(ep));                              ClientConsumerDelegate stub = new ClientConsumerDelegate(consumerID, prefetchSize);                                putConsumerEndpoint(consumerID, ep); // caching consumer locally                  connectionEndpoint.getServerPeer().putConsumerEndpoint(consumerID, ep); // cachin consumer in server peer                  log.debug("created and registered " + ep);            return stub;      }      catch (Throwable t)      {         throw ExceptionUtil.handleJMSInvocation(t, this + " createConsumerDelegate");      }   }