String subscriptionName)
String subscriptionName, boolean isCC)
	public ConsumerDelegate createConsumerDelegate(Destination jmsDestination,                                                  String selector,                                                  boolean noLocal,                                                  String subscriptionName)		throws JMSException   {      if ("".equals(selector))      {         selector = null;      }            JBossDestination d = (JBossDestination)jmsDestination;            if (log.isTraceEnabled()) { log.trace("Creating ServerConsumerDelegate for dest:" + d.getName() +             " selector:" + selector + " noLocal:" + noLocal + " subName:" + subscriptionName); }                  if (d.isTemporary())      {         //Can only create a consumer for a temporary destination on the same connection         //that created it         if (!this.connectionEndpoint.temporaryDestinations.contains(d))         {            throw new IllegalStateException("Cannot create a message consumer " +                  "on a different connection to that which created the temporary destination");         }      }            // look-up destination      DestinationManagerImpl dm = serverPeer.getDestinationManager();          Channel destination = dm.getCoreDestination(jmsDestination);           // create the MessageConsumer dynamic proxy      String stackName = "ConsumerStack";      AdviceStack stack = AspectManager.instance().getAdviceStack(stackName);      // TODO why do I need to the advisor to create the interceptor stack?      Interceptor[] interceptors = stack.createInterceptors(serverPeer.getSessionAdvisor(), null);      JMSConsumerInvocationHandler h = new JMSConsumerInvocationHandler(interceptors);      String consumerID = generateConsumerID();      SimpleMetaData metadata = new SimpleMetaData();      metadata.addMetaData(Dispatcher.DISPATCHER, Dispatcher.OID, serverPeer.getConsumerAdvisor().getName(), PayloadKey.AS_IS);      metadata.addMetaData(InvokerInterceptor.REMOTING,                           InvokerInterceptor.INVOKER_LOCATOR,                           serverPeer.getLocator(),                           PayloadKey.AS_IS);      metadata.addMetaData(InvokerInterceptor.REMOTING,                           InvokerInterceptor.SUBSYSTEM,                           "JMS",                           PayloadKey.AS_IS);      // TODO: Is this really necessary? Can't I just use the consumerID?      metadata.addMetaData(JMSAdvisor.JMS, JMSAdvisor.CONNECTION_ID, connectionEndpoint.getConnectionID(), PayloadKey.AS_IS);      metadata.addMetaData(JMSAdvisor.JMS, JMSAdvisor.SESSION_ID, sessionID, PayloadKey.AS_IS);      metadata.addMetaData(JMSAdvisor.JMS, JMSAdvisor.CONSUMER_ID, consumerID, PayloadKey.AS_IS);      h.getMetaData().mergeIn(metadata);      // TODO      ClassLoader loader = getClass().getClassLoader();      Class[] interfaces = new Class[] { ConsumerDelegate.class };      ConsumerDelegate delegate = (ConsumerDelegate)Proxy.newProxyInstance(loader, interfaces, h);      if (callbackHandler == null)      {         throw new JMSException("null callback handler");      }                     DurableSubscriptionHolder subscription = null;      if (subscriptionName != null)      {         String clientID = connectionEndpoint.getClientID();         if (clientID == null)         {            throw new JMSException("Cannot create durable subscriber without having set client ID");         }                           //It's a durable subscription - have we already got one with that name?         ClientManager clientManager = serverPeer.getClientManager();                             subscription = clientManager.getDurableSubscription(clientID, subscriptionName);                                          if (subscription != null)         {                        if (log.isTraceEnabled()) { log.trace("Subscription with name " + subscriptionName +                                       " already exists"); }                                                                        /* From javax.jms.Session Javadoc:             *              * 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.             */            //Has the selector changed?            boolean selectorChanged = (selector == null && subscription.getSelector() != null) ||                      (subscription.getSelector() == null && selector != null) ||                      (subscription.getSelector() != null && selector != null &&                        !subscription.getSelector().equals(selector));                        if (log.isTraceEnabled()) { log.trace("Selector has changed? " + selectorChanged); }                        //Has the Topic changed?                           boolean topicChanged = subscription.getTopic() != destination;                        if (log.isTraceEnabled()) { log.trace("Topic has changed? " + topicChanged); }                       if (selectorChanged || topicChanged)            {               if (log.isTraceEnabled()) { log.trace("Changed so deleting old subscription"); }                              DurableSubscriptionHolder removed = this.serverPeer.getClientManager().                  removeDurableSubscription(this.connectionEndpoint.clientID, subscriptionName);               if (removed == null)               {                  throw new InvalidDestinationException("Cannot find durable subscription with name " +                     subscriptionName + " to unsubscribe");               }               subscription.getTopic().remove(subscription.getQueue());               subscription = null;            }         }                  if (subscription == null)         {            if (!(destination instanceof org.jboss.messaging.core.local.Topic))            {               throw new JMSException("Can only create a durable subscription on a topic");            }                        if (log.isTraceEnabled()) { log.trace("Creating new durable subscription on topic " + destination); }                        Topic topic = (Topic)destination;            MessageStore ms = connectionEndpoint.getServerPeer().getMessageStore();            TransactionManager tm = connectionEndpoint.getServerPeer().getTransactionManager();            subscription = new DurableSubscriptionHolder(subscriptionName, topic,                                                         new Queue(consumerID, ms, tm),                                                         selector);            clientManager.addDurableSubscription(clientID, subscriptionName, subscription);            //start it            destination.add(subscription.getQueue());         }      }            ServerConsumerDelegate scd =         new ServerConsumerDelegate(consumerID,                                    subscriptionName == null ? destination : subscription.getQueue(),                                    callbackHandler, this, selector, noLocal);                  putConsumerDelegate(consumerID, scd);      connectionEndpoint.receivers.put(consumerID, scd);      if (log.isTraceEnabled()) log.trace("created consumer endpoint (destination=" + jmsDestination + ")");      return delegate;   }