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.cursors;
018    
019    import java.util.Collections;
020    import java.util.HashMap;
021    import java.util.List;
022    import java.util.Map;
023    import java.util.concurrent.CopyOnWriteArrayList;
024    
025    import org.apache.activemq.advisory.AdvisorySupport;
026    import org.apache.activemq.broker.Broker;
027    import org.apache.activemq.broker.ConnectionContext;
028    import org.apache.activemq.broker.region.Destination;
029    import org.apache.activemq.broker.region.DurableTopicSubscription;
030    import org.apache.activemq.broker.region.MessageReference;
031    import org.apache.activemq.broker.region.Topic;
032    import org.apache.activemq.command.Message;
033    import org.apache.activemq.usage.SystemUsage;
034    import org.slf4j.Logger;
035    import org.slf4j.LoggerFactory;
036    
037    /**
038     * persist pending messages pending message (messages awaiting dispatch to a
039     * consumer) cursor
040     */
041    public class StoreDurableSubscriberCursor extends AbstractPendingMessageCursor {
042    
043        private static final Logger LOG = LoggerFactory.getLogger(StoreDurableSubscriberCursor.class);
044        private final String clientId;
045        private final String subscriberName;
046        private final Map<Destination, TopicStorePrefetch> topics = new HashMap<Destination, TopicStorePrefetch>();
047        private final List<PendingMessageCursor> storePrefetches = new CopyOnWriteArrayList<PendingMessageCursor>();
048        private final PendingMessageCursor nonPersistent;
049        private PendingMessageCursor currentCursor;
050        private final DurableTopicSubscription subscription;
051        private boolean immediatePriorityDispatch = true;
052    
053        /**
054         * @param broker Broker for this cursor
055         * @param clientId clientId for this cursor
056         * @param subscriberName subscriber name for this cursor
057         * @param maxBatchSize currently ignored
058         * @param subscription  subscription for this cursor
059         */
060        public StoreDurableSubscriberCursor(Broker broker,String clientId, String subscriberName,int maxBatchSize, DurableTopicSubscription subscription) {
061            super(AbstractPendingMessageCursor.isPrioritizedMessageSubscriber(broker,subscription));
062            this.subscription=subscription;
063            this.clientId = clientId;
064            this.subscriberName = subscriberName;
065            if (broker.getBrokerService().isPersistent()) {
066                this.nonPersistent = new FilePendingMessageCursor(broker,clientId + subscriberName,this.prioritizedMessages);
067            } else {
068                this.nonPersistent = new VMPendingMessageCursor(this.prioritizedMessages);
069            }
070    
071            this.nonPersistent.setMaxBatchSize(maxBatchSize);
072            this.nonPersistent.setSystemUsage(systemUsage);
073            this.storePrefetches.add(this.nonPersistent);
074    
075            if (prioritizedMessages) {
076                setMaxAuditDepth(10*getMaxAuditDepth());
077            }
078        }
079    
080        @Override
081        public synchronized void start() throws Exception {
082            if (!isStarted()) {
083                super.start();
084                for (PendingMessageCursor tsp : storePrefetches) {
085                    tsp.setMessageAudit(getMessageAudit());
086                    tsp.start();
087                }
088            }
089        }
090    
091        @Override
092        public synchronized void stop() throws Exception {
093            if (isStarted()) {
094                if (subscription.isKeepDurableSubsActive()) {
095                    super.gc();
096                    for (PendingMessageCursor tsp : storePrefetches) {
097                        tsp.gc();
098                    }
099                } else {
100                    super.stop();
101                    for (PendingMessageCursor tsp : storePrefetches) {
102                        tsp.stop();
103                    }
104                    getMessageAudit().clear();
105                }
106            }
107        }
108    
109        /**
110         * Add a destination
111         *
112         * @param context
113         * @param destination
114         * @throws Exception
115         */
116        @Override
117        public synchronized void add(ConnectionContext context, Destination destination) throws Exception {
118            if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination.getActiveMQDestination())) {
119                TopicStorePrefetch tsp = new TopicStorePrefetch(this.subscription,(Topic)destination, clientId, subscriberName);
120                tsp.setMaxBatchSize(destination.getMaxPageSize());
121                tsp.setSystemUsage(systemUsage);
122                tsp.setMessageAudit(getMessageAudit());
123                tsp.setEnableAudit(isEnableAudit());
124                tsp.setMemoryUsageHighWaterMark(getMemoryUsageHighWaterMark());
125                tsp.setUseCache(isUseCache());
126                tsp.setCacheEnabled(isUseCache() && tsp.isEmpty());
127                topics.put(destination, tsp);
128                storePrefetches.add(tsp);
129                if (isStarted()) {
130                    tsp.start();
131                }
132            }
133        }
134    
135        /**
136         * remove a destination
137         *
138         * @param context
139         * @param destination
140         * @throws Exception
141         */
142        @Override
143        public synchronized List<MessageReference> remove(ConnectionContext context, Destination destination) throws Exception {
144            PendingMessageCursor tsp = topics.remove(destination);
145            if (tsp != null) {
146                storePrefetches.remove(tsp);
147            }
148            return Collections.EMPTY_LIST;
149        }
150    
151        /**
152         * @return true if there are no pending messages
153         */
154        @Override
155        public synchronized boolean isEmpty() {
156            for (PendingMessageCursor tsp : storePrefetches) {
157                if( !tsp.isEmpty() )
158                    return false;
159            }
160            return true;
161        }
162    
163        @Override
164        public synchronized boolean isEmpty(Destination destination) {
165            boolean result = true;
166            TopicStorePrefetch tsp = topics.get(destination);
167            if (tsp != null) {
168                result = tsp.isEmpty();
169            }
170            return result;
171        }
172    
173        /**
174         * Informs the Broker if the subscription needs to intervention to recover
175         * it's state e.g. DurableTopicSubscriber may do
176         *
177         * @see org.apache.activemq.broker.region.cursors.AbstractPendingMessageCursor
178         * @return true if recovery required
179         */
180        @Override
181        public boolean isRecoveryRequired() {
182            return false;
183        }
184    
185        @Override
186        public synchronized void addMessageLast(MessageReference node) throws Exception {
187            if (node != null) {
188                Message msg = node.getMessage();
189                if (isStarted()) {
190                    if (!msg.isPersistent()) {
191                        nonPersistent.addMessageLast(node);
192                    }
193                }
194                if (msg.isPersistent()) {
195                    Destination dest = (Destination) msg.getRegionDestination();
196                    TopicStorePrefetch tsp = topics.get(dest);
197                    if (tsp != null) {
198                        tsp.addMessageLast(node);
199                        if (prioritizedMessages && immediatePriorityDispatch && tsp.isPaging()) {
200                            if (msg.getPriority() > tsp.getLastRecoveredPriority()) {
201                                tsp.recoverMessage(node.getMessage(), true);
202                                if (LOG.isTraceEnabled()) {
203                                    LOG.trace("cached high priority (" + msg.getPriority() + ") message:" +
204                                        msg.getMessageId() + ", current paged batch priority: " +
205                                        tsp.getLastRecoveredPriority() + ", cache size:" + tsp.batchList.size());
206                                }
207                            }
208                        }
209                    }
210                }
211    
212            }
213        }
214    
215        @Override
216        public boolean isTransient() {
217            return subscription.isKeepDurableSubsActive();
218        }
219    
220        @Override
221        public void addMessageFirst(MessageReference node) throws Exception {
222            // for keep durable subs active, need to deal with redispatch
223            if (node != null) {
224                Message msg = node.getMessage();
225                if (!msg.isPersistent()) {
226                    nonPersistent.addMessageFirst(node);
227                } else {
228                    Destination dest = (Destination) msg.getRegionDestination();
229                    TopicStorePrefetch tsp = topics.get(dest);
230                    if (tsp != null) {
231                        tsp.addMessageFirst(node);
232                    }
233                }
234            }
235        }
236    
237        @Override
238        public synchronized void addRecoveredMessage(MessageReference node) throws Exception {
239            nonPersistent.addMessageLast(node);
240        }
241    
242        @Override
243        public synchronized void clear() {
244            for (PendingMessageCursor tsp : storePrefetches) {
245                tsp.clear();
246            }
247        }
248    
249        @Override
250        public synchronized boolean hasNext() {
251            boolean result = true;
252            if (result) {
253                try {
254                    currentCursor = getNextCursor();
255                } catch (Exception e) {
256                    LOG.error("Failed to get current cursor ", e);
257                    throw new RuntimeException(e);
258                }
259                result = currentCursor != null ? currentCursor.hasNext() : false;
260            }
261            return result;
262        }
263    
264        @Override
265        public synchronized MessageReference next() {
266            MessageReference result = currentCursor != null ? currentCursor.next() : null;
267            return result;
268        }
269    
270        @Override
271        public synchronized void remove() {
272            if (currentCursor != null) {
273                currentCursor.remove();
274            }
275        }
276    
277        @Override
278        public synchronized void remove(MessageReference node) {
279            for (PendingMessageCursor tsp : storePrefetches) {
280                tsp.remove(node);
281            }
282        }
283    
284        @Override
285        public synchronized void reset() {
286            for (PendingMessageCursor storePrefetch : storePrefetches) {
287                storePrefetch.reset();
288            }
289        }
290    
291        @Override
292        public synchronized void release() {
293            this.currentCursor = null;
294            for (PendingMessageCursor storePrefetch : storePrefetches) {
295                storePrefetch.release();
296            }
297        }
298    
299        @Override
300        public synchronized int size() {
301            int pendingCount=0;
302            for (PendingMessageCursor tsp : storePrefetches) {
303                pendingCount += tsp.size();
304            }
305            return pendingCount;
306        }
307    
308        @Override
309        public void setMaxBatchSize(int newMaxBatchSize) {
310            for (PendingMessageCursor storePrefetch : storePrefetches) {
311                storePrefetch.setMaxBatchSize(newMaxBatchSize);
312            }
313            super.setMaxBatchSize(newMaxBatchSize);
314        }
315    
316        @Override
317        public synchronized void gc() {
318            for (PendingMessageCursor tsp : storePrefetches) {
319                tsp.gc();
320            }
321        }
322    
323        @Override
324        public void setSystemUsage(SystemUsage usageManager) {
325            super.setSystemUsage(usageManager);
326            for (PendingMessageCursor tsp : storePrefetches) {
327                tsp.setSystemUsage(usageManager);
328            }
329        }
330    
331        @Override
332        public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) {
333            super.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
334            for (PendingMessageCursor cursor : storePrefetches) {
335                cursor.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
336            }
337        }
338    
339        @Override
340        public void setMaxProducersToAudit(int maxProducersToAudit) {
341            super.setMaxProducersToAudit(maxProducersToAudit);
342            for (PendingMessageCursor cursor : storePrefetches) {
343                cursor.setMaxAuditDepth(maxAuditDepth);
344            }
345        }
346    
347        @Override
348        public void setMaxAuditDepth(int maxAuditDepth) {
349            super.setMaxAuditDepth(maxAuditDepth);
350            for (PendingMessageCursor cursor : storePrefetches) {
351                cursor.setMaxAuditDepth(maxAuditDepth);
352            }
353        }
354    
355        @Override
356        public void setEnableAudit(boolean enableAudit) {
357            super.setEnableAudit(enableAudit);
358            for (PendingMessageCursor cursor : storePrefetches) {
359                cursor.setEnableAudit(enableAudit);
360            }
361        }
362    
363        @Override
364        public  void setUseCache(boolean useCache) {
365            super.setUseCache(useCache);
366            for (PendingMessageCursor cursor : storePrefetches) {
367                cursor.setUseCache(useCache);
368            }
369        }
370    
371        protected synchronized PendingMessageCursor getNextCursor() throws Exception {
372            if (currentCursor == null || currentCursor.isEmpty()) {
373                currentCursor = null;
374                for (PendingMessageCursor tsp : storePrefetches) {
375                    if (tsp.hasNext()) {
376                        currentCursor = tsp;
377                        break;
378                    }
379                }
380                // round-robin
381                if (storePrefetches.size()>1) {
382                    PendingMessageCursor first = storePrefetches.remove(0);
383                    storePrefetches.add(first);
384                }
385            }
386            return currentCursor;
387        }
388    
389        @Override
390        public String toString() {
391            return "StoreDurableSubscriber(" + clientId + ":" + subscriberName + ")";
392        }
393    
394        public boolean isImmediatePriorityDispatch() {
395            return immediatePriorityDispatch;
396        }
397    
398        public void setImmediatePriorityDispatch(boolean immediatePriorityDispatch) {
399            this.immediatePriorityDispatch = immediatePriorityDispatch;
400        }
401    
402    }