content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.activemq.artemis.jms.client;
  
  import static org.apache.activemq.artemis.api.core.ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST;
  
  import javax.jms.BytesMessage;
  import javax.jms.Destination;
  import javax.jms.IllegalStateException;
  import javax.jms.InvalidDestinationException;
  import javax.jms.JMSException;
  import javax.jms.MapMessage;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageListener;
  import javax.jms.MessageProducer;
  import javax.jms.ObjectMessage;
  import javax.jms.Queue;
  import javax.jms.QueueBrowser;
  import javax.jms.QueueReceiver;
  import javax.jms.QueueSender;
  import javax.jms.QueueSession;
  import javax.jms.Session;
  import javax.jms.StreamMessage;
  import javax.jms.TemporaryQueue;
  import javax.jms.TemporaryTopic;
  import javax.jms.TextMessage;
  import javax.jms.Topic;
  import javax.jms.TopicPublisher;
  import javax.jms.TopicSession;
  import javax.jms.TopicSubscriber;
  import javax.jms.TransactionInProgressException;
  import javax.transaction.xa.XAResource;
  import java.io.Serializable;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.UUID;
  import java.util.concurrent.ConcurrentHashMap;
  
  import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
  import org.apache.activemq.artemis.api.core.ActiveMQException;
  import org.apache.activemq.artemis.api.core.ActiveMQQueueExistsException;
  import org.apache.activemq.artemis.api.core.QueueAttributes;
  import org.apache.activemq.artemis.api.core.QueueConfiguration;
  import org.apache.activemq.artemis.api.core.SimpleString;
  import org.apache.activemq.artemis.api.core.client.ClientConsumer;
  import org.apache.activemq.artemis.api.core.client.ClientProducer;
  import org.apache.activemq.artemis.api.core.client.ClientSession;
  import org.apache.activemq.artemis.api.core.client.ClientSession.AddressQuery;
  import org.apache.activemq.artemis.api.core.client.ClientSession.QueueQuery;
  import org.apache.activemq.artemis.api.core.RoutingType;
  import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQBytesCompatibleMessage;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQCompatibleMessage;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQMapCompatibleMessage;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQObjectCompatibleMessage;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQStreamCompatibleMessage;
  import org.apache.activemq.artemis.jms.client.compatible1X.ActiveMQTextCompatibleMessage;
  import org.apache.activemq.artemis.selector.filter.FilterException;
  import org.apache.activemq.artemis.selector.impl.SelectorParser;
  import org.apache.activemq.artemis.utils.AutoCreateUtil;
  import org.apache.activemq.artemis.utils.CompositeAddress;
  import org.apache.activemq.artemis.utils.SelectorTranslator;
  
  /**
   * ActiveMQ Artemis implementation of a JMS {@link Session}.
   * <p>
   * Note that we *do not* support JMS ASF (Application Server Facilities) optional constructs such as
   * {@code ConnectionConsumer}
   */
  public class ActiveMQSession implements QueueSession, TopicSession {
  
     public static final int TYPE_GENERIC_SESSION = 0;
  
     public static final int TYPE_QUEUE_SESSION = 1;
  
     public static final int TYPE_TOPIC_SESSION = 2;
  
     private static SimpleString REJECTING_FILTER = SimpleString.of("_AMQX=-1");
  
     private final ConnectionFactoryOptions options;
  
     private final ActiveMQConnection connection;
  
     private final ClientSession session;
  
     private final int sessionType;
  
     private final int ackMode;
  
     private final boolean transacted;
  
     private final boolean xa;
  
     private boolean recoverCalled;
  
     private final Set<ActiveMQMessageConsumer> consumers = new HashSet<>();
  
     private final boolean cacheDestination;
  
     private final boolean enable1xPrefixes;
  
     private final Map<String, Topic> topicCache = new ConcurrentHashMap<>();
  
     private final Map<String, Queue> queueCache = new ConcurrentHashMap<>();
  
  
     protected ActiveMQSession(final ConnectionFactoryOptions options,
                               final ActiveMQConnection connection,
                               final boolean transacted,
                               final boolean xa,
                               final int ackMode,
                               final boolean cacheDestination,
                               final boolean enable1xPrefixes,
                               final ClientSession session,
                               final int sessionType) {
        this.options = options;
  
        this.connection = connection;
  
        this.ackMode = ackMode;
  
        this.session = session;
  
        this.sessionType = sessionType;
  
        this.transacted = transacted;
  
        this.xa = xa;
  
        this.cacheDestination = cacheDestination;
  
        this.enable1xPrefixes = enable1xPrefixes;
     }
  
     // Session implementation ----------------------------------------
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public BytesMessage createBytesMessage() throws JMSException {
        checkClosed();
  
        ActiveMQBytesMessage message;
        if (enable1xPrefixes) {
           message = new ActiveMQBytesCompatibleMessage(session);
        } else {
           message = new ActiveMQBytesMessage(session);
        }
        return message;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MapMessage createMapMessage() throws JMSException {
        checkClosed();
  
        ActiveMQMapMessage message;
        if (enable1xPrefixes) {
           message = new ActiveMQMapCompatibleMessage(session);
        } else {
           message = new ActiveMQMapMessage(session);
        }
        return message;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public Message createMessage() throws JMSException {
        checkClosed();
  
        ActiveMQMessage message;
        if (enable1xPrefixes) {
           message = new ActiveMQCompatibleMessage(session);
        } else {
           message = new ActiveMQMessage(session);
        }
        return message;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public ObjectMessage createObjectMessage() throws JMSException {
        checkClosed();
  
        ActiveMQObjectMessage message;
        if (enable1xPrefixes) {
           message = new ActiveMQObjectCompatibleMessage(session, options);
        } else {
           message = new ActiveMQObjectMessage(session, options);
        }
        return message;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public ObjectMessage createObjectMessage(final Serializable object) throws JMSException {
        checkClosed();
  
        ActiveMQObjectMessage msg;
        if (enable1xPrefixes) {
           msg = new ActiveMQObjectCompatibleMessage(session, options);
        } else {
           msg = new ActiveMQObjectMessage(session, options);
        }
        msg.setObject(object);
  
        return msg;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public StreamMessage createStreamMessage() throws JMSException {
        checkClosed();
  
        ActiveMQStreamMessage message;
        if (enable1xPrefixes) {
           message = new ActiveMQStreamCompatibleMessage(session);
        } else {
           message = new ActiveMQStreamMessage(session);
        }
        return message;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TextMessage createTextMessage() throws JMSException {
        checkClosed();
  
        ActiveMQTextMessage msg;
        if (enable1xPrefixes) {
           msg = new ActiveMQTextCompatibleMessage(session);
        } else {
           msg = new ActiveMQTextMessage(session);
        }
        msg.setText(null);
  
        return msg;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TextMessage createTextMessage(final String text) throws JMSException {
        checkClosed();
  
        ActiveMQTextMessage msg;
        if (enable1xPrefixes) {
           msg = new ActiveMQTextCompatibleMessage(session);
        } else {
           msg = new ActiveMQTextMessage(session);
        }
        msg.setText(text);
  
        return msg;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public boolean getTransacted() throws JMSException {
        checkClosed();
  
        return transacted;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public int getAcknowledgeMode() throws JMSException {
        checkClosed();
  
        return ackMode;
     }
  
     public boolean isXA() {
        return xa;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void commit() throws JMSException {
        if (!transacted) {
           throw new IllegalStateException("Cannot commit a non-transacted session");
        }
        if (xa) {
           throw new TransactionInProgressException("Cannot call commit on an XA session");
        }
        try {
           session.commit();
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void rollback() throws JMSException {
        if (!transacted) {
           throw new IllegalStateException("Cannot rollback a non-transacted session");
        }
        if (xa) {
           throw new TransactionInProgressException("Cannot call rollback on an XA session");
        }
  
        try {
           session.rollback();
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void close() throws JMSException {
        connection.getThreadAwareContext().assertNotCompletionListenerThread();
        connection.getThreadAwareContext().assertNotMessageListenerThread();
        synchronized (connection) {
           try {
              for (ActiveMQMessageConsumer cons : new HashSet<>(consumers)) {
                 cons.close();
              }
  
              session.close();
  
              connection.removeSession(this);
           } catch (ActiveMQException e) {
              throw JMSExceptionHelper.convertFromActiveMQException(e);
           }
        }
        topicCache.clear();
        queueCache.clear();
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void recover() throws JMSException {
        if (transacted) {
           throw new IllegalStateException("Cannot recover a transacted session");
        }
  
        try {
           session.rollback(true);
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
  
        recoverCalled = true;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageListener getMessageListener() throws JMSException {
        checkClosed();
  
        return null;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void setMessageListener(final MessageListener listener) throws JMSException {
        checkClosed();
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void run() {
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageProducer createProducer(final Destination destination) throws JMSException {
        if (destination != null && !(destination instanceof ActiveMQDestination)) {
           throw new InvalidDestinationException("Not an ActiveMQ Artemis Destination:" + destination);
        }
  
        try {
           ActiveMQDestination jbd = (ActiveMQDestination) destination;
  
           if (jbd != null) {
              checkDestination(jbd);
           }
  
           ClientProducer producer = session.createProducer(jbd == null ? null : jbd.getSimpleAddress());
  
           return new ActiveMQMessageProducer(connection, producer, jbd, this, options);
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     void checkDestination(ActiveMQDestination destination) throws JMSException {
        SimpleString address = destination.getSimpleAddress();
        if (!destination.isCreated()) {
           try {
              ClientSession.AddressQuery addressQuery = session.addressQuery(address);
  
              // First we create the address
              if (!addressQuery.isExists()) {
                 if (destination.isQueue()) {
                    if (addressQuery.isAutoCreateAddresses() && addressQuery.isAutoCreateQueues()) {
                       session.createAddress(address, RoutingType.ANYCAST, true);
                    } else {
                       throw new InvalidDestinationException("Destination " + address + " does not exist, autoCreateAddresses=" + addressQuery.isAutoCreateAddresses() + " , autoCreateQueues=" + addressQuery.isAutoCreateQueues());
                    }
                 } else {
                    if (addressQuery.isAutoCreateAddresses()) {
                       session.createAddress(address, RoutingType.MULTICAST, true);
                    } else {
                       throw new InvalidDestinationException("Destination " + address + " does not exist, autoCreateAddresses=" + addressQuery.isAutoCreateAddresses());
                    }
                 }
              } else {
                 if ((!addressQuery.isSupportsMulticast() && !destination.isQueue()) || (!addressQuery.isSupportsAnycast() && destination.isQueue())) {
                    throw new InvalidDestinationException("Destination " + address + " exists, but does not support " + (destination.isQueue() ? RoutingType.ANYCAST.name() : RoutingType.MULTICAST.name()) + " routing");
                 }
              }
  
              // Second we create the queue, the address would have existed or successfully created.
              if (destination.isQueue()) {
                 ClientSession.QueueQuery queueQuery = session.queueQuery(address);
                 if (!queueQuery.isExists()) {
                    if (addressQuery.isAutoCreateQueues()) {
                       if (destination.isTemporary()) {
                          createTemporaryQueue(destination, RoutingType.ANYCAST, address, null, addressQuery);
                       } else {
                          createQueue(destination, RoutingType.ANYCAST, address, null, true, true, addressQuery);
                       }
                    } else {
                       throw new InvalidDestinationException("Destination " + address + " does not exist, address exists but autoCreateQueues=" + addressQuery.isAutoCreateQueues());
                    }
                 }
              } else if (CompositeAddress.isFullyQualified(address)) { // it could be a topic using FQQN
                 ClientSession.QueueQuery queueQuery = session.queueQuery(address);
                 if (!queueQuery.isExists()) {
                    if (addressQuery.isAutoCreateQueues()) {
                       if (destination.isTemporary()) {
                          createTemporaryQueue(destination, RoutingType.MULTICAST, address, null, addressQuery);
                       } else {
                          createQueue(destination, RoutingType.MULTICAST, address, null, true, true, addressQuery);
                       }
                    } else {
                       throw new InvalidDestinationException("Destination " + address + " does not exist, address exists but autoCreateQueues=" + addressQuery.isAutoCreateQueues());
                    }
                 }
              }
           } catch (ActiveMQQueueExistsException thatsOK) {
              // nothing to be done
           } catch (ActiveMQException e) {
              throw JMSExceptionHelper.convertFromActiveMQException(e);
           }
           // this is done at the end, if no exceptions are thrown
           destination.setCreated(true);
        }
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createConsumer(final Destination destination) throws JMSException {
        return createConsumer(destination, null, false);
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createConsumer(final Destination destination,
                                           final String messageSelector) throws JMSException {
        return createConsumer(destination, messageSelector, false);
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createConsumer(final Destination destination,
                                           final String messageSelector,
                                           final boolean noLocal) throws JMSException {
        if (destination == null) {
           throw new InvalidDestinationException("Cannot create a consumer with a null destination");
        }
  
        if (!(destination instanceof ActiveMQDestination jbdest)) {
           throw new InvalidDestinationException("Not an ActiveMQDestination:" + destination);
        }
  
        if (jbdest.isTemporary() && !connection.containsTemporaryQueue(jbdest.getSimpleAddress())) {
           throw new JMSException("Can not create consumer for temporary destination " + destination +
                                     " from another JMS connection");
        }
  
        return createConsumer(jbdest, null, messageSelector, noLocal, ConsumerDurability.NON_DURABLE);
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public Queue createQueue(final String queueName) throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
           throw new IllegalStateException("Cannot create a queue using a TopicSession");
        }
  
        try {
           Queue queue = null;
           if (cacheDestination) {
              queue = queueCache.get(queueName);
           }
           if (queue == null) {
              queue = internalCreateQueue(queueName);
           }
           if (cacheDestination) {
              queueCache.put(queueName, queue);
           }
           return queue;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
     protected Queue internalCreateQueue(String queueName) throws ActiveMQException, JMSException {
        ActiveMQQueue queue = lookupQueue(queueName, false);
  
        if (queue == null) {
           queue = lookupQueue(queueName, true);
        }
  
        if (queue == null) {
           queue = internalCreateQueueCompatibility("jms.queue." + queueName);
        }
        if (queue == null) {
           throw new JMSException("There is no queue with name " + queueName);
        } else {
           return queue;
        }
     }
  
     protected ActiveMQQueue internalCreateQueueCompatibility(String queueName) throws ActiveMQException, JMSException {
        ActiveMQQueue queue = lookupQueue(queueName, false);
  
        if (queue == null) {
           queue = lookupQueue(queueName, true);
        }
  
        return queue;
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public Topic createTopic(final String topicName) throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a topic on a QueueSession");
        }
        try {
           Topic topic = null;
           if (cacheDestination) {
              topic = topicCache.get(topicName);
           }
           if (topic == null) {
              topic = internalCreateTopic(topicName, false);
           }
           if (cacheDestination) {
              topicCache.put(topicName, topic);
           }
           return topic;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     protected Topic internalCreateTopic(String topicName, boolean retry) throws ActiveMQException, JMSException {
        ActiveMQTopic topic = lookupTopic(topicName, false);
  
        if (topic == null) {
           topic = lookupTopic(topicName, true);
        }
  
        if (topic == null) {
           if (!retry) {
              return internalCreateTopic("jms.topic." + topicName, true);
           }
           throw new JMSException("There is no topic with name " + topicName);
        } else {
           return topic;
        }
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TopicSubscriber createDurableSubscriber(final Topic topic, final String name) throws JMSException {
        return createDurableSubscriber(topic, name, null, false);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TopicSubscriber createDurableSubscriber(final Topic topic,
                                                    final String name,
                                                    String messageSelector,
                                                    final boolean noLocal) throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a durable subscriber on a QueueSession");
        }
        checkTopic(topic);
        if (!(topic instanceof ActiveMQDestination jbdest)) {
           throw new InvalidDestinationException("Not an ActiveMQTopic:" + topic);
        }
        if ("".equals(messageSelector)) {
           messageSelector = null;
        }
  
        if (jbdest.isQueue()) {
           throw new InvalidDestinationException("Cannot create a subscriber on a queue");
        }
  
        return createConsumer(jbdest, name, messageSelector, noLocal, ConsumerDurability.DURABLE);
     }
  
     private void checkTopic(Topic topic) throws InvalidDestinationException {
        if (topic == null) {
           throw ActiveMQJMSClientBundle.BUNDLE.nullTopic();
        }
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) throws JMSException {
        return createSharedConsumer(topic, sharedSubscriptionName, null);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createSharedConsumer(Topic topic, String name, String messageSelector) throws JMSException {
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a shared consumer on a QueueSession");
        }
        checkTopic(topic);
        ActiveMQTopic localTopic;
        if (topic instanceof ActiveMQTopic activeMQTopic) {
           localTopic = activeMQTopic;
        } else {
           localTopic = new ActiveMQTopic(topic.getTopicName());
        }
        return internalCreateSharedConsumer(localTopic, name, messageSelector, ConsumerDurability.NON_DURABLE);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createDurableConsumer(Topic topic, String name) throws JMSException {
        return createDurableConsumer(topic, name, null, false);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createDurableConsumer(Topic topic,
                                                  String name,
                                                  String messageSelector,
                                                  boolean noLocal) throws JMSException {
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a durable consumer on a QueueSession");
        }
        checkTopic(topic);
        ActiveMQTopic localTopic;
        if (topic instanceof ActiveMQTopic activeMQTopic) {
           localTopic = activeMQTopic;
        } else {
           localTopic = new ActiveMQTopic(topic.getTopicName());
        }
        return createConsumer(localTopic, name, messageSelector, noLocal, ConsumerDurability.DURABLE);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createSharedDurableConsumer(Topic topic, String name) throws JMSException {
        return createSharedDurableConsumer(topic, name, null);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public MessageConsumer createSharedDurableConsumer(Topic topic,
                                                        String name,
                                                        String messageSelector) throws JMSException {
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a shared durable consumer on a QueueSession");
        }
  
        checkTopic(topic);
  
        ActiveMQTopic localTopic;
  
        if (topic instanceof ActiveMQTopic activeMQTopic) {
           localTopic = activeMQTopic;
        } else {
           localTopic = new ActiveMQTopic(topic.getTopicName());
        }
        return internalCreateSharedConsumer(localTopic, name, messageSelector, ConsumerDurability.DURABLE);
     }
  
     @Deprecated(forRemoval = true)
     public String getDeserializationBlackList() {
        return getDeserializationDenyList();
     }
  
     @Deprecated(forRemoval = true)
     public String getDeserializationWhiteList() {
        return getDeserializationAllowList();
     }
  
     public String getDeserializationDenyList() {
        return connection.getDeserializationDenyList();
     }
  
     public String getDeserializationAllowList() {
        return connection.getDeserializationAllowList();
     }
  
     enum ConsumerDurability {
        DURABLE, NON_DURABLE;
     }
  
     private ActiveMQMessageConsumer internalCreateSharedConsumer(final ActiveMQDestination dest,
                                                                  final String subscriptionName,
                                                                  String selectorString,
                                                                  ConsumerDurability durability) throws JMSException {
        try {
  
           if (dest.isQueue()) {
              // This is not really possible unless someone makes a mistake on code
              // createSharedConsumer only accepts Topics by declaration
              throw new RuntimeException("Internal error: createSharedConsumer is only meant for Topics");
           }
  
           if (subscriptionName == null) {
              throw ActiveMQJMSClientBundle.BUNDLE.invalidSubscriptionName();
           }
  
           selectorString = "".equals(selectorString) ? null : selectorString;
  
           SimpleString coreFilterString = null;
  
           if (selectorString != null) {
              coreFilterString = SimpleString.of(SelectorTranslator.convertToActiveMQFilterString(selectorString));
           }
  
           ClientConsumer consumer;
  
           SimpleString autoDeleteQueueName = null;
  
           AddressQuery response = session.addressQuery(dest.getSimpleAddress());
  
           if (!response.isExists() && !response.isAutoCreateAddresses()) {
              throw ActiveMQJMSClientBundle.BUNDLE.destinationDoesNotExist(dest.getSimpleAddress());
           }
  
           SimpleString queueName;
  
           if (dest.isTemporary() && durability == ConsumerDurability.DURABLE) {
              throw new InvalidDestinationException("Cannot create a durable subscription on a temporary topic");
           }
  
           queueName = ActiveMQDestination.createQueueNameForSubscription(durability == ConsumerDurability.DURABLE, connection.getClientID(), subscriptionName);
  
           QueueQuery subResponse = session.queueQuery(queueName);
  
           if ((!subResponse.isExists() || !Objects.equals(subResponse.getAddress(), dest.getSimpleAddress()) || !Objects.equals(subResponse.getFilterString(), coreFilterString)) && !Boolean.TRUE.equals(subResponse.isConfigurationManaged())) {
              try {
                 createSharedQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, durability == ConsumerDurability.DURABLE, response);
              } catch (ActiveMQQueueExistsException ignored) {
                 // We ignore this because querying and then creating the queue wouldn't be idempotent
                 // we could also add a parameter to ignore existence what would require a bigger work around to avoid
                 // compatibility.
              }
           }
  
           consumer = createClientConsumer(dest, queueName, null);
  
           ActiveMQMessageConsumer jbc = new ActiveMQMessageConsumer(options, connection, this, consumer, false, dest, selectorString, autoDeleteQueueName);
  
           consumers.add(jbc);
  
           return jbc;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     private ActiveMQMessageConsumer createConsumer(final ActiveMQDestination dest,
                                                    final String subscriptionName,
                                                    String selectorString,
                                                    final boolean noLocal,
                                                    ConsumerDurability durability) throws JMSException {
        try {
           selectorString = "".equals(selectorString) ? null : selectorString;
  
           if (noLocal) {
              connection.setHasNoLocal();
  
              String filter;
              if (connection.getClientID() != null) {
                 filter = ActiveMQConnection.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + connection.getClientID() +
                    "'";
              } else {
                 filter = ActiveMQConnection.CONNECTION_ID_PROPERTY_NAME.toString() + "<>'" + connection.getUID() + "'";
              }
  
              if (selectorString != null) {
                 selectorString += " AND " + filter;
              } else {
                 selectorString = filter;
              }
           }
  
           SimpleString coreFilterString = null;
  
           if (selectorString != null) {
              coreFilterString = SimpleString.of(SelectorTranslator.convertToActiveMQFilterString(selectorString));
           }
  
           ClientConsumer consumer;
  
           SimpleString autoDeleteQueueName = null;
  
           if (dest.isQueue()) {
              try {
                 AutoCreateUtil.autoCreateQueue(session, dest.getSimpleAddress(), null);
              } catch (ActiveMQException ex) {
                 if (ex.getType() == QUEUE_DOES_NOT_EXIST) {
                    throw new InvalidDestinationException("Destination " + dest.getName() + " does not exist");
                 }
                 throw ex;
              }
  
              dest.setCreated(true);
  
              consumer = createClientConsumer(dest, null, coreFilterString);
           } else {
              AddressQuery response = session.addressQuery(dest.getSimpleAddress());
  
              if (!response.isExists()) {
                 if (response.isAutoCreateAddresses()) {
                    session.createAddress(dest.getSimpleAddress(), RoutingType.MULTICAST, true);
                 } else {
                    throw new InvalidDestinationException("Topic " + dest.getName() + " does not exist");
                 }
              }
  
              dest.setCreated(true);
  
              SimpleString queueName;
  
              if (subscriptionName == null) {
                 if (durability != ConsumerDurability.NON_DURABLE)
                    throw new RuntimeException("Subscription name cannot be null for durable topic consumer");
                 // Non durable sub
  
  
                 if (CompositeAddress.isFullyQualified(dest.getAddress())) {
                    queueName = createFQQNSubscription(dest, coreFilterString, response);
                 } else {
                    queueName = SimpleString.of(UUID.randomUUID().toString());
                    createTemporaryQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, response);
                 }
  
                 consumer = createClientConsumer(dest, queueName, coreFilterString);
                 autoDeleteQueueName = queueName;
              } else {
                 // Durable sub
                 if (durability != ConsumerDurability.DURABLE)
                    throw new RuntimeException("Subscription name must be null for non-durable topic consumer");
                 if (connection.getClientID() == null) {
                    throw new IllegalStateException("Cannot create durable subscription - client ID has not been set");
                 }
  
                 if (dest.isTemporary()) {
                    throw new InvalidDestinationException("Cannot create a durable subscription on a temporary topic");
                 }
  
                 queueName = ActiveMQDestination.createQueueNameForSubscription(true, connection.getClientID(), subscriptionName);
  
                 QueueQuery subResponse = session.queueQuery(queueName);
  
                 if (!subResponse.isExists()) {
                    // durable subscription queues are not technically considered to be auto-created
                    createQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, true, false, response);
                 } else {
                    // Already exists
                    if (subResponse.getConsumerCount() > 0) {
                       throw new IllegalStateException("Cannot create a subscriber on the durable subscription since it already has subscriber(s)");
                    }
  
                    // 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.
  
                    SimpleString oldFilterString = subResponse.getFilterString();
  
                    boolean selectorChanged = coreFilterString == null && oldFilterString != null ||
                       oldFilterString == null && coreFilterString != null ||
                       oldFilterString != null &&
                          coreFilterString != null &&
                          !oldFilterString.equals(coreFilterString);
  
                    SimpleString oldTopicName = subResponse.getAddress();
  
                    boolean topicChanged = !oldTopicName.equals(dest.getSimpleAddress());
  
                    if ((selectorChanged || topicChanged) && !Boolean.TRUE.equals(subResponse.isConfigurationManaged())) {
                       // Delete the old durable sub
                       session.deleteQueue(queueName);
  
                       // Create the new one
                       createQueue(dest, RoutingType.MULTICAST, queueName, coreFilterString, true, false, response);
                    }
                 }
  
                 consumer = createClientConsumer(dest, queueName, null);
              }
           }
  
           ActiveMQMessageConsumer jbc = new ActiveMQMessageConsumer(options, connection, this, consumer, noLocal, dest, selectorString, autoDeleteQueueName);
  
           consumers.add(jbc);
  
           return jbc;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     // This method is for the actual queue creation on the Multicast queue / subscription
     private SimpleString createFQQNSubscription(ActiveMQDestination dest,
                                          SimpleString coreFilterString,
                                          AddressQuery response) throws ActiveMQException, JMSException {
        SimpleString queueName;
        queueName = CompositeAddress.extractQueueName(dest.getSimpleAddress());
        if (!response.isExists() || !response.getQueueNames().contains(AutoCreateUtil.getCoreQueueName(session, dest.getSimpleAddress()))) {
           if (response.isAutoCreateQueues()) {
              try {
                 createQueue(dest, RoutingType.MULTICAST, dest.getSimpleAddress(), coreFilterString, true, true, response);
                 return queueName;
              } catch (ActiveMQQueueExistsException e) {
                 // The queue was created by another client/admin between the query check and send create queue packet
                 // on this case we will switch to the regular verification to validate the coreFilterString
              }
           } else {
              throw new InvalidDestinationException("Destination " + dest.getName() + " does not exist");
           }
        }
  
        QueueQuery queueQuery = session.queueQuery(queueName);
  
        if (!queueQuery.isExists()) {
           throw new InvalidDestinationException("Destination " + queueName + " does not exist");
        }
  
        if (coreFilterString != null && queueQuery.getFilterString() != null && !coreFilterString.equals(queueQuery.getFilterString())) {
           throw new JMSException(queueName + " filter mismatch [" + coreFilterString + "] is different than existing filter [" + queueQuery.getFilterString() + "]");
        }
        return queueName;
     }
  
     private ClientConsumer createClientConsumer(ActiveMQDestination destination, SimpleString queueName, SimpleString coreFilterString) throws ActiveMQException {
        QueueAttributes queueAttributes = Objects.requireNonNullElseGet(destination.getQueueAttributes(), () -> new QueueAttributes());
        int priority = Objects.requireNonNullElse(queueAttributes.getConsumerPriority(), ActiveMQDefaultConfiguration.getDefaultConsumerPriority());
        return session.createConsumer(queueName == null ? destination.getSimpleAddress() : queueName, coreFilterString, priority, false);
     }
  
     public void ackAllConsumers() throws JMSException {
        checkClosed();
     }
  
  
     /**
      * {@inheritDoc}
      */
     @Override
     public QueueBrowser createBrowser(final Queue queue) throws JMSException {
        return createBrowser(queue, null);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public QueueBrowser createBrowser(final Queue queue, String filterString) throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
           throw new IllegalStateException("Cannot create a browser on a TopicSession");
        }
        if (queue == null) {
           throw new InvalidDestinationException("Cannot create a browser with a null queue");
        }
        if (!(queue instanceof ActiveMQDestination activeMQDestination)) {
           throw new InvalidDestinationException("Not an ActiveMQQueue:" + queue);
        }
        if ("".equals(filterString)) {
           filterString = null;
        }
  
        // eager test of the filter syntax as required by JMS spec
        try {
           if (filterString != null) {
              SelectorParser.parse(filterString.trim());
           }
        } catch (FilterException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(ActiveMQJMSClientBundle.BUNDLE.invalidFilter(SimpleString.of(filterString), e));
        }
  
        if (!activeMQDestination.isQueue()) {
           throw new InvalidDestinationException("Cannot create a browser on a topic");
        }
  
        try {
           AddressQuery response = session.addressQuery(SimpleString.of(activeMQDestination.getAddress()));
           if (!response.isExists()) {
              if (response.isAutoCreateQueues()) {
                 createQueue(activeMQDestination, RoutingType.ANYCAST, activeMQDestination.getSimpleAddress(), null, true, true, response);
              } else {
                 throw new InvalidDestinationException("Destination " + activeMQDestination.getName() + " does not exist");
              }
           }
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
  
        return new ActiveMQQueueBrowser(options, (ActiveMQQueue) activeMQDestination, filterString, session, enable1xPrefixes);
  
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TemporaryQueue createTemporaryQueue() throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
           throw new IllegalStateException("Cannot create a temporary queue using a TopicSession");
        }
  
        try {
           final ActiveMQTemporaryQueue queue;
           if (enable1xPrefixes) {
              queue  = ActiveMQDestination.createTemporaryQueue(this, PacketImpl.OLD_TEMP_QUEUE_PREFIX.toString());
           } else {
              queue  = ActiveMQDestination.createTemporaryQueue(this);
           }
  
           SimpleString simpleAddress = queue.getSimpleAddress();
  
           session.createQueue(QueueConfiguration.of(simpleAddress).setRoutingType(RoutingType.ANYCAST).setDurable(false).setTemporary(true));
  
           connection.addTemporaryQueue(simpleAddress);
  
           queue.setCreated(true);
  
           return queue;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TemporaryTopic createTemporaryTopic() throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot create a temporary topic on a QueueSession");
        }
  
        try {
           final ActiveMQTemporaryTopic topic;
           if (enable1xPrefixes) {
              topic  = ActiveMQDestination.createTemporaryTopic(this, PacketImpl.OLD_TEMP_TOPIC_PREFIX.toString());
           } else {
              topic  = ActiveMQDestination.createTemporaryTopic(this);
           }
  
           SimpleString simpleAddress = topic.getSimpleAddress();
  
           // We create a dummy subscription on the topic, that never receives messages - this is so we can perform JMS
           // checks when routing messages to a topic that
           // does not exist - otherwise we would not be able to distinguish from a non existent topic and one with no
           // subscriptions - core has no notion of a topic
  
           session.createQueue(QueueConfiguration.of(simpleAddress).setAddress(simpleAddress).setFilterString(ActiveMQSession.REJECTING_FILTER).setDurable(false).setTemporary(true));
  
           connection.addTemporaryQueue(simpleAddress);
  
           return topic;
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public void unsubscribe(final String name) throws JMSException {
        // As per spec. section 4.11
        if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
           throw new IllegalStateException("Cannot unsubscribe using a QueueSession");
        }
  
        SimpleString queueName = ActiveMQDestination.createQueueNameForSubscription(true, connection.getClientID(), name);
  
        try {
           QueueQuery response = session.queueQuery(queueName);
  
           if (!response.isExists()) {
              throw new InvalidDestinationException("Cannot unsubscribe, subscription with name " + name +
                                                       " does not exist");
           }
  
           if (response.getConsumerCount() != 0) {
              throw new IllegalStateException("Cannot unsubscribe durable subscription " + name +
                                                 " since it has active subscribers");
           }
  
           session.deleteQueue(queueName);
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     // XASession implementation
  
     public Session getSession() throws JMSException {
        if (!xa) {
           throw new IllegalStateException("Isn't an XASession");
        }
  
        return this;
     }
  
     public XAResource getXAResource() {
        return session.getXAResource();
     }
  
     // QueueSession implementation
  
     /**
      * {@inheritDoc}
      */
     @Override
     public QueueReceiver createReceiver(final Queue queue, final String messageSelector) throws JMSException {
        return (QueueReceiver) createConsumer(queue, messageSelector);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public QueueReceiver createReceiver(final Queue queue) throws JMSException {
        return (QueueReceiver) createConsumer(queue);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public QueueSender createSender(final Queue queue) throws JMSException {
        return (QueueSender) createProducer(queue);
     }
  
     // XAQueueSession implementation
  
     public QueueSession getQueueSession() throws JMSException {
        return (QueueSession) getSession();
     }
  
     // TopicSession implementation
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TopicPublisher createPublisher(final Topic topic) throws JMSException {
        return (TopicPublisher) createProducer(topic);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TopicSubscriber createSubscriber(final Topic topic,
                                             final String messageSelector,
                                             final boolean noLocal) throws JMSException {
        return (TopicSubscriber) createConsumer(topic, messageSelector, noLocal);
     }
  
     /**
      * {@inheritDoc}
      */
     @Override
     public TopicSubscriber createSubscriber(final Topic topic) throws JMSException {
        return (TopicSubscriber) createConsumer(topic);
     }
  
     // XATopicSession implementation
  
     public TopicSession getTopicSession() throws JMSException {
        return (TopicSession) getSession();
     }
  
     @Override
     public String toString() {
        return "ActiveMQSession->" + session;
     }
  
     public ClientSession getCoreSession() {
        return session;
     }
  
     public boolean isRecoverCalled() {
        return recoverCalled;
     }
  
     public void setRecoverCalled(final boolean recoverCalled) {
        this.recoverCalled = recoverCalled;
     }
  
     public void deleteTemporaryTopic(final ActiveMQDestination tempTopic) throws JMSException {
        if (!tempTopic.isTemporary()) {
           throw new InvalidDestinationException("Not a temporary topic " + tempTopic);
        }
  
        try {
           AddressQuery response = session.addressQuery(tempTopic.getSimpleAddress());
  
           if (!response.isExists()) {
              throw new InvalidDestinationException("Cannot delete temporary topic " + tempTopic.getName() +
                                                       " does not exist");
           }
  
           if (response.getQueueNames().size() > 1) {
              throw new IllegalStateException("Cannot delete temporary topic " + tempTopic.getName() +
                                                 " since it has subscribers");
           }
  
           SimpleString address = tempTopic.getSimpleAddress();
  
           session.deleteQueue(address);
  
           connection.removeTemporaryQueue(address);
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     public void deleteTemporaryQueue(final ActiveMQDestination tempQueue) throws JMSException {
        if (!tempQueue.isTemporary()) {
           throw new InvalidDestinationException("Not a temporary queue " + tempQueue);
        }
        try {
           QueueQuery response = session.queueQuery(tempQueue.getSimpleAddress());
  
           if (!response.isExists()) {
              throw new InvalidDestinationException("Cannot delete temporary queue " + tempQueue.getName() +
                                                       " does not exist");
           }
  
           if (response.getConsumerCount() > 0) {
              throw new IllegalStateException("Cannot delete temporary queue " + tempQueue.getName() +
                                                 " since it has subscribers");
           }
  
           SimpleString address = tempQueue.getSimpleAddress();
  
           session.deleteQueue(address);
  
           connection.removeTemporaryQueue(address);
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     public void start() throws JMSException {
        try {
           session.start();
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     public void stop() throws JMSException {
        try {
           session.stop();
        } catch (ActiveMQException e) {
           throw JMSExceptionHelper.convertFromActiveMQException(e);
        }
     }
  
     public void removeConsumer(final ActiveMQMessageConsumer consumer) {
        consumers.remove(consumer);
     }
  
     public boolean isEnable1xPrefixes() {
        return enable1xPrefixes;
     }
  
     void deleteQueue(final SimpleString queueName) throws JMSException {
        if (!session.isClosed()) {
           try {
              session.deleteQueue(queueName);
           } catch (ActiveMQException ignore) {
              // Exception on deleting queue shouldn't prevent close from completing
           }
        }
     }
  
     public ActiveMQConnection getConnection() {
        return connection;
     }
  
  
     void checkClosed() throws JMSException {
        if (session.isClosed()) {
           throw new IllegalStateException("Session is closed");
        }
     }
  
     void createTemporaryQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, ClientSession.AddressQuery addressQuery) throws ActiveMQException {
        QueueConfiguration queueConfiguration = Objects.requireNonNullElse(destination.getQueueConfiguration(), QueueConfiguration.of(queueName));
        AutoCreateUtil.setRequiredQueueConfigurationIfNotSet(queueConfiguration, addressQuery, routingType, filter, false);
        session.createQueue(queueConfiguration.setName(queueName).setAddress(destination.getAddress()).setDurable(false).setTemporary(true));
     }
  
     void createSharedQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, boolean durable, ClientSession.AddressQuery addressQuery) throws ActiveMQException {
        QueueConfiguration queueConfiguration = Objects.requireNonNullElseGet(destination.getQueueConfiguration(), () -> QueueConfiguration.of(queueName));
        AutoCreateUtil.setRequiredQueueConfigurationIfNotSet(queueConfiguration, addressQuery, routingType, filter, durable);
        session.createSharedQueue(queueConfiguration.setName(queueName).setAddress(destination.getAddress()).setDurable(durable));
     }
  
     void createQueue(ActiveMQDestination destination, RoutingType routingType, SimpleString queueName, SimpleString filter, boolean durable, boolean autoCreated, ClientSession.AddressQuery addressQuery) throws ActiveMQException {
        QueueConfiguration queueConfiguration = Objects.requireNonNullElseGet(destination.getQueueConfiguration(), () -> QueueConfiguration.of(queueName));
        AutoCreateUtil.setRequiredQueueConfigurationIfNotSet(queueConfiguration, addressQuery, routingType, filter, durable);
        session.createQueue(queueConfiguration.setName(queueName).setAddress(destination.getAddress()).setAutoCreated(autoCreated).setDurable(durable));
     }
  
  
  
     private ActiveMQQueue lookupQueue(final String queueName, boolean isTemporary) throws ActiveMQException {
        String queueNameToUse = queueName;
        if (enable1xPrefixes) {
           queueNameToUse = (isTemporary ? PacketImpl.OLD_TEMP_QUEUE_PREFIX.toString() : PacketImpl.OLD_QUEUE_PREFIX.toString()) + queueName;
        }
  
        ActiveMQQueue queue;
  
        if (isTemporary) {
           queue = ActiveMQDestination.createTemporaryQueue(queueNameToUse);
        } else {
           queue = ActiveMQDestination.createQueue(queueNameToUse);
        }
  
        if (queueName != queueNameToUse) {
           queue.setName(queueName);
        }
  
        QueueQuery response = session.queueQuery(queue.getSimpleAddress());
  
        if (!response.isExists() && !response.isAutoCreateQueues()) {
           return null;
        } else {
           return queue;
        }
     }
  
     private ActiveMQTopic lookupTopic(final String topicName, final boolean isTemporary) throws ActiveMQException {
        String topicNameToUse = topicName;
        if (enable1xPrefixes) {
           topicNameToUse = (isTemporary ? PacketImpl.OLD_TEMP_TOPIC_PREFIX.toString() : PacketImpl.OLD_TOPIC_PREFIX.toString()) + topicName;
        }
  
        ActiveMQTopic topic;
  
        if (isTemporary) {
           topic = ActiveMQDestination.createTemporaryTopic(topicNameToUse);
        } else {
           topic = ActiveMQDestination.createTopic(topicNameToUse);
        }
  
        if (topicNameToUse != topicName) {
           topic.setName(topicName);
        }
  
  
        AddressQuery query = session.addressQuery(topic.getSimpleAddress());
  
        if (!query.isExists() && !query.isAutoCreateAddresses()) {
           return null;
        } else {
           return topic;
        }
     }
  
  }
  