activemq-broker/src/main/java/org/apache/activemq/broker/region/AbstractSubscription.java - activemq - Git at Google

 /**
  * 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 javax.jms.InvalidSelectorException;
 import javax.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;
     }
 }
	/**
	* 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 javax.jms.InvalidSelectorException;
	import javax.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;
	}
	}