Source code

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 */
017package org.apache.activemq.broker.region;
018
019import java.io.IOException;
020import java.util.ArrayList;
021import java.util.LinkedList;
022import java.util.List;
023import java.util.Map;
024import java.util.concurrent.CancellationException;
025import java.util.concurrent.ConcurrentHashMap;
026import java.util.concurrent.ConcurrentMap;
027import java.util.concurrent.CopyOnWriteArrayList;
028import java.util.concurrent.Future;
029import java.util.concurrent.locks.ReentrantReadWriteLock;
030
031import org.apache.activemq.advisory.AdvisorySupport;
032import org.apache.activemq.broker.BrokerService;
033import org.apache.activemq.broker.ConnectionContext;
034import org.apache.activemq.broker.ProducerBrokerExchange;
035import org.apache.activemq.broker.region.policy.DispatchPolicy;
036import org.apache.activemq.broker.region.policy.LastImageSubscriptionRecoveryPolicy;
037import org.apache.activemq.broker.region.policy.RetainedMessageSubscriptionRecoveryPolicy;
038import org.apache.activemq.broker.region.policy.SimpleDispatchPolicy;
039import org.apache.activemq.broker.region.policy.SubscriptionRecoveryPolicy;
040import org.apache.activemq.broker.util.InsertionCountList;
041import org.apache.activemq.command.ActiveMQDestination;
042import org.apache.activemq.command.ConsumerInfo;
043import org.apache.activemq.command.ExceptionResponse;
044import org.apache.activemq.command.Message;
045import org.apache.activemq.command.MessageAck;
046import org.apache.activemq.command.MessageId;
047import org.apache.activemq.command.ProducerAck;
048import org.apache.activemq.command.ProducerInfo;
049import org.apache.activemq.command.Response;
050import org.apache.activemq.command.SubscriptionInfo;
051import org.apache.activemq.filter.MessageEvaluationContext;
052import org.apache.activemq.filter.NonCachedMessageEvaluationContext;
053import org.apache.activemq.store.MessageRecoveryListener;
054import org.apache.activemq.store.TopicMessageStore;
055import org.apache.activemq.thread.Task;
056import org.apache.activemq.thread.TaskRunner;
057import org.apache.activemq.thread.TaskRunnerFactory;
058import org.apache.activemq.transaction.Synchronization;
059import org.apache.activemq.util.SubscriptionKey;
060import org.slf4j.Logger;
061import org.slf4j.LoggerFactory;
062
063/**
064 * The Topic is a destination that sends a copy of a message to every active
065 * Subscription registered.
066 */
067public class Topic extends BaseDestination implements Task {
068    protected static final Logger LOG = LoggerFactory.getLogger(Topic.class);
069    private final TopicMessageStore topicStore;
070    protected final CopyOnWriteArrayList<Subscription> consumers = new CopyOnWriteArrayList<Subscription>();
071    private final ReentrantReadWriteLock dispatchLock = new ReentrantReadWriteLock();
072    private DispatchPolicy dispatchPolicy = new SimpleDispatchPolicy();
073    private SubscriptionRecoveryPolicy subscriptionRecoveryPolicy;
074    private final ConcurrentMap<SubscriptionKey, DurableTopicSubscription> durableSubscribers = new ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription>();
075    private final TaskRunner taskRunner;
076    private final LinkedList<Runnable> messagesWaitingForSpace = new LinkedList<Runnable>();
077    private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() {
078        @Override
079        public void run() {
080            try {
081                Topic.this.taskRunner.wakeup();
082            } catch (InterruptedException e) {
083            }
084        };
085    };
086
087    public Topic(BrokerService brokerService, ActiveMQDestination destination, TopicMessageStore store,
088            DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception {
089        super(brokerService, store, destination, parentStats);
090        this.topicStore = store;
091        subscriptionRecoveryPolicy = new RetainedMessageSubscriptionRecoveryPolicy(null);
092        this.taskRunner = taskFactory.createTaskRunner(this, "Topic  " + destination.getPhysicalName());
093    }
094
095    @Override
096    public void initialize() throws Exception {
097        super.initialize();
098        // set non default subscription recovery policy (override policyEntries)
099        if (AdvisorySupport.isMasterBrokerAdvisoryTopic(destination)) {
100            subscriptionRecoveryPolicy = new LastImageSubscriptionRecoveryPolicy();
101            setAlwaysRetroactive(true);
102        }
103        if (store != null) {
104            // AMQ-2586: Better to leave this stat at zero than to give the user
105            // misleading metrics.
106            // int messageCount = store.getMessageCount();
107            // destinationStatistics.getMessages().setCount(messageCount);
108            store.start();
109        }
110    }
111
112    @Override
113    public List<Subscription> getConsumers() {
114        synchronized (consumers) {
115            return new ArrayList<Subscription>(consumers);
116        }
117    }
118
119    public boolean lock(MessageReference node, LockOwner sub) {
120        return true;
121    }
122
123    @Override
124    public void addSubscription(ConnectionContext context, final Subscription sub) throws Exception {
125        if (!sub.getConsumerInfo().isDurable()) {
126
127            // Do a retroactive recovery if needed.
128            if (sub.getConsumerInfo().isRetroactive() || isAlwaysRetroactive()) {
129
130                // synchronize with dispatch method so that no new messages are sent
131                // while we are recovering a subscription to avoid out of order messages.
132                dispatchLock.writeLock().lock();
133                try {
134                    boolean applyRecovery = false;
135                    synchronized (consumers) {
136                        if (!consumers.contains(sub)){
137                            sub.add(context, this);
138                            consumers.add(sub);
139                            applyRecovery=true;
140                            super.addSubscription(context, sub);
141                        }
142                    }
143                    if (applyRecovery){
144                        subscriptionRecoveryPolicy.recover(context, this, sub);
145                    }
146                } finally {
147                    dispatchLock.writeLock().unlock();
148                }
149
150            } else {
151                synchronized (consumers) {
152                    if (!consumers.contains(sub)){
153                        sub.add(context, this);
154                        consumers.add(sub);
155                        super.addSubscription(context, sub);
156                    }
157                }
158            }
159        } else {
160            DurableTopicSubscription dsub = (DurableTopicSubscription) sub;
161            super.addSubscription(context, sub);
162            sub.add(context, this);
163            if(dsub.isActive()) {
164                synchronized (consumers) {
165                    boolean hasSubscription = false;
166
167                    if (consumers.size() == 0) {
168                        hasSubscription = false;
169                    } else {
170                        for (Subscription currentSub : consumers) {
171                            if (currentSub.getConsumerInfo().isDurable()) {
172                                DurableTopicSubscription dcurrentSub = (DurableTopicSubscription) currentSub;
173                                if (dcurrentSub.getSubscriptionKey().equals(dsub.getSubscriptionKey())) {
174                                    hasSubscription = true;
175                                    break;
176                                }
177                            }
178                        }
179                    }
180
181                    if (!hasSubscription) {
182                        consumers.add(sub);
183                    }
184                }
185            }
186            durableSubscribers.put(dsub.getSubscriptionKey(), dsub);
187        }
188    }
189
190    @Override
191    public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) throws Exception {
192        if (!sub.getConsumerInfo().isDurable()) {
193            super.removeSubscription(context, sub, lastDeliveredSequenceId);
194            synchronized (consumers) {
195                consumers.remove(sub);
196            }
197        }
198        sub.remove(context, this);
199    }
200
201    public void deleteSubscription(ConnectionContext context, SubscriptionKey key) throws Exception {
202        if (topicStore != null) {
203            topicStore.deleteSubscription(key.clientId, key.subscriptionName);
204            DurableTopicSubscription removed = durableSubscribers.remove(key);
205            if (removed != null) {
206                destinationStatistics.getConsumers().decrement();
207                // deactivate and remove
208                removed.deactivate(false, 0l);
209                consumers.remove(removed);
210            }
211        }
212    }
213
214    private boolean hasDurableSubChanged(SubscriptionInfo info1, ConsumerInfo info2) {
215        if (info1.getSelector() != null ^ info2.getSelector() != null) {
216            return true;
217        }
218        if (info1.getSelector() != null && !info1.getSelector().equals(info2.getSelector())) {
219            return true;
220        }
221
222        return false;
223    }
224
225    public void activate(ConnectionContext context, final DurableTopicSubscription subscription) throws Exception {
226        // synchronize with dispatch method so that no new messages are sent
227        // while we are recovering a subscription to avoid out of order messages.
228        dispatchLock.writeLock().lock();
229        try {
230
231            if (topicStore == null) {
232                return;
233            }
234
235            // Recover the durable subscription.
236            String clientId = subscription.getSubscriptionKey().getClientId();
237            String subscriptionName = subscription.getSubscriptionKey().getSubscriptionName();
238            SubscriptionInfo info = topicStore.lookupSubscription(clientId, subscriptionName);
239            if (info != null) {
240                // Check to see if selector changed.
241                if (hasDurableSubChanged(info, subscription.getConsumerInfo())) {
242                    // Need to delete the subscription
243                    topicStore.deleteSubscription(clientId, subscriptionName);
244                    info = null;
245                    synchronized (consumers) {
246                        consumers.remove(subscription);
247                    }
248                } else {
249                    synchronized (consumers) {
250                        if (!consumers.contains(subscription)) {
251                            consumers.add(subscription);
252                        }
253                    }
254                }
255            }
256
257            // Do we need to create the subscription?
258            if (info == null) {
259                info = new SubscriptionInfo();
260                info.setClientId(clientId);
261                info.setSelector(subscription.getConsumerInfo().getSelector());
262                info.setSubscriptionName(subscriptionName);
263                info.setDestination(getActiveMQDestination());
264                info.setNoLocal(subscription.getConsumerInfo().isNoLocal());
265                // This destination is an actual destination id.
266                info.setSubscribedDestination(subscription.getConsumerInfo().getDestination());
267                // This destination might be a pattern
268                synchronized (consumers) {
269                    consumers.add(subscription);
270                    topicStore.addSubscription(info, subscription.getConsumerInfo().isRetroactive());
271                }
272            }
273
274            final MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext();
275            msgContext.setDestination(destination);
276            if (subscription.isRecoveryRequired()) {
277                topicStore.recoverSubscription(clientId, subscriptionName, new MessageRecoveryListener() {
278                    @Override
279                    public boolean recoverMessage(Message message) throws Exception {
280                        message.setRegionDestination(Topic.this);
281                        try {
282                            msgContext.setMessageReference(message);
283                            if (subscription.matches(message, msgContext)) {
284                                subscription.add(message);
285                            }
286                        } catch (IOException e) {
287                            LOG.error("Failed to recover this message {}", message, e);
288                        }
289                        return true;
290                    }
291
292                    @Override
293                    public boolean recoverMessageReference(MessageId messageReference) throws Exception {
294                        throw new RuntimeException("Should not be called.");
295                    }
296
297                    @Override
298                    public boolean hasSpace() {
299                        return true;
300                    }
301
302                    @Override
303                    public boolean isDuplicate(MessageId id) {
304                        return false;
305                    }
306                });
307            }
308        } finally {
309            dispatchLock.writeLock().unlock();
310        }
311    }
312
313    public void deactivate(ConnectionContext context, DurableTopicSubscription sub, List<MessageReference> dispatched) throws Exception {
314        synchronized (consumers) {
315            consumers.remove(sub);
316        }
317        sub.remove(context, this, dispatched);
318    }
319
320    public void recoverRetroactiveMessages(ConnectionContext context, Subscription subscription) throws Exception {
321        if (subscription.getConsumerInfo().isRetroactive()) {
322            subscriptionRecoveryPolicy.recover(context, this, subscription);
323        }
324    }
325
326    @Override
327    public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception {
328        final ConnectionContext context = producerExchange.getConnectionContext();
329
330        final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo();
331        producerExchange.incrementSend();
332        final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0
333                && !context.isInRecoveryMode();
334
335        // There is delay between the client sending it and it arriving at the
336        // destination.. it may have expired.
337        if (message.isExpired()) {
338            broker.messageExpired(context, message, null);
339            getDestinationStatistics().getExpired().increment();
340            if (sendProducerAck) {
341                ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
342                context.getConnection().dispatchAsync(ack);
343            }
344            return;
345        }
346
347        if (memoryUsage.isFull()) {
348            isFull(context, memoryUsage);
349            fastProducer(context, producerInfo);
350
351            if (isProducerFlowControl() && context.isProducerFlowControl()) {
352
353                if (warnOnProducerFlowControl) {
354                    warnOnProducerFlowControl = false;
355                    LOG.info("{}, Usage Manager memory limit reached {}. Producers will be throttled to the rate at which messages are removed from this destination to prevent flooding it. See http://activemq.apache.org/producer-flow-control.html for more info.",
356                            getActiveMQDestination().getQualifiedName(), memoryUsage.getLimit());
357                }
358
359                if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) {
360                    throw new javax.jms.ResourceAllocationException("Usage Manager memory limit ("
361                            + memoryUsage.getLimit() + ") reached. Rejecting send for producer (" + message.getProducerId()
362                            + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
363                            + " See http://activemq.apache.org/producer-flow-control.html for more info");
364                }
365
366                // We can avoid blocking due to low usage if the producer is sending a sync message or
367                // if it is using a producer window
368                if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) {
369                    synchronized (messagesWaitingForSpace) {
370                        messagesWaitingForSpace.add(new Runnable() {
371                            @Override
372                            public void run() {
373                                try {
374
375                                    // While waiting for space to free up... the
376                                    // message may have expired.
377                                    if (message.isExpired()) {
378                                        broker.messageExpired(context, message, null);
379                                        getDestinationStatistics().getExpired().increment();
380                                    } else {
381                                        doMessageSend(producerExchange, message);
382                                    }
383
384                                    if (sendProducerAck) {
385                                        ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message
386                                                .getSize());
387                                        context.getConnection().dispatchAsync(ack);
388                                    } else {
389                                        Response response = new Response();
390                                        response.setCorrelationId(message.getCommandId());
391                                        context.getConnection().dispatchAsync(response);
392                                    }
393
394                                } catch (Exception e) {
395                                    if (!sendProducerAck && !context.isInRecoveryMode()) {
396                                        ExceptionResponse response = new ExceptionResponse(e);
397                                        response.setCorrelationId(message.getCommandId());
398                                        context.getConnection().dispatchAsync(response);
399                                    }
400                                }
401                            }
402                        });
403
404                        registerCallbackForNotFullNotification();
405                        context.setDontSendReponse(true);
406                        return;
407                    }
408
409                } else {
410                    // Producer flow control cannot be used, so we have do the flow control
411                    // at the broker by blocking this thread until there is space available.
412
413                    if (memoryUsage.isFull()) {
414                        if (context.isInTransaction()) {
415
416                            int count = 0;
417                            while (!memoryUsage.waitForSpace(1000)) {
418                                if (context.getStopping().get()) {
419                                    throw new IOException("Connection closed, send aborted.");
420                                }
421                                if (count > 2 && context.isInTransaction()) {
422                                    count = 0;
423                                    int size = context.getTransaction().size();
424                                    LOG.warn("Waiting for space to send transacted message - transaction elements = {} need more space to commit. Message = {}", size, message);
425                                }
426                                count++;
427                            }
428                        } else {
429                            waitForSpace(
430                                    context,
431                                    producerExchange,
432                                    memoryUsage,
433                                    "Usage Manager Memory Usage limit reached. Stopping producer ("
434                                            + message.getProducerId()
435                                            + ") to prevent flooding "
436                                            + getActiveMQDestination().getQualifiedName()
437                                            + "."
438                                            + " See http://activemq.apache.org/producer-flow-control.html for more info");
439                        }
440                    }
441
442                    // The usage manager could have delayed us by the time
443                    // we unblock the message could have expired..
444                    if (message.isExpired()) {
445                        getDestinationStatistics().getExpired().increment();
446                        LOG.debug("Expired message: {}", message);
447                        return;
448                    }
449                }
450            }
451        }
452
453        doMessageSend(producerExchange, message);
454        messageDelivered(context, message);
455        if (sendProducerAck) {
456            ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize());
457            context.getConnection().dispatchAsync(ack);
458        }
459    }
460
461    /**
462     * do send the message - this needs to be synchronized to ensure messages
463     * are stored AND dispatched in the right order
464     *
465     * @param producerExchange
466     * @param message
467     * @throws IOException
468     * @throws Exception
469     */
470    synchronized void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message)
471            throws IOException, Exception {
472        final ConnectionContext context = producerExchange.getConnectionContext();
473        message.setRegionDestination(this);
474        message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
475        Future<Object> result = null;
476
477        if (topicStore != null && message.isPersistent() && !canOptimizeOutPersistence()) {
478            if (systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) {
479                final String logMessage = "Persistent store is Full, " + getStoreUsageHighWaterMark() + "% of "
480                        + systemUsage.getStoreUsage().getLimit() + ". Stopping producer (" + message.getProducerId()
481                        + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "."
482                        + " See http://activemq.apache.org/producer-flow-control.html for more info";
483                if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) {
484                    throw new javax.jms.ResourceAllocationException(logMessage);
485                }
486
487                waitForSpace(context,producerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage);
488            }
489            result = topicStore.asyncAddTopicMessage(context, message,isOptimizeStorage());
490        }
491
492        message.incrementReferenceCount();
493
494        if (context.isInTransaction()) {
495            context.getTransaction().addSynchronization(new Synchronization() {
496                @Override
497                public void afterCommit() throws Exception {
498                    // It could take while before we receive the commit
499                    // operation.. by that time the message could have
500                    // expired..
501                    if (broker.isExpired(message)) {
502                        getDestinationStatistics().getExpired().increment();
503                        broker.messageExpired(context, message, null);
504                        message.decrementReferenceCount();
505                        return;
506                    }
507                    try {
508                        dispatch(context, message);
509                    } finally {
510                        message.decrementReferenceCount();
511                    }
512                }
513
514                @Override
515                public void afterRollback() throws Exception {
516                    message.decrementReferenceCount();
517                }
518            });
519
520        } else {
521            try {
522                dispatch(context, message);
523            } finally {
524                message.decrementReferenceCount();
525            }
526        }
527
528        if (result != null && !result.isCancelled()) {
529            try {
530                result.get();
531            } catch (CancellationException e) {
532                // ignore - the task has been cancelled if the message
533                // has already been deleted
534            }
535        }
536    }
537
538    private boolean canOptimizeOutPersistence() {
539        return durableSubscribers.size() == 0;
540    }
541
542    @Override
543    public String toString() {
544        return "Topic: destination=" + destination.getPhysicalName() + ", subscriptions=" + consumers.size();
545    }
546
547    @Override
548    public void acknowledge(ConnectionContext context, Subscription sub, final MessageAck ack,
549            final MessageReference node) throws IOException {
550        if (topicStore != null && node.isPersistent()) {
551            DurableTopicSubscription dsub = (DurableTopicSubscription) sub;
552            SubscriptionKey key = dsub.getSubscriptionKey();
553            topicStore.acknowledge(context, key.getClientId(), key.getSubscriptionName(), node.getMessageId(),
554                    convertToNonRangedAck(ack, node));
555        }
556        messageConsumed(context, node);
557    }
558
559    @Override
560    public void gc() {
561    }
562
563    public Message loadMessage(MessageId messageId) throws IOException {
564        return topicStore != null ? topicStore.getMessage(messageId) : null;
565    }
566
567    @Override
568    public void start() throws Exception {
569        this.subscriptionRecoveryPolicy.start();
570        if (memoryUsage != null) {
571            memoryUsage.start();
572        }
573
574        if (getExpireMessagesPeriod() > 0 && !AdvisorySupport.isAdvisoryTopic(getActiveMQDestination())) {
575            scheduler.executePeriodically(expireMessagesTask, getExpireMessagesPeriod());
576        }
577    }
578
579    @Override
580    public void stop() throws Exception {
581        if (taskRunner != null) {
582            taskRunner.shutdown();
583        }
584        this.subscriptionRecoveryPolicy.stop();
585        if (memoryUsage != null) {
586            memoryUsage.stop();
587        }
588        if (this.topicStore != null) {
589            this.topicStore.stop();
590        }
591
592         scheduler.cancel(expireMessagesTask);
593    }
594
595    @Override
596    public Message[] browse() {
597        final List<Message> result = new ArrayList<Message>();
598        doBrowse(result, getMaxBrowsePageSize());
599        return result.toArray(new Message[result.size()]);
600    }
601
602    private void doBrowse(final List<Message> browseList, final int max) {
603        try {
604            if (topicStore != null) {
605                final List<Message> toExpire = new ArrayList<Message>();
606                topicStore.recover(new MessageRecoveryListener() {
607                    @Override
608                    public boolean recoverMessage(Message message) throws Exception {
609                        if (message.isExpired()) {
610                            toExpire.add(message);
611                        }
612                        browseList.add(message);
613                        return true;
614                    }
615
616                    @Override
617                    public boolean recoverMessageReference(MessageId messageReference) throws Exception {
618                        return true;
619                    }
620
621                    @Override
622                    public boolean hasSpace() {
623                        return browseList.size() < max;
624                    }
625
626                    @Override
627                    public boolean isDuplicate(MessageId id) {
628                        return false;
629                    }
630                });
631                final ConnectionContext connectionContext = createConnectionContext();
632                for (Message message : toExpire) {
633                    for (DurableTopicSubscription sub : durableSubscribers.values()) {
634                        if (!sub.isActive()) {
635                            messageExpired(connectionContext, sub, message);
636                        }
637                    }
638                }
639                Message[] msgs = subscriptionRecoveryPolicy.browse(getActiveMQDestination());
640                if (msgs != null) {
641                    for (int i = 0; i < msgs.length && browseList.size() < max; i++) {
642                        browseList.add(msgs[i]);
643                    }
644                }
645            }
646        } catch (Throwable e) {
647            LOG.warn("Failed to browse Topic: {}", getActiveMQDestination().getPhysicalName(), e);
648        }
649    }
650
651    @Override
652    public boolean iterate() {
653        synchronized (messagesWaitingForSpace) {
654            while (!memoryUsage.isFull() && !messagesWaitingForSpace.isEmpty()) {
655                Runnable op = messagesWaitingForSpace.removeFirst();
656                op.run();
657            }
658
659            if (!messagesWaitingForSpace.isEmpty()) {
660                registerCallbackForNotFullNotification();
661            }
662        }
663        return false;
664    }
665
666    private void registerCallbackForNotFullNotification() {
667        // If the usage manager is not full, then the task will not
668        // get called..
669        if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) {
670            // so call it directly here.
671            sendMessagesWaitingForSpaceTask.run();
672        }
673    }
674
675    // Properties
676    // -------------------------------------------------------------------------
677
678    public DispatchPolicy getDispatchPolicy() {
679        return dispatchPolicy;
680    }
681
682    public void setDispatchPolicy(DispatchPolicy dispatchPolicy) {
683        this.dispatchPolicy = dispatchPolicy;
684    }
685
686    public SubscriptionRecoveryPolicy getSubscriptionRecoveryPolicy() {
687        return subscriptionRecoveryPolicy;
688    }
689
690    public void setSubscriptionRecoveryPolicy(SubscriptionRecoveryPolicy recoveryPolicy) {
691        if (this.subscriptionRecoveryPolicy != null && this.subscriptionRecoveryPolicy instanceof RetainedMessageSubscriptionRecoveryPolicy) {
692            // allow users to combine retained message policy with other ActiveMQ policies
693            RetainedMessageSubscriptionRecoveryPolicy policy = (RetainedMessageSubscriptionRecoveryPolicy) this.subscriptionRecoveryPolicy;
694            policy.setWrapped(recoveryPolicy);
695        } else {
696            this.subscriptionRecoveryPolicy = recoveryPolicy;
697        }
698    }
699
700    // Implementation methods
701    // -------------------------------------------------------------------------
702
703    @Override
704    public final void wakeup() {
705    }
706
707    protected void dispatch(final ConnectionContext context, Message message) throws Exception {
708        // AMQ-2586: Better to leave this stat at zero than to give the user
709        // misleading metrics.
710        // destinationStatistics.getMessages().increment();
711        destinationStatistics.getEnqueues().increment();
712        destinationStatistics.getMessageSize().addSize(message.getSize());
713        MessageEvaluationContext msgContext = null;
714
715        dispatchLock.readLock().lock();
716        try {
717            if (!subscriptionRecoveryPolicy.add(context, message)) {
718                return;
719            }
720            synchronized (consumers) {
721                if (consumers.isEmpty()) {
722                    onMessageWithNoConsumers(context, message);
723                    return;
724                }
725            }
726            msgContext = context.getMessageEvaluationContext();
727            msgContext.setDestination(destination);
728            msgContext.setMessageReference(message);
729            if (!dispatchPolicy.dispatch(message, msgContext, consumers)) {
730                onMessageWithNoConsumers(context, message);
731            }
732
733        } finally {
734            dispatchLock.readLock().unlock();
735            if (msgContext != null) {
736                msgContext.clear();
737            }
738        }
739    }
740
741    private final Runnable expireMessagesTask = new Runnable() {
742        @Override
743        public void run() {
744            List<Message> browsedMessages = new InsertionCountList<Message>();
745            doBrowse(browsedMessages, getMaxExpirePageSize());
746        }
747    };
748
749    @Override
750    public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) {
751        broker.messageExpired(context, reference, subs);
752        // AMQ-2586: Better to leave this stat at zero than to give the user
753        // misleading metrics.
754        // destinationStatistics.getMessages().decrement();
755        destinationStatistics.getExpired().increment();
756        MessageAck ack = new MessageAck();
757        ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
758        ack.setDestination(destination);
759        ack.setMessageID(reference.getMessageId());
760        try {
761            if (subs instanceof DurableTopicSubscription) {
762                ((DurableTopicSubscription)subs).removePending(reference);
763            }
764            acknowledge(context, subs, ack, reference);
765        } catch (Exception e) {
766            LOG.error("Failed to remove expired Message from the store ", e);
767        }
768    }
769
770    @Override
771    protected Logger getLog() {
772        return LOG;
773    }
774
775    protected boolean isOptimizeStorage(){
776        boolean result = false;
777
778        if (isDoOptimzeMessageStorage() && durableSubscribers.isEmpty()==false){
779                result = true;
780                for (DurableTopicSubscription s : durableSubscribers.values()) {
781                    if (s.isActive()== false){
782                        result = false;
783                        break;
784                    }
785                    if (s.getPrefetchSize()==0){
786                        result = false;
787                        break;
788                    }
789                    if (s.isSlowConsumer()){
790                        result = false;
791                        break;
792                    }
793                    if (s.getInFlightUsage() > getOptimizeMessageStoreInFlightLimit()){
794                        result = false;
795                        break;
796                    }
797                }
798        }
799        return result;
800    }
801
802    /**
803     * force a reread of the store - after transaction recovery completion
804     */
805    @Override
806    public void clearPendingMessages() {
807        dispatchLock.readLock().lock();
808        try {
809            for (DurableTopicSubscription durableTopicSubscription : durableSubscribers.values()) {
810                clearPendingAndDispatch(durableTopicSubscription);
811            }
812        } finally {
813            dispatchLock.readLock().unlock();
814        }
815    }
816
817    private void clearPendingAndDispatch(DurableTopicSubscription durableTopicSubscription) {
818        synchronized (durableTopicSubscription.pendingLock) {
819            durableTopicSubscription.pending.clear();
820            try {
821                durableTopicSubscription.dispatchPending();
822            } catch (IOException exception) {
823                LOG.warn("After clear of pending, failed to dispatch to: {}, for: {}, pending: {}", new Object[]{
824                        durableTopicSubscription,
825                        destination,
826                        durableTopicSubscription.pending }, exception);
827            }
828        }
829    }
830
831    public Map<SubscriptionKey, DurableTopicSubscription> getDurableTopicSubs() {
832        return durableSubscribers;
833    }
834}