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.util.Collections;
  import java.util.List;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import jakarta.jms.InvalidSelectorException;
  import jakarta.jms.JMSException;
  import javax.management.ObjectName;
  
  import org.apache.activemq.broker.Broker;
  import org.apache.activemq.broker.ConnectionContext;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ConsumerId;
  import org.apache.activemq.command.ConsumerInfo;
  import org.apache.activemq.command.MessageAck;
  import org.apache.activemq.filter.BooleanExpression;
  import org.apache.activemq.filter.DestinationFilter;
  import org.apache.activemq.filter.LogicExpression;
  import org.apache.activemq.filter.MessageEvaluationContext;
  import org.apache.activemq.filter.NoLocalExpression;
  import org.apache.activemq.selector.SelectorParser;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public abstract class AbstractSubscription implements Subscription {
  
      private static final Logger LOG = LoggerFactory.getLogger(AbstractSubscription.class);
  
      protected Broker broker;
      protected ConnectionContext context;
      protected ConsumerInfo info;
      protected final DestinationFilter destinationFilter;
      protected final CopyOnWriteArrayList<Destination> destinations = new CopyOnWriteArrayList<Destination>();
      protected final AtomicInteger prefetchExtension = new AtomicInteger(0);
  
      private boolean usePrefetchExtension = true;
      private BooleanExpression selectorExpression;
      private ObjectName objectName;
      private int cursorMemoryHighWaterMark = 70;
      private boolean slowConsumer;
      private long lastAckTime;
      private final SubscriptionStatistics subscriptionStatistics = new SubscriptionStatistics();
  
      public AbstractSubscription(Broker broker,ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
          this.broker = broker;
          this.context = context;
          this.info = info;
          this.destinationFilter = DestinationFilter.parseFilter(info.getDestination());
          this.selectorExpression = parseSelector(info);
          this.lastAckTime = System.currentTimeMillis();
      }
  
      private static BooleanExpression parseSelector(ConsumerInfo info) throws InvalidSelectorException {
          BooleanExpression rc = null;
          if (info.getSelector() != null) {
              rc = SelectorParser.parse(info.getSelector());
          }
          if (info.isNoLocal()) {
              if (rc == null) {
                  rc = new NoLocalExpression(info.getConsumerId().getConnectionId());
              } else {
                  rc = LogicExpression.createAND(new NoLocalExpression(info.getConsumerId().getConnectionId()), rc);
              }
          }
          if (info.getAdditionalPredicate() != null) {
              if (rc == null) {
                  rc = info.getAdditionalPredicate();
              } else {
                  rc = LogicExpression.createAND(info.getAdditionalPredicate(), rc);
              }
          }
          return rc;
      }
  
      @Override
      public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception {
          this.lastAckTime = System.currentTimeMillis();
          subscriptionStatistics.getConsumedCount().increment();
      }
  
      @Override
      public boolean matches(MessageReference node, MessageEvaluationContext context) throws IOException {
          ConsumerId targetConsumerId = node.getTargetConsumerId();
          if (targetConsumerId != null) {
              if (!targetConsumerId.equals(info.getConsumerId())) {
                  return false;
              }
          }
          try {
              return (selectorExpression == null || selectorExpression.matches(context)) && this.context.isAllowedToConsume(node);
          } catch (JMSException e) {
              LOG.info("Selector failed to evaluate: {}", e.getMessage(), e);
              return false;
          }
      }
  
      @Override
      public boolean isWildcard() {
          return destinationFilter.isWildcard();
      }
  
      @Override
      public boolean matches(ActiveMQDestination destination) {
          return destinationFilter.matches(destination);
      }
  
      @Override
      public void add(ConnectionContext context, Destination destination) throws Exception {
          destinations.add(destination);
      }
  
      @Override
      public List<MessageReference> remove(ConnectionContext context, Destination destination) throws Exception {
          destinations.remove(destination);
          return Collections.EMPTY_LIST;
      }
  
      @Override
      public ConsumerInfo getConsumerInfo() {
          return info;
      }
  
      @Override
      public void gc() {
      }
  
      @Override
      public ConnectionContext getContext() {
          return context;
      }
  
      public ConsumerInfo getInfo() {
          return info;
      }
  
      public BooleanExpression getSelectorExpression() {
          return selectorExpression;
      }
  
      @Override
      public String getSelector() {
          return info.getSelector();
      }
  
      @Override
      public void setSelector(String selector) throws InvalidSelectorException {
          ConsumerInfo copy = info.copy();
          copy.setSelector(selector);
          BooleanExpression newSelector = parseSelector(copy);
          // its valid so lets actually update it now
          info.setSelector(selector);
          this.selectorExpression = newSelector;
      }
  
      @Override
      public ObjectName getObjectName() {
          return objectName;
      }
  
      @Override
      public void setObjectName(ObjectName objectName) {
          this.objectName = objectName;
      }
  
      @Override
      public int getPrefetchSize() {
          return info.getPrefetchSize();
      }
  
      public boolean isUsePrefetchExtension() {
          return usePrefetchExtension;
      }
  
      public void setUsePrefetchExtension(boolean usePrefetchExtension) {
          this.usePrefetchExtension = usePrefetchExtension;
      }
  
      public void setPrefetchSize(int newSize) {
          info.setPrefetchSize(newSize);
      }
  
      @Override
      public boolean isRecoveryRequired() {
          return true;
      }
  
      @Override
      public boolean isSlowConsumer() {
          return slowConsumer;
      }
  
      public void setSlowConsumer(boolean val) {
          slowConsumer = val;
      }
  
      @Override
      public boolean addRecoveredMessage(ConnectionContext context, MessageReference message) throws Exception {
          boolean result = false;
          MessageEvaluationContext msgContext = context.getMessageEvaluationContext();
          try {
              Destination regionDestination = (Destination) message.getRegionDestination();
              msgContext.setDestination(regionDestination.getActiveMQDestination());
              msgContext.setMessageReference(message);
              result = matches(message, msgContext);
              if (result) {
                  doAddRecoveredMessage(message);
              }
          } finally {
              msgContext.clear();
          }
          return result;
      }
  
      @Override
      public ActiveMQDestination getActiveMQDestination() {
          return info != null ? info.getDestination() : null;
      }
  
      @Override
      public boolean isBrowser() {
          return info != null && info.isBrowser();
      }
  
      @Override
      public long getInFlightMessageSize() {
          return subscriptionStatistics.getInflightMessageSize().getTotalSize();
      }
  
      @Override
      public int getInFlightUsage() {
          int prefetchSize = info.getPrefetchSize();
          if (prefetchSize > 0) {
              return (getInFlightSize() * 100) / prefetchSize;
          }
          return Integer.MAX_VALUE;
      }
  
      /**
       * Add a destination
       * @param destination
       */
      public void addDestination(Destination destination) {
      }
  
      /**
       * Remove a destination
       * @param destination
       */
      public void removeDestination(Destination destination) {
      }
  
      @Override
      public int getCursorMemoryHighWaterMark(){
          return this.cursorMemoryHighWaterMark;
      }
  
      @Override
      public void setCursorMemoryHighWaterMark(int cursorMemoryHighWaterMark){
          this.cursorMemoryHighWaterMark=cursorMemoryHighWaterMark;
      }
  
      @Override
      public int countBeforeFull() {
          return info.getPrefetchSize() - getDispatchedQueueSize();
      }
  
      @Override
      public void unmatched(MessageReference node) throws IOException {
          // only durable topic subs have something to do here
      }
  
      protected void doAddRecoveredMessage(MessageReference message) throws Exception {
          add(message);
      }
  
      @Override
      public long getTimeOfLastMessageAck() {
          return lastAckTime;
      }
  
      public void setTimeOfLastMessageAck(long value) {
          this.lastAckTime = value;
      }
  
      @Override
      public long getConsumedCount(){
          return subscriptionStatistics.getConsumedCount().getCount();
      }
  
      @Override
      public void incrementConsumedCount(){
          subscriptionStatistics.getConsumedCount().increment();
      }
  
      @Override
      public void resetConsumedCount(){
          subscriptionStatistics.getConsumedCount().reset();
      }
  
      @Override
      public SubscriptionStatistics getSubscriptionStatistics() {
          return subscriptionStatistics;
      }
  
      public void wakeupDestinationsForDispatch() {
          for (Destination dest : destinations) {
              dest.wakeup();
          }
      }
  
      public AtomicInteger getPrefetchExtension() {
          return this.prefetchExtension;
      }
  
      protected void contractPrefetchExtension(int amount) {
          if (isUsePrefetchExtension() && getPrefetchSize() != 0) {
              decrementPrefetchExtension(amount);
          }
      }
  
      protected void expandPrefetchExtension(int amount) {
          if (isUsePrefetchExtension() && getPrefetchSize() != 0) {
              incrementPrefetchExtension(amount);
          }
      }
  
      protected void decrementPrefetchExtension(int amount) {
          while (true) {
              int currentExtension = prefetchExtension.get();
              int newExtension = Math.max(0, currentExtension - amount);
              if (prefetchExtension.compareAndSet(currentExtension, newExtension)) {
                  break;
              }
          }
      }
  
      private void incrementPrefetchExtension(int amount) {
          while (true) {
              int currentExtension = prefetchExtension.get();
              int newExtension = Math.max(currentExtension, currentExtension + amount);
              if (prefetchExtension.compareAndSet(currentExtension, newExtension)) {
                  break;
              }
          }
      }
  
      public CopyOnWriteArrayList<Destination> getDestinations() {
          return destinations;
      }
  }
  