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