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 boolean isSlave() {
120 return broker.getBrokerService().isSlave();
121 }
122
123 public ConnectionContext getContext() {
124 return context;
125 }
126
127 public ConsumerInfo getInfo() {
128 return info;
129 }
130
131 public BooleanExpression getSelectorExpression() {
132 return selectorExpression;
133 }
134
135 public String getSelector() {
136 return info.getSelector();
137 }
138
139 public void setSelector(String selector) throws InvalidSelectorException {
140 ConsumerInfo copy = info.copy();
141 copy.setSelector(selector);
142 BooleanExpression newSelector = parseSelector(copy);
143 // its valid so lets actually update it now
144 info.setSelector(selector);
145 this.selectorExpression = newSelector;
146 }
147
148 public ObjectName getObjectName() {
149 return objectName;
150 }
151
152 public void setObjectName(ObjectName objectName) {
153 this.objectName = objectName;
154 }
155
156 public int getPrefetchSize() {
157 return info.getPrefetchSize();
158 }
159 public void setPrefetchSize(int newSize) {
160 info.setPrefetchSize(newSize);
161 }
162
163 public boolean isRecoveryRequired() {
164 return true;
165 }
166
167 public boolean isSlowConsumer() {
168 return slowConsumer;
169 }
170
171 public void setSlowConsumer(boolean val) {
172 slowConsumer = val;
173 }
174
175 public boolean addRecoveredMessage(ConnectionContext context, MessageReference message) throws Exception {
176 boolean result = false;
177 MessageEvaluationContext msgContext = context.getMessageEvaluationContext();
178 try {
179 msgContext.setDestination(message.getRegionDestination().getActiveMQDestination());
180 msgContext.setMessageReference(message);
181 result = matches(message, msgContext);
182 if (result) {
183 doAddRecoveredMessage(message);
184 }
185
186 } finally {
187 msgContext.clear();
188 }
189 return result;
190 }
191
192 public ActiveMQDestination getActiveMQDestination() {
193 return info != null ? info.getDestination() : null;
194 }
195
196 public boolean isBrowser() {
197 return info != null && info.isBrowser();
198 }
199
200 public int getInFlightUsage() {
201 if (info.getPrefetchSize() > 0) {
202 return (getInFlightSize() * 100)/info.getPrefetchSize();
203 }
204 return Integer.MAX_VALUE;
205 }
206
207 /**
208 * Add a destination
209 * @param destination
210 */
211 public void addDestination(Destination destination) {
212
213 }
214
215
216 /**
217 * Remove a destination
218 * @param destination
219 */
220 public void removeDestination(Destination destination) {
221
222 }
223
224 public int getCursorMemoryHighWaterMark(){
225 return this.cursorMemoryHighWaterMark;
226 }
227
228 public void setCursorMemoryHighWaterMark(int cursorMemoryHighWaterMark){
229 this.cursorMemoryHighWaterMark=cursorMemoryHighWaterMark;
230 }
231
232 public int countBeforeFull() {
233 return getDispatchedQueueSize() - info.getPrefetchSize();
234 }
235
236 public void unmatched(MessageReference node) throws IOException {
237 // only durable topic subs have something to do here
238 }
239
240 protected void doAddRecoveredMessage(MessageReference message) throws Exception {
241 add(message);
242 }
243 }