001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.broker.region;
018    
019    import java.io.IOException;
020    import java.util.Collections;
021    import java.util.List;
022    import java.util.concurrent.CopyOnWriteArrayList;
023    import javax.jms.InvalidSelectorException;
024    import javax.jms.JMSException;
025    import javax.management.ObjectName;
026    import org.apache.activemq.broker.Broker;
027    import org.apache.activemq.broker.ConnectionContext;
028    import org.apache.activemq.command.ActiveMQDestination;
029    import org.apache.activemq.command.ConsumerId;
030    import org.apache.activemq.command.ConsumerInfo;
031    import org.apache.activemq.filter.BooleanExpression;
032    import org.apache.activemq.filter.DestinationFilter;
033    import org.apache.activemq.filter.LogicExpression;
034    import org.apache.activemq.filter.MessageEvaluationContext;
035    import org.apache.activemq.filter.NoLocalExpression;
036    import org.apache.activemq.selector.SelectorParser;
037    import org.slf4j.Logger;
038    import org.slf4j.LoggerFactory;
039    
040    public abstract class AbstractSubscription implements Subscription {
041    
042        private static final Logger LOG = LoggerFactory.getLogger(AbstractSubscription.class);
043        protected Broker broker;
044        protected ConnectionContext context;
045        protected ConsumerInfo info;
046        protected final DestinationFilter destinationFilter;
047        protected final CopyOnWriteArrayList<Destination> destinations = new CopyOnWriteArrayList<Destination>();
048        private BooleanExpression selectorExpression;
049        private ObjectName objectName;
050        private int cursorMemoryHighWaterMark = 70;
051        private boolean slowConsumer;
052    
053    
054        public AbstractSubscription(Broker broker,ConnectionContext context, ConsumerInfo info) throws InvalidSelectorException {
055            this.broker = broker;
056            this.context = context;
057            this.info = info;
058            this.destinationFilter = DestinationFilter.parseFilter(info.getDestination());
059            this.selectorExpression = parseSelector(info);
060        }
061    
062        private static BooleanExpression parseSelector(ConsumerInfo info) throws InvalidSelectorException {
063            BooleanExpression rc = null;
064            if (info.getSelector() != null) {
065                rc = SelectorParser.parse(info.getSelector());
066            }
067            if (info.isNoLocal()) {
068                if (rc == null) {
069                    rc = new NoLocalExpression(info.getConsumerId().getConnectionId());
070                } else {
071                    rc = LogicExpression.createAND(new NoLocalExpression(info.getConsumerId().getConnectionId()), rc);
072                }
073            }
074            if (info.getAdditionalPredicate() != null) {
075                if (rc == null) {
076                    rc = info.getAdditionalPredicate();
077                } else {
078                    rc = LogicExpression.createAND(info.getAdditionalPredicate(), rc);
079                }
080            }
081            return rc;
082        }
083    
084        public boolean matches(MessageReference node, MessageEvaluationContext context) throws IOException {
085            ConsumerId targetConsumerId = node.getTargetConsumerId();
086            if (targetConsumerId != null) {
087                if (!targetConsumerId.equals(info.getConsumerId())) {
088                    return false;
089                }
090            }
091            try {
092                return (selectorExpression == null || selectorExpression.matches(context)) && this.context.isAllowedToConsume(node);
093            } catch (JMSException e) {
094                LOG.info("Selector failed to evaluate: " + e.getMessage(), e);
095                return false;
096            }
097        }
098    
099        public boolean matches(ActiveMQDestination destination) {
100            return destinationFilter.matches(destination);
101        }
102    
103        public void add(ConnectionContext context, Destination destination) throws Exception {
104            destinations.add(destination);
105        }
106    
107        public List<MessageReference> remove(ConnectionContext context, Destination destination) throws Exception {
108            destinations.remove(destination);
109            return Collections.EMPTY_LIST;
110        }
111    
112        public ConsumerInfo getConsumerInfo() {
113            return info;
114        }
115    
116        public void gc() {
117        }
118    
119        public ConnectionContext getContext() {
120            return context;
121        }
122    
123        public ConsumerInfo getInfo() {
124            return info;
125        }
126    
127        public BooleanExpression getSelectorExpression() {
128            return selectorExpression;
129        }
130    
131        public String getSelector() {
132            return info.getSelector();
133        }
134    
135        public void setSelector(String selector) throws InvalidSelectorException {
136            ConsumerInfo copy = info.copy();
137            copy.setSelector(selector);
138            BooleanExpression newSelector = parseSelector(copy);
139            // its valid so lets actually update it now
140            info.setSelector(selector);
141            this.selectorExpression = newSelector;
142        }
143    
144        public ObjectName getObjectName() {
145            return objectName;
146        }
147    
148        public void setObjectName(ObjectName objectName) {
149            this.objectName = objectName;
150        }
151    
152        public int getPrefetchSize() {
153            return info.getPrefetchSize();
154        }
155        public void setPrefetchSize(int newSize) {
156            info.setPrefetchSize(newSize);
157        }
158    
159        public boolean isRecoveryRequired() {
160            return true;
161        }
162        
163        public boolean isSlowConsumer() {
164            return slowConsumer;
165        }
166        
167        public void setSlowConsumer(boolean val) {
168            slowConsumer = val;
169        }
170    
171        public boolean addRecoveredMessage(ConnectionContext context, MessageReference message) throws Exception {
172            boolean result = false;
173            MessageEvaluationContext msgContext = context.getMessageEvaluationContext();
174            try {
175                Destination regionDestination = (Destination) message.getRegionDestination();
176                msgContext.setDestination(regionDestination.getActiveMQDestination());
177                msgContext.setMessageReference(message);
178                result = matches(message, msgContext);
179                if (result) {
180                    doAddRecoveredMessage(message);
181                }
182    
183            } finally {
184                msgContext.clear();
185            }
186            return result;
187        }
188    
189        public ActiveMQDestination getActiveMQDestination() {
190            return info != null ? info.getDestination() : null;
191        }
192        
193        public boolean isBrowser() {
194            return info != null && info.isBrowser();
195        }
196        
197        public int getInFlightUsage() {
198            if (info.getPrefetchSize() > 0) {
199            return (getInFlightSize() * 100)/info.getPrefetchSize();
200            }
201            return Integer.MAX_VALUE;
202        }
203        
204        /**
205         * Add a destination
206         * @param destination
207         */
208        public void addDestination(Destination destination) {
209            
210        }
211           
212        
213        /**
214         * Remove a destination
215         * @param destination
216         */
217        public void removeDestination(Destination destination) {
218            
219        }
220        
221        public int getCursorMemoryHighWaterMark(){
222            return this.cursorMemoryHighWaterMark;
223        }
224    
225            public void setCursorMemoryHighWaterMark(int cursorMemoryHighWaterMark){
226                    this.cursorMemoryHighWaterMark=cursorMemoryHighWaterMark;
227            }
228        
229        public int countBeforeFull() {
230            return getDispatchedQueueSize() - info.getPrefetchSize();
231        }
232    
233        public void unmatched(MessageReference node) throws IOException {
234            // only durable topic subs have something to do here
235        }
236    
237        protected void doAddRecoveredMessage(MessageReference message) throws Exception {
238            add(message);
239        }
240    }