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                                LOG.trace("cached high priority ({} message: {}, current paged batch priority: {}, cache size: {}", new Object[]{ msg.getPriority(), msg.getMessageId(), tsp.getLastRecoveredPriority(), tsp.batchList.size()});
203                            }
204                        }
205                    }
206                }
207    
208            }
209        }
210    
211        @Override
212        public boolean isTransient() {
213            return subscription.isKeepDurableSubsActive();
214        }
215    
216        @Override
217        public void addMessageFirst(MessageReference node) throws Exception {
218            // for keep durable subs active, need to deal with redispatch
219            if (node != null) {
220                Message msg = node.getMessage();
221                if (!msg.isPersistent()) {
222                    nonPersistent.addMessageFirst(node);
223                } else {
224                    Destination dest = (Destination) msg.getRegionDestination();
225                    TopicStorePrefetch tsp = topics.get(dest);
226                    if (tsp != null) {
227                        tsp.addMessageFirst(node);
228                    }
229                }
230            }
231        }
232    
233        @Override
234        public synchronized void addRecoveredMessage(MessageReference node) throws Exception {
235            nonPersistent.addMessageLast(node);
236        }
237    
238        @Override
239        public synchronized void clear() {
240            for (PendingMessageCursor tsp : storePrefetches) {
241                tsp.clear();
242            }
243        }
244    
245        @Override
246        public synchronized boolean hasNext() {
247            boolean result = true;
248            if (result) {
249                try {
250                    currentCursor = getNextCursor();
251                } catch (Exception e) {
252                    LOG.error("Failed to get current cursor ", e);
253                    throw new RuntimeException(e);
254                }
255                result = currentCursor != null ? currentCursor.hasNext() : false;
256            }
257            return result;
258        }
259    
260        @Override
261        public synchronized MessageReference next() {
262            MessageReference result = currentCursor != null ? currentCursor.next() : null;
263            return result;
264        }
265    
266        @Override
267        public synchronized void remove() {
268            if (currentCursor != null) {
269                currentCursor.remove();
270            }
271        }
272    
273        @Override
274        public synchronized void remove(MessageReference node) {
275            for (PendingMessageCursor tsp : storePrefetches) {
276                tsp.remove(node);
277            }
278        }
279    
280        @Override
281        public synchronized void reset() {
282            for (PendingMessageCursor storePrefetch : storePrefetches) {
283                storePrefetch.reset();
284            }
285        }
286    
287        @Override
288        public synchronized void release() {
289            this.currentCursor = null;
290            for (PendingMessageCursor storePrefetch : storePrefetches) {
291                storePrefetch.release();
292            }
293        }
294    
295        @Override
296        public synchronized int size() {
297            int pendingCount=0;
298            for (PendingMessageCursor tsp : storePrefetches) {
299                pendingCount += tsp.size();
300            }
301            return pendingCount;
302        }
303    
304        @Override
305        public void setMaxBatchSize(int newMaxBatchSize) {
306            for (PendingMessageCursor storePrefetch : storePrefetches) {
307                storePrefetch.setMaxBatchSize(newMaxBatchSize);
308            }
309            super.setMaxBatchSize(newMaxBatchSize);
310        }
311    
312        @Override
313        public synchronized void gc() {
314            for (PendingMessageCursor tsp : storePrefetches) {
315                tsp.gc();
316            }
317        }
318    
319        @Override
320        public void setSystemUsage(SystemUsage usageManager) {
321            super.setSystemUsage(usageManager);
322            for (PendingMessageCursor tsp : storePrefetches) {
323                tsp.setSystemUsage(usageManager);
324            }
325        }
326    
327        @Override
328        public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) {
329            super.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
330            for (PendingMessageCursor cursor : storePrefetches) {
331                cursor.setMemoryUsageHighWaterMark(memoryUsageHighWaterMark);
332            }
333        }
334    
335        @Override
336        public void setMaxProducersToAudit(int maxProducersToAudit) {
337            super.setMaxProducersToAudit(maxProducersToAudit);
338            for (PendingMessageCursor cursor : storePrefetches) {
339                cursor.setMaxAuditDepth(maxAuditDepth);
340            }
341        }
342    
343        @Override
344        public void setMaxAuditDepth(int maxAuditDepth) {
345            super.setMaxAuditDepth(maxAuditDepth);
346            for (PendingMessageCursor cursor : storePrefetches) {
347                cursor.setMaxAuditDepth(maxAuditDepth);
348            }
349        }
350    
351        @Override
352        public void setEnableAudit(boolean enableAudit) {
353            super.setEnableAudit(enableAudit);
354            for (PendingMessageCursor cursor : storePrefetches) {
355                cursor.setEnableAudit(enableAudit);
356            }
357        }
358    
359        @Override
360        public  void setUseCache(boolean useCache) {
361            super.setUseCache(useCache);
362            for (PendingMessageCursor cursor : storePrefetches) {
363                cursor.setUseCache(useCache);
364            }
365        }
366    
367        protected synchronized PendingMessageCursor getNextCursor() throws Exception {
368            if (currentCursor == null || currentCursor.isEmpty()) {
369                currentCursor = null;
370                for (PendingMessageCursor tsp : storePrefetches) {
371                    if (tsp.hasNext()) {
372                        currentCursor = tsp;
373                        break;
374                    }
375                }
376                // round-robin
377                if (storePrefetches.size()>1) {
378                    PendingMessageCursor first = storePrefetches.remove(0);
379                    storePrefetches.add(first);
380                }
381            }
382            return currentCursor;
383        }
384    
385        @Override
386        public String toString() {
387            return "StoreDurableSubscriber(" + clientId + ":" + subscriberName + ")";
388        }
389    
390        public boolean isImmediatePriorityDispatch() {
391            return immediatePriorityDispatch;
392        }
393    
394        public void setImmediatePriorityDispatch(boolean immediatePriorityDispatch) {
395            this.immediatePriorityDispatch = immediatePriorityDispatch;
396        }
397    
398    }