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.broker.region;
  
  import java.io.IOException;
  import java.io.UncheckedIOException;
  import java.util.ArrayList;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.concurrent.CancellationException;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.Future;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  
  import java.util.stream.Collectors;
  import org.apache.activemq.advisory.AdvisorySupport;
  import org.apache.activemq.broker.BrokerService;
  import org.apache.activemq.broker.ConnectionContext;
  import org.apache.activemq.broker.ProducerBrokerExchange;
  import org.apache.activemq.broker.region.policy.DispatchPolicy;
  import org.apache.activemq.broker.region.policy.LastImageSubscriptionRecoveryPolicy;
  import org.apache.activemq.broker.region.policy.RetainedMessageSubscriptionRecoveryPolicy;
  import org.apache.activemq.broker.region.policy.SimpleDispatchPolicy;
  import org.apache.activemq.broker.region.policy.SubscriptionRecoveryPolicy;
  import org.apache.activemq.broker.util.InsertionCountList;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ConsumerInfo;
  import org.apache.activemq.command.ExceptionResponse;
  import org.apache.activemq.command.Message;
  import org.apache.activemq.command.MessageAck;
  import org.apache.activemq.command.MessageId;
  import org.apache.activemq.command.ProducerAck;
  import org.apache.activemq.command.ProducerInfo;
  import org.apache.activemq.command.Response;
  import org.apache.activemq.command.SubscriptionInfo;
  import org.apache.activemq.filter.MessageEvaluationContext;
  import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
  import org.apache.activemq.store.MessageRecoveryListener;
  import org.apache.activemq.store.MessageStore.StoreType;
  import org.apache.activemq.store.NoLocalSubscriptionAware;
  import org.apache.activemq.store.PersistenceAdapter;
  import org.apache.activemq.store.TopicMessageStore;
  import org.apache.activemq.thread.Task;
  import org.apache.activemq.thread.TaskRunner;
  import org.apache.activemq.thread.TaskRunnerFactory;
  import org.apache.activemq.transaction.Synchronization;
  import org.apache.activemq.util.SubscriptionKey;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import jakarta.jms.JMSException;
  import jakarta.jms.MessageFormatException;
  import jakarta.jms.MessageFormatRuntimeException;
  
  import static org.apache.activemq.transaction.Transaction.IN_USE_STATE;
  
  /**
   * The Topic is a destination that sends a copy of a message to every active
   * Subscription registered.
   */
  public class Topic extends BaseDestination implements Task {
      protected static final Logger LOG = LoggerFactory.getLogger(Topic.class);
      private final TopicMessageStore topicStore;
      protected final CopyOnWriteArrayList<Subscription> consumers = new CopyOnWriteArrayList<Subscription>();
      private final ReentrantReadWriteLock dispatchLock = new ReentrantReadWriteLock();
      private DispatchPolicy dispatchPolicy = new SimpleDispatchPolicy();
      private SubscriptionRecoveryPolicy subscriptionRecoveryPolicy;
      private final ConcurrentMap<SubscriptionKey, DurableTopicSubscription> durableSubscribers = new ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription>();
      private final TaskRunner taskRunner;
      private final TaskRunnerFactory taskRunnerFactor;
      private final LinkedList<Runnable> messagesWaitingForSpace = new LinkedList<Runnable>();
      private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
          @Override
          public void run() {
              try {
                  Topic.this.taskRunner.wakeup();
              } catch (InterruptedException e) {
              }
          }
      };
  
      public Topic(BrokerService brokerService, ActiveMQDestination destination, TopicMessageStore store,
              DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception {
          super(brokerService, store, destination, parentStats);
          this.topicStore = store;
          subscriptionRecoveryPolicy = new RetainedMessageSubscriptionRecoveryPolicy(null);
          this.taskRunner = taskFactory.createTaskRunner(this, "Topic  " + destination.getPhysicalName());
          this.taskRunnerFactor = taskFactory;
      }
  
      @Override
      public void initialize() throws Exception {
          super.initialize();
          // set non default subscription recovery policy (override policyEntries)
          if (AdvisorySupport.isMasterBrokerAdvisoryTopic(destination)) {
              subscriptionRecoveryPolicy = new LastImageSubscriptionRecoveryPolicy();
              setAlwaysRetroactive(true);
          }
          if (store != null) {
              // AMQ-2586: Better to leave this stat at zero than to give the user
              // misleading metrics.
              // int messageCount = store.getMessageCount();
              // destinationStatistics.getMessages().setCount(messageCount);
              store.start();
          }
      }
  
      @Override
      public List<Subscription> getConsumers() {
          synchronized (consumers) {
              return new ArrayList<Subscription>(consumers);
          }
      }
  
      public boolean lock(MessageReference node, LockOwner sub) {
          return true;
      }
  
      @Override
      public void addSubscription(ConnectionContext context, final Subscription sub) throws Exception {
          if (!sub.getConsumerInfo().isDurable()) {
  
              // Do a retroactive recovery if needed.
              if (sub.getConsumerInfo().isRetroactive() || isAlwaysRetroactive()) {
  
                  // synchronize with dispatch method so that no new messages are sent
                  // while we are recovering a subscription to avoid out of order messages.
                  dispatchLock.writeLock().lock();
                  try {
                      boolean applyRecovery = false;
                      synchronized (consumers) {
                          if (!consumers.contains(sub)){
                              sub.add(context, this);
                              consumers.add(sub);
                              applyRecovery=true;
                              super.addSubscription(context, sub);
                          }
                      }
                      if (applyRecovery){
                          subscriptionRecoveryPolicy.recover(context, this, sub);
                      }
                  } finally {
                      dispatchLock.writeLock().unlock();
                  }
  
              } else {
                  synchronized (consumers) {
                      if (!consumers.contains(sub)){
                          sub.add(context, this);
                          consumers.add(sub);
                          super.addSubscription(context, sub);
                      }
                  }
              }
          } else {
              DurableTopicSubscription dsub = (DurableTopicSubscription) sub;
              super.addSubscription(context, sub);
              sub.add(context, this);
              if(dsub.isActive()) {
                  synchronized (consumers) {
                      boolean hasSubscription = false;
  
                      if (consumers.size() == 0) {
                          hasSubscription = false;
                      } else {
                          for (Subscription currentSub : consumers) {
                              if (currentSub.getConsumerInfo().isDurable()) {
                                  DurableTopicSubscription dcurrentSub = (DurableTopicSubscription) currentSub;
                                  if (dcurrentSub.getSubscriptionKey().equals(dsub.getSubscriptionKey())) {
                                      hasSubscription = true;
                                      break;
                                  }
                              }
                          }
                      }
  
                      if (!hasSubscription) {
                          consumers.add(sub);
                      }
                  }
              }
              durableSubscribers.put(dsub.getSubscriptionKey(), dsub);
          }
      }
  
      @Override
      public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) throws Exception {
          if (!sub.getConsumerInfo().isDurable()) {
              boolean removed = false;
              synchronized (consumers) {
                  removed = consumers.remove(sub);
              }
              if (removed) {
                  super.removeSubscription(context, sub, lastDeliveredSequenceId);
              }
          }
          sub.remove(context, this);
      }
  
      public void deleteSubscription(ConnectionContext context, SubscriptionKey key) throws Exception {
          if (topicStore != null) {
              topicStore.deleteSubscription(key.clientId, key.subscriptionName);
              DurableTopicSubscription removed = durableSubscribers.remove(key);
              if (removed != null) {
                  destinationStatistics.getConsumers().decrement();
                  // deactivate and remove
                  removed.deactivate(false, 0l);
                  consumers.remove(removed);
              }
          }
      }
  
      private boolean hasDurableSubChanged(SubscriptionInfo info1, ConsumerInfo info2) throws IOException {
          if (hasSelectorChanged(info1, info2)) {
              return true;
          }
  
          return hasNoLocalChanged(info1, info2);
      }
  
      private boolean hasNoLocalChanged(SubscriptionInfo info1, ConsumerInfo info2) throws IOException {
          //Not all persistence adapters store the noLocal value for a subscription
          PersistenceAdapter adapter = broker.getBrokerService().getPersistenceAdapter();
          if (adapter instanceof NoLocalSubscriptionAware) {
              if (info1.isNoLocal() ^ info2.isNoLocal()) {
                  return true;
              }
          }
  
          return false;
      }
  
      private boolean hasSelectorChanged(SubscriptionInfo info1, ConsumerInfo info2) {
          if (info1.getSelector() != null ^ info2.getSelector() != null) {
              return true;
          }
  
          if (info1.getSelector() != null && !info1.getSelector().equals(info2.getSelector())) {
              return true;
          }
  
          return false;
      }
  
      public void activate(ConnectionContext context, final DurableTopicSubscription subscription) throws Exception {
          // synchronize with dispatch method so that no new messages are sent
          // while we are recovering a subscription to avoid out of order messages.
          dispatchLock.writeLock().lock();
          try {
  
              if (topicStore == null) {
                  return;
              }
  
              // Recover the durable subscription.
              String clientId = subscription.getSubscriptionKey().getClientId();
              String subscriptionName = subscription.getSubscriptionKey().getSubscriptionName();
              SubscriptionInfo info = topicStore.lookupSubscription(clientId, subscriptionName);
              if (info != null) {
                  // Check to see if selector changed.
                  if (hasDurableSubChanged(info, subscription.getConsumerInfo())) {
                      // Need to delete the subscription
                      topicStore.deleteSubscription(clientId, subscriptionName);
                      info = null;
                      // Force a rebuild of the selector chain for the subscription otherwise
                      // the stored subscription is updated but the selector expression is not
                      // and the subscription will not behave according to the new configuration.
                      subscription.setSelector(subscription.getConsumerInfo().getSelector());
                      synchronized (consumers) {
                          consumers.remove(subscription);
                      }
                  } else {
                      synchronized (consumers) {
                          if (!consumers.contains(subscription)) {
                              consumers.add(subscription);
                          }
                      }
                  }
              }
  
              // Do we need to create the subscription?
              if (info == null) {
                  info = new SubscriptionInfo();
                  info.setClientId(clientId);
                  info.setSelector(subscription.getConsumerInfo().getSelector());
                  info.setSubscriptionName(subscriptionName);
                  info.setDestination(getActiveMQDestination());
                  info.setNoLocal(subscription.getConsumerInfo().isNoLocal());
                  // This destination is an actual destination id.
                  info.setSubscribedDestination(subscription.getConsumerInfo().getDestination());
                  // This destination might be a pattern
                  synchronized (consumers) {
                      consumers.add(subscription);
                      topicStore.addSubscription(info, subscription.getConsumerInfo().isRetroactive());
                  }
              }
  
              final MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext();
              msgContext.setDestination(destination);
              if (subscription.isRecoveryRequired()) {
                  topicStore.recoverSubscription(clientId, subscriptionName, new MessageRecoveryListener() {
                      @Override
                      public boolean recoverMessage(Message message) throws Exception {
                          message.setRegionDestination(Topic.this);
                          try {
                              msgContext.setMessageReference(message);
                              if (subscription.matches(message, msgContext)) {
                                  subscription.add(message);
                              }
                          } catch (IOException e) {
                              LOG.error("Failed to recover this message {}", message, e);
                          }
                          return true;
                      }
  
                      @Override
                      public boolean recoverMessageReference(MessageId messageReference) throws Exception {
                          throw new RuntimeException("Should not be called.");
                      }
  
                      @Override
                      public boolean hasSpace() {
                          return true;
                      }
  
                      @Override
                      public boolean isDuplicate(MessageId id) {
                          return false;
                      }
                  });
              }
          } finally {
              dispatchLock.writeLock().unlock();
          }
      }
  
      public void deactivate(ConnectionContext context, DurableTopicSubscription sub, List<MessageReference> dispatched) throws Exception {
          synchronized (consumers) {
              consumers.remove(sub);
          }
          sub.remove(context, this, dispatched);
      }
  
      public void recoverRetroactiveMessages(ConnectionContext context, Subscription subscription) throws Exception {
          if (subscription.getConsumerInfo().isRetroactive()) {
              subscriptionRecoveryPolicy.recover(context, this, subscription);
          }
      }
  
      @Override
      public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception {
          final ConnectionContext context = producerExchange.getConnectionContext();
  
          final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo();
          producerExchange.incrementSend();
          final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0
                  && !context.isInRecoveryMode();
  
          message.setRegionDestination(this);
  
          if(getMessageInterceptorStrategy() != null) {
              try {
                  getMessageInterceptorStrategy().process(producerExchange, message);
              } catch (MessageFormatRuntimeException e) {
                  throw new MessageFormatException(e.getMessage(), e.getErrorCode());
              }
          }
  
          // There is delay between the client sending it and it arriving at the
          // destination.. it may have expired.
          if (message.isExpired()) {
              broker.messageExpired(context, message, null);
              getDestinationStatistics().getExpired().increment();
              if (sendProducerAck) {
                  ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
                  context.getConnection().dispatchAsync(ack);
              }
              return;
          }
  
          if (memoryUsage.isFull()) {
              isFull(context, memoryUsage);
              fastProducer(context, producerInfo);
  
              if (isProducerFlowControl() && context.isProducerFlowControl()) {
  
                  if (isFlowControlLogRequired()) {
                      LOG.warn("{}, Usage Manager memory limit reached {}. Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it. See http://activemq.apache.org/producer-flow-control.html for more info.",
                              getActiveMQDestination().getQualifiedName(), memoryUsage.getLimit());
                  } else {
                      LOG.debug("{}, Usage Manager memory limit reached {}. Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it. See http://activemq.apache.org/producer-flow-control.html for more info.",
                              getActiveMQDestination().getQualifiedName(), memoryUsage.getLimit());
                  }
  
                  if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) {
                      throw new jakarta.jms.ResourceAllocationException("Usage Manager memory limit ("
                              + memoryUsage.getLimit() + ") reached. Rejecting send for producer (" + message.getProducerId()
                              + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
                              + " See http://activemq.apache.org/producer-flow-control.html for more info");
                  }
  
                  // We can avoid blocking due to low usage if the producer is sending a sync message or
                  // if it is using a producer window
                  if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) {
                      synchronized (messagesWaitingForSpace) {
                          messagesWaitingForSpace.add(new Runnable() {
                              @Override
                              public void run() {
                                  try {
  
                                      // While waiting for space to free up...
                                      // the transaction may be done
                                      if (message.isInTransaction()) {
                                          if (context.getTransaction() == null || context.getTransaction().getState() > IN_USE_STATE) {
                                              throw new JMSException("Send transaction completed while waiting for space");
                                          }
                                      }
  
                                      // the message may have expired.
                                      if (message.isExpired()) {
                                          broker.messageExpired(context, message, null);
                                          getDestinationStatistics().getExpired().increment();
                                      } else {
                                          doMessageSend(producerExchange, message);
                                      }
  
                                      if (sendProducerAck) {
                                          ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message
                                                  .getSize());
                                          context.getConnection().dispatchAsync(ack);
                                      } else {
                                          Response response = new Response();
                                          response.setCorrelationId(message.getCommandId());
                                          context.getConnection().dispatchAsync(response);
                                      }
  
                                  } catch (Exception e) {
                                      if (!sendProducerAck && !context.isInRecoveryMode()) {
                                          ExceptionResponse response = new ExceptionResponse(e);
                                          response.setCorrelationId(message.getCommandId());
                                          context.getConnection().dispatchAsync(response);
                                      }
                                  }
                              }
                          });
  
                          registerCallbackForNotFullNotification();
                          context.setDontSendReponse(true);
                          return;
                      }
  
                  } else {
                      // Producer flow control cannot be used, so we have do the flow control
                      // at the broker by blocking this thread until there is space available.
  
                      if (memoryUsage.isFull()) {
                          if (context.isInTransaction()) {
  
                              int count = 0;
                              while (!memoryUsage.waitForSpace(1000)) {
                                  if (context.getStopping().get()) {
                                      throw new IOException("Connection closed, send aborted.");
                                  }
                                  if (count > 2 && context.isInTransaction()) {
                                      count = 0;
                                      int size = context.getTransaction().size();
                                      LOG.warn("Waiting for space to send transacted message - transaction elements = {} need more space to commit. Message = {}", size, message);
                                  }
                                  count++;
                              }
                          } else {
                              waitForSpace(
                                      context,
                                      producerExchange,
                                      memoryUsage,
                                      "Usage Manager Memory Usage limit reached. Stopping producer ("
                                              + message.getProducerId()
                                              + ") to prevent flooding "
                                              + getActiveMQDestination().getQualifiedName()
                                              + "."
                                              + " See http://activemq.apache.org/producer-flow-control.html for more info");
                          }
                      }
  
                      // The usage manager could have delayed us by the time
                      // we unblock the message could have expired..
                      if (message.isExpired()) {
                          getDestinationStatistics().getExpired().increment();
                          LOG.debug("Expired message: {}", message);
                          return;
                      }
                  }
              }
          }
  
          doMessageSend(producerExchange, message);
          messageDelivered(context, message);
          if (sendProducerAck) {
              ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
              context.getConnection().dispatchAsync(ack);
          }
      }
  
      /**
       * do send the message - this needs to be synchronized to ensure messages
       * are stored AND dispatched in the right order
       *
       * @param producerExchange
       * @param message
       * @throws IOException
       * @throws Exception
       */
      synchronized void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message)
              throws IOException, Exception {
          final ConnectionContext context = producerExchange.getConnectionContext();
          message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
          Future<Object> result = null;
  
          if (topicStore != null && message.isPersistent() && !canOptimizeOutPersistence()) {
              if (systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) {
                  final String logMessage = "Persistent store is Full, " + getStoreUsageHighWaterMark() + "% of "
                          + systemUsage.getStoreUsage().getLimit() + ". Stopping producer (" + message.getProducerId()
                          + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
                          + " See http://activemq.apache.org/producer-flow-control.html for more info";
                  if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) {
                      throw new jakarta.jms.ResourceAllocationException(logMessage);
                  }
  
                  waitForSpace(context,producerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage);
              }
              result = topicStore.asyncAddTopicMessage(context, message,isOptimizeStorage());
  
              //Moved the reduceMemoryfootprint clearing to the dispatch method
          }
  
          message.incrementReferenceCount();
  
          if (context.isInTransaction() && (context.getTransaction() != null)) {
              context.getTransaction().addSynchronization(new Synchronization() {
                  @Override
                  public void afterCommit() throws Exception {
                      // It could take while before we receive the commit
                      // operation.. by that time the message could have
                      // expired..
                      if (message.isExpired()) {
                          if (broker.isExpired(message)) {
                              getDestinationStatistics().getExpired().increment();
                              broker.messageExpired(context, message, null);
                          }
                          message.decrementReferenceCount();
                          return;
                      }
                      try {
                          dispatch(context, message);
                      } finally {
                          message.decrementReferenceCount();
                      }
                  }
  
                  @Override
                  public void afterRollback() throws Exception {
                      message.decrementReferenceCount();
                  }
              });
  
          } else {
              try {
                  dispatch(context, message);
              } finally {
                  message.decrementReferenceCount();
              }
          }
  
          if (result != null && !result.isCancelled()) {
              try {
                  result.get();
              } catch (CancellationException e) {
                  // ignore - the task has been cancelled if the message
                  // has already been deleted
              }
          }
      }
  
      private boolean canOptimizeOutPersistence() {
          return durableSubscribers.size() == 0;
      }
  
      @Override
      public String toString() {
          return "Topic: destination=" + destination.getPhysicalName() + ", subscriptions=" + consumers.size();
      }
  
      @Override
      public void acknowledge(ConnectionContext context, Subscription sub, final MessageAck ack,
              final MessageReference node) throws IOException {
          if (topicStore != null && node.isPersistent()) {
              if (sub instanceof DurableTopicSubscription) {
                  DurableTopicSubscription dsub = (DurableTopicSubscription) sub;
                  SubscriptionKey key = dsub.getSubscriptionKey();
                  topicStore.acknowledge(context, key.getClientId(), key.getSubscriptionName(),
                      node.getMessageId(),
                      convertToNonRangedAck(ack, node));
              }
          }
          messageConsumed(context, node);
      }
  
      @Override
      public void gc() {
      }
  
      public Message loadMessage(MessageId messageId) throws IOException {
          return topicStore != null ? topicStore.getMessage(messageId) : null;
      }
  
      @Override
      public void start() throws Exception {
          if (started.compareAndSet(false, true)) {
              this.subscriptionRecoveryPolicy.start();
              if (memoryUsage != null) {
                  memoryUsage.start();
              }
  
              if (getExpireMessagesPeriod() > 0 && !AdvisorySupport.isAdvisoryTopic(getActiveMQDestination())) {
                  scheduler.executePeriodically(expireMessagesTask, getExpireMessagesPeriod());
              }
          }
      }
  
      @Override
      public void stop() throws Exception {
          if (started.compareAndSet(true, false)) {
              if (taskRunner != null) {
                  taskRunner.shutdown();
              }
              this.subscriptionRecoveryPolicy.stop();
              if (memoryUsage != null) {
                  memoryUsage.stop();
              }
              if (this.topicStore != null) {
                  this.topicStore.stop();
              }
  
              scheduler.cancel(expireMessagesTask);
          }
      }
  
      @Override
      public Message[] browse() {
          final List<Message> result = new ArrayList<Message>();
          doBrowse(result, getMaxBrowsePageSize());
          return result.toArray(new Message[result.size()]);
      }
  
      public void doBrowse(final List<Message> browseList, final int max) {
          try {
              if (topicStore != null) {
                  final List<Message> toExpire = new ArrayList<Message>();
                  topicStore.recover(new MessageRecoveryListener() {
                      @Override
                      public boolean recoverMessage(Message message) throws Exception {
                          if (message.isExpired()) {
                              toExpire.add(message);
                          }
                          browseList.add(message);
                          return true;
                      }
  
                      @Override
                      public boolean recoverMessageReference(MessageId messageReference) throws Exception {
                          return true;
                      }
  
                      @Override
                      public boolean hasSpace() {
                          return browseList.size() < max;
                      }
  
                      @Override
                      public boolean isDuplicate(MessageId id) {
                          return false;
                      }
                  });
                  final ConnectionContext connectionContext = createConnectionContext();
                  for (Message message : toExpire) {
                      for (DurableTopicSubscription sub : durableSubscribers.values()) {
                          if (!sub.isActive() || sub.isEnableMessageExpirationOnActiveDurableSubs()) {
                              message.setRegionDestination(this);
                              // AMQ-9721 - Remove message from the cursor if it exists after
                              // loading from the store.  Store recoverExpired() does not inc
                              // the ref count so we don't need to decrement here, but if
                              // the cursor finds its own copy in memory it will dec that ref.
                              sub.removePending(message);
                              messageExpired(connectionContext, sub, message);
                          }
                      }
                  }
                  Message[] msgs = subscriptionRecoveryPolicy.browse(getActiveMQDestination());
                  if (msgs != null) {
                      for (int i = 0; i < msgs.length && browseList.size() < max; i++) {
                          browseList.add(msgs[i]);
                      }
                  }
              }
          } catch (Throwable e) {
              LOG.warn("Failed to browse Topic: {}", getActiveMQDestination().getPhysicalName(), e);
          }
      }
  
      @Override
      public boolean iterate() {
          synchronized (messagesWaitingForSpace) {
              while (!memoryUsage.isFull() && !messagesWaitingForSpace.isEmpty()) {
                  Runnable op = messagesWaitingForSpace.removeFirst();
                  op.run();
              }
  
              if (!messagesWaitingForSpace.isEmpty()) {
                  registerCallbackForNotFullNotification();
              }
          }
          return false;
      }
  
      private void registerCallbackForNotFullNotification() {
          // If the usage manager is not full, then the task will not
          // get called..
          if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) {
              // so call it directly here.
              sendMessagesWaitingForSpaceTask.run();
          }
      }
  
      // Properties
      // -------------------------------------------------------------------------
  
      public DispatchPolicy getDispatchPolicy() {
          return dispatchPolicy;
      }
  
      public void setDispatchPolicy(DispatchPolicy dispatchPolicy) {
          this.dispatchPolicy = dispatchPolicy;
      }
  
      public SubscriptionRecoveryPolicy getSubscriptionRecoveryPolicy() {
          return subscriptionRecoveryPolicy;
      }
  
      public void setSubscriptionRecoveryPolicy(SubscriptionRecoveryPolicy recoveryPolicy) {
          if (this.subscriptionRecoveryPolicy != null && this.subscriptionRecoveryPolicy instanceof RetainedMessageSubscriptionRecoveryPolicy) {
              // allow users to combine retained message policy with other ActiveMQ policies
              RetainedMessageSubscriptionRecoveryPolicy policy = (RetainedMessageSubscriptionRecoveryPolicy) this.subscriptionRecoveryPolicy;
              policy.setWrapped(recoveryPolicy);
          } else {
              this.subscriptionRecoveryPolicy = recoveryPolicy;
          }
      }
  
      // Implementation methods
      // -------------------------------------------------------------------------
  
      @Override
      public final void wakeup() {
      }
  
      protected void dispatch(final ConnectionContext context, Message message) throws Exception {
          // AMQ-2586: Better to leave this stat at zero than to give the user
          // misleading metrics.
          // destinationStatistics.getMessages().increment();
          destinationStatistics.getEnqueues().increment();
  
          final var tmpMessageFlowStats = destinationStatistics.getMessageFlowStats();
          if(tmpMessageFlowStats != null) {
              tmpMessageFlowStats.enqueueStats(context.getClientId(), message.getMessageId().toString(), message.getTimestamp(), message.getBrokerInTime());
          }
  
          if(isAdvancedNetworkStatisticsEnabled() && context != null && context.isNetworkConnection()) {
              destinationStatistics.getNetworkEnqueues().increment();
          }
  
          destinationStatistics.getMessageSize().addSize(message.getSize());
          MessageEvaluationContext msgContext = null;
  
          dispatchLock.readLock().lock();
          try {
              if (!subscriptionRecoveryPolicy.add(context, message)) {
                  return;
              }
              synchronized (consumers) {
                  if (consumers.isEmpty()) {
                      onMessageWithNoConsumers(context, message);
                      return;
                  }
              }
  
              // Clear memory before dispatch - need to clear here because the call to
              //subscriptionRecoveryPolicy.add() will unmarshall the state
              if (isReduceMemoryFootprint() && message.isMarshalled()) {
                  message.clearUnMarshalledState();
              }
  
              msgContext = context.getMessageEvaluationContext();
              msgContext.setDestination(destination);
              msgContext.setMessageReference(message);
              if (!dispatchPolicy.dispatch(message, msgContext, consumers)) {
                  onMessageWithNoConsumers(context, message);
              }
  
          } finally {
              dispatchLock.readLock().unlock();
              if (msgContext != null) {
                  msgContext.clear();
              }
          }
      }
  
  
      /**
       * Simple recovery listener that will check if the topic memory usage is full
       * when hasSpace() is called. This could be enhanced in the future if needed.
       */
      private final MessageRecoveryListener expiryListener =  new MessageRecoveryListener() {
  
          @Override
          public boolean recoverMessage(Message message) {
              return true;
          }
  
          @Override
          public boolean recoverMessageReference(MessageId ref)  {
              return true;
          }
  
          @Override
          public boolean hasSpace() {
              return !Topic.this.memoryUsage.isFull();
          }
  
          @Override
          public boolean isDuplicate(MessageId ref) {
              return false;
          }
      };
  
      private final AtomicBoolean expiryTaskInProgress = new AtomicBoolean(false);
      private final Runnable expireMessagesWork = () -> {
          try {
              final TopicMessageStore store = Topic.this.topicStore;
              if (store != null && store.getType() == StoreType.KAHADB) {
                  if (store.getMessageCount() == 0) {
                      LOG.debug("Skipping topic expiration check for {}, store size is 0", destination);
                      return;
                  }
  
                  // get the sub keys that should be checked for expired messages
                  final var subs = durableSubscribers.entrySet().stream()
                      .filter(entry -> isEligibleForExpiration(entry.getValue()))
                      .map(Entry::getKey).collect(Collectors.toSet());
  
                  if (subs.isEmpty()) {
                      LOG.debug("Skipping topic expiration check for {}, no eligible subscriptions to check", destination);
                      return;
                  }
  
                  // For each eligible subscription, return the messages in the store that are expired
                  // The same message refs are shared between subs if duplicated so this is efficient
                  var expired = store.recoverExpired(subs, getMaxExpirePageSize(),
                      expiryListener);
  
                  final ConnectionContext connectionContext = createConnectionContext();
                  // Go through any expired messages and remove for each sub
                  for (Entry<SubscriptionKey, List<Message>> entry : expired.entrySet()) {
                      DurableTopicSubscription sub = durableSubscribers.get(entry.getKey());
                      List<Message> expiredMessages = entry.getValue();
  
                      // If the sub still exists and there are expired messages then process
                      if (sub != null && !expiredMessages.isEmpty()) {
                          // There's a small race condition here if the sub comes online,
                          // but it's not a big deal as at worst there maybe be duplicate acks for
                          // the expired message but the store can handle it
                          if (isEligibleForExpiration(sub)) {
                              expiredMessages.forEach(message -> {
                                  message.setRegionDestination(Topic.this);
                                  try {
                                      // AMQ-9721 - Remove message from the cursor if it exists after
                                      // loading from the store.  Store recoverExpired() does not inc
                                      // the ref count so we don't need to decrement here, but if
                                      // the cursor finds its own copy in memory it will dec that ref.
                                      sub.removePending(message);
                                  } catch (IOException e) {
                                      throw new UncheckedIOException(e);
                                  }
                                  messageExpired(connectionContext, sub, message);
                              });
                          }
                      }
                  }
              } else {
                  // If not KahaDB, fall back to the legacy browse method because
                  // the recoverExpired() method is not supported
                  doBrowse(new InsertionCountList<>(), getMaxExpirePageSize());
              }
          } catch (Throwable e) {
              LOG.warn("Failed to expire messages on Topic: {}", getActiveMQDestination().getPhysicalName(), e);
          } finally {
              expiryTaskInProgress.set(false);
          }
      };
  
      private final Runnable expireMessagesTask = new Runnable() {
          @Override
          public void run() {
              if (expiryTaskInProgress.compareAndSet(false, true)) {
                  taskRunnerFactor.execute(expireMessagesWork);
              }
          }
      };
  
      @Override
      public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) {
          broker.messageExpired(context, reference, subs);
          // AMQ-2586: Better to leave this stat at zero than to give the user
          // misleading metrics.
          // destinationStatistics.getMessages().decrement();
          destinationStatistics.getExpired().increment();
          MessageAck ack = new MessageAck();
          ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
          ack.setDestination(destination);
          ack.setMessageID(reference.getMessageId());
          try {
              acknowledge(context, subs, ack, reference);
          } catch (Exception e) {
              LOG.error("Failed to remove expired Message from the store ", e);
          }
      }
  
      @Override
      protected Logger getLog() {
          return LOG;
      }
  
      protected boolean isOptimizeStorage(){
          boolean result = false;
  
          if (isDoOptimzeMessageStorage() && durableSubscribers.isEmpty()==false){
                  result = true;
                  for (DurableTopicSubscription s : durableSubscribers.values()) {
                      if (s.isActive()== false){
                          result = false;
                          break;
                      }
                      if (s.getPrefetchSize()==0){
                          result = false;
                          break;
                      }
                      if (s.isSlowConsumer()){
                          result = false;
                          break;
                      }
                      if (s.getInFlightUsage() > getOptimizeMessageStoreInFlightLimit()){
                          result = false;
                          break;
                      }
                  }
          }
          return result;
      }
  
      /**
       * force a reread of the store - after transaction recovery completion
       * @param pendingAdditionsCount
       */
      @Override
      public void clearPendingMessages(int pendingAdditionsCount) {
          dispatchLock.readLock().lock();
          try {
              for (DurableTopicSubscription durableTopicSubscription : durableSubscribers.values()) {
                  clearPendingAndDispatch(durableTopicSubscription);
              }
          } finally {
              dispatchLock.readLock().unlock();
          }
      }
  
      private void clearPendingAndDispatch(DurableTopicSubscription durableTopicSubscription) {
          synchronized (durableTopicSubscription.pendingLock) {
              durableTopicSubscription.pending.clear();
              try {
                  durableTopicSubscription.dispatchPending();
              } catch (IOException exception) {
                  LOG.warn("After clear of pending, failed to dispatch to: {}, for: {}, pending: {}, exception: {}",
                          durableTopicSubscription,
                          destination,
                          durableTopicSubscription.pending,
                          exception);
              }
          }
      }
  
      private static boolean isEligibleForExpiration(DurableTopicSubscription sub) {
          return sub.isEnableMessageExpirationOnActiveDurableSubs() || !sub.isActive();
      }
  
      public Map<SubscriptionKey, DurableTopicSubscription> getDurableTopicSubs() {
          return durableSubscribers;
      }
  }
  