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.concurrent.atomic.AtomicLong;
024
025import javax.jms.JMSException;
026
027import org.apache.activemq.ActiveMQMessageAudit;
028import org.apache.activemq.broker.Broker;
029import org.apache.activemq.broker.ConnectionContext;
030import org.apache.activemq.broker.region.cursors.FilePendingMessageCursor;
031import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
032import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
033import org.apache.activemq.broker.region.policy.MessageEvictionStrategy;
034import org.apache.activemq.broker.region.policy.OldestMessageEvictionStrategy;
035import org.apache.activemq.command.ConsumerControl;
036import org.apache.activemq.command.ConsumerInfo;
037import org.apache.activemq.command.Message;
038import org.apache.activemq.command.MessageAck;
039import org.apache.activemq.command.MessageDispatch;
040import org.apache.activemq.command.MessageDispatchNotification;
041import org.apache.activemq.command.MessageId;
042import org.apache.activemq.command.MessagePull;
043import org.apache.activemq.command.Response;
044import org.apache.activemq.thread.Scheduler;
045import org.apache.activemq.transaction.Synchronization;
046import org.apache.activemq.transport.TransmitCallback;
047import org.apache.activemq.usage.SystemUsage;
048import org.slf4j.Logger;
049import org.slf4j.LoggerFactory;
050
051public class TopicSubscription extends AbstractSubscription {
052
053    private static final Logger LOG = LoggerFactory.getLogger(TopicSubscription.class);
054    private static final AtomicLong CURSOR_NAME_COUNTER = new AtomicLong(0);
055
056    protected PendingMessageCursor matched;
057    protected final SystemUsage usageManager;
058    boolean singleDestination = true;
059    Destination destination;
060    private final Scheduler scheduler;
061
062    private int maximumPendingMessages = -1;
063    private MessageEvictionStrategy messageEvictionStrategy = new OldestMessageEvictionStrategy();
064    private int discarded;
065    private final Object matchedListMutex = new Object();
066    private int memoryUsageHighWaterMark = 95;
067    // allow duplicate suppression in a ring network of brokers
068    protected int maxProducersToAudit = 1024;
069    protected int maxAuditDepth = 1000;
070    protected boolean enableAudit = false;
071    protected ActiveMQMessageAudit audit;
072    protected boolean active = false;
073    protected boolean discarding = false;
074
075    //Used for inflight message size calculations
076    protected final Object dispatchLock = new Object();
077    protected final List<MessageReference> dispatched = new ArrayList<MessageReference>();
078
079    public TopicSubscription(Broker broker,ConnectionContext context, ConsumerInfo info, SystemUsage usageManager) throws Exception {
080        super(broker, context, info);
081        this.usageManager = usageManager;
082        String matchedName = "TopicSubscription:" + CURSOR_NAME_COUNTER.getAndIncrement() + "[" + info.getConsumerId().toString() + "]";
083        if (info.getDestination().isTemporary() || broker.getTempDataStore()==null ) {
084            this.matched = new VMPendingMessageCursor(false);
085        } else {
086            this.matched = new FilePendingMessageCursor(broker,matchedName,false);
087        }
088
089        this.scheduler = broker.getScheduler();
090    }
091
092    public void init() throws Exception {
093        this.matched.setSystemUsage(usageManager);
094        this.matched.setMemoryUsageHighWaterMark(getCursorMemoryHighWaterMark());
095        this.matched.start();
096        if (enableAudit) {
097            audit= new ActiveMQMessageAudit(maxAuditDepth, maxProducersToAudit);
098        }
099        this.active=true;
100    }
101
102    @Override
103    public void add(MessageReference node) throws Exception {
104        if (isDuplicate(node)) {
105            return;
106        }
107        // Lets use an indirect reference so that we can associate a unique
108        // locator /w the message.
109        node = new IndirectMessageReference(node.getMessage());
110        getSubscriptionStatistics().getEnqueues().increment();
111        synchronized (matchedListMutex) {
112            // if this subscriber is already discarding a message, we don't want to add
113            // any more messages to it as those messages can only be advisories generated in the process,
114            // which can trigger the recursive call loop
115            if (discarding) return;
116
117            if (!isFull() && matched.isEmpty()) {
118                // if maximumPendingMessages is set we will only discard messages which
119                // have not been dispatched (i.e. we allow the prefetch buffer to be filled)
120                dispatch(node);
121                setSlowConsumer(false);
122            } else {
123                if (info.getPrefetchSize() > 1 && matched.size() > info.getPrefetchSize()) {
124                    // Slow consumers should log and set their state as such.
125                    if (!isSlowConsumer()) {
126                        LOG.warn("{}: has twice its prefetch limit pending, without an ack; it appears to be slow", toString());
127                        setSlowConsumer(true);
128                        for (Destination dest: destinations) {
129                            dest.slowConsumer(getContext(), this);
130                        }
131                    }
132                }
133                if (maximumPendingMessages != 0) {
134                    boolean warnedAboutWait = false;
135                    while (active) {
136                        while (matched.isFull()) {
137                            if (getContext().getStopping().get()) {
138                                LOG.warn("{}: stopped waiting for space in pendingMessage cursor for: {}", toString(), node.getMessageId());
139                                getSubscriptionStatistics().getEnqueues().decrement();
140                                return;
141                            }
142                            if (!warnedAboutWait) {
143                                LOG.info("{}: Pending message cursor [{}] is full, temp usag ({}%) or memory usage ({}%) limit reached, blocking message add() pending the release of resources.",
144                                        new Object[]{
145                                                toString(),
146                                                matched,
147                                                matched.getSystemUsage().getTempUsage().getPercentUsage(),
148                                                matched.getSystemUsage().getMemoryUsage().getPercentUsage()
149                                        });
150                                warnedAboutWait = true;
151                            }
152                            matchedListMutex.wait(20);
153                        }
154                        // Temporary storage could be full - so just try to add the message
155                        // see https://issues.apache.org/activemq/browse/AMQ-2475
156                        if (matched.tryAddMessageLast(node, 10)) {
157                            break;
158                        }
159                    }
160                    if (maximumPendingMessages > 0) {
161                        // calculate the high water mark from which point we
162                        // will eagerly evict expired messages
163                        int max = messageEvictionStrategy.getEvictExpiredMessagesHighWatermark();
164                        if (maximumPendingMessages > 0 && maximumPendingMessages < max) {
165                            max = maximumPendingMessages;
166                        }
167                        if (!matched.isEmpty() && matched.size() > max) {
168                            removeExpiredMessages();
169                        }
170                        // lets discard old messages as we are a slow consumer
171                        while (!matched.isEmpty() && matched.size() > maximumPendingMessages) {
172                            int pageInSize = matched.size() - maximumPendingMessages;
173                            // only page in a 1000 at a time - else we could blow the memory
174                            pageInSize = Math.max(1000, pageInSize);
175                            LinkedList<MessageReference> list = null;
176                            MessageReference[] oldMessages=null;
177                            synchronized(matched){
178                                list = matched.pageInList(pageInSize);
179                                oldMessages = messageEvictionStrategy.evictMessages(list);
180                                for (MessageReference ref : list) {
181                                    ref.decrementReferenceCount();
182                                }
183                            }
184                            int messagesToEvict = 0;
185                            if (oldMessages != null){
186                                messagesToEvict = oldMessages.length;
187                                for (int i = 0; i < messagesToEvict; i++) {
188                                    MessageReference oldMessage = oldMessages[i];
189                                    discard(oldMessage);
190                                }
191                            }
192                            // lets avoid an infinite loop if we are given a bad eviction strategy
193                            // for a bad strategy lets just not evict
194                            if (messagesToEvict == 0) {
195                                LOG.warn("No messages to evict returned for {} from eviction strategy: {} out of {} candidates", new Object[]{
196                                        destination, messageEvictionStrategy, list.size()
197                                });
198                                break;
199                            }
200                        }
201                    }
202                    dispatchMatched();
203                }
204            }
205        }
206    }
207
208    private boolean isDuplicate(MessageReference node) {
209        boolean duplicate = false;
210        if (enableAudit && audit != null) {
211            duplicate = audit.isDuplicate(node);
212            if (LOG.isDebugEnabled()) {
213                if (duplicate) {
214                    LOG.debug("{}, ignoring duplicate add: {}", this, node.getMessageId());
215                }
216            }
217        }
218        return duplicate;
219    }
220
221    /**
222     * Discard any expired messages from the matched list. Called from a
223     * synchronized block.
224     *
225     * @throws IOException
226     */
227    protected void removeExpiredMessages() throws IOException {
228        try {
229            matched.reset();
230            while (matched.hasNext()) {
231                MessageReference node = matched.next();
232                node.decrementReferenceCount();
233                if (node.isExpired()) {
234                    matched.remove();
235                    getSubscriptionStatistics().getDispatched().increment();
236                    node.decrementReferenceCount();
237                    if (broker.isExpired(node)) {
238                        ((Destination) node.getRegionDestination()).getDestinationStatistics().getExpired().increment();
239                        broker.messageExpired(getContext(), node, this);
240                    }
241                    break;
242                }
243            }
244        } finally {
245            matched.release();
246        }
247    }
248
249    @Override
250    public void processMessageDispatchNotification(MessageDispatchNotification mdn) {
251        synchronized (matchedListMutex) {
252            try {
253                matched.reset();
254                while (matched.hasNext()) {
255                    MessageReference node = matched.next();
256                    node.decrementReferenceCount();
257                    if (node.getMessageId().equals(mdn.getMessageId())) {
258                        synchronized(dispatchLock) {
259                            matched.remove();
260                            getSubscriptionStatistics().getDispatched().increment();
261                            dispatched.add(node);
262                            getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
263                            node.decrementReferenceCount();
264                        }
265                        break;
266                    }
267                }
268            } finally {
269                matched.release();
270            }
271        }
272    }
273
274    @Override
275    public synchronized void acknowledge(final ConnectionContext context, final MessageAck ack) throws Exception {
276        super.acknowledge(context, ack);
277
278        if (ack.isStandardAck()) {
279            updateStatsOnAck(context, ack);
280        } else if (ack.isPoisonAck()) {
281            if (ack.isInTransaction()) {
282                throw new JMSException("Poison ack cannot be transacted: " + ack);
283            }
284            updateStatsOnAck(context, ack);
285            if (getPrefetchSize() != 0) {
286                decrementPrefetchExtension(ack.getMessageCount());
287            }
288        } else if (ack.isIndividualAck()) {
289            updateStatsOnAck(context, ack);
290            if (getPrefetchSize() != 0 && ack.isInTransaction()) {
291                incrementPrefetchExtension(ack.getMessageCount());
292            }
293        } else if (ack.isExpiredAck()) {
294            updateStatsOnAck(ack);
295            if (getPrefetchSize() != 0) {
296                incrementPrefetchExtension(ack.getMessageCount());
297            }
298        } else if (ack.isDeliveredAck()) {
299            // Message was delivered but not acknowledged: update pre-fetch counters.
300            if (getPrefetchSize() != 0) {
301                incrementPrefetchExtension(ack.getMessageCount());
302            }
303        } else if (ack.isRedeliveredAck()) {
304            // No processing for redelivered needed
305            return;
306        } else {
307            throw new JMSException("Invalid acknowledgment: " + ack);
308        }
309
310        dispatchMatched();
311    }
312
313    private void updateStatsOnAck(final ConnectionContext context, final MessageAck ack) {
314        if (context.isInTransaction()) {
315            context.getTransaction().addSynchronization(new Synchronization() {
316
317                @Override
318                public void beforeEnd() {
319                    if (getPrefetchSize() != 0) {
320                        decrementPrefetchExtension(ack.getMessageCount());
321                    }
322                }
323
324                @Override
325                public void afterCommit() throws Exception {
326                    updateStatsOnAck(ack);
327                    dispatchMatched();
328                }
329            });
330        } else {
331            updateStatsOnAck(ack);
332        }
333    }
334
335    @Override
336    public Response pullMessage(ConnectionContext context, final MessagePull pull) throws Exception {
337
338        // The slave should not deliver pull messages.
339        if (getPrefetchSize() == 0) {
340
341            final long currentDispatchedCount = getSubscriptionStatistics().getDispatched().getCount();
342            prefetchExtension.set(pull.getQuantity());
343            dispatchMatched();
344
345            // If there was nothing dispatched.. we may need to setup a timeout.
346            if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || pull.isAlwaysSignalDone()) {
347
348                // immediate timeout used by receiveNoWait()
349                if (pull.getTimeout() == -1) {
350                    // Send a NULL message to signal nothing pending.
351                    dispatch(null);
352                    prefetchExtension.set(0);
353                }
354
355                if (pull.getTimeout() > 0) {
356                    scheduler.executeAfterDelay(new Runnable() {
357
358                        @Override
359                        public void run() {
360                            pullTimeout(currentDispatchedCount, pull.isAlwaysSignalDone());
361                        }
362                    }, pull.getTimeout());
363                }
364            }
365        }
366        return null;
367    }
368
369    /**
370     * Occurs when a pull times out. If nothing has been dispatched since the
371     * timeout was setup, then send the NULL message.
372     */
373    private final void pullTimeout(long currentDispatchedCount, boolean alwaysSendDone) {
374        synchronized (matchedListMutex) {
375            if (currentDispatchedCount == getSubscriptionStatistics().getDispatched().getCount() || alwaysSendDone) {
376                try {
377                    dispatch(null);
378                } catch (Exception e) {
379                    context.getConnection().serviceException(e);
380                } finally {
381                    prefetchExtension.set(0);
382                }
383            }
384        }
385    }
386
387    /**
388     * Update the statistics on message ack.
389     * @param ack
390     */
391    private void updateStatsOnAck(final MessageAck ack) {
392        synchronized(dispatchLock) {
393            boolean inAckRange = false;
394            List<MessageReference> removeList = new ArrayList<MessageReference>();
395            for (final MessageReference node : dispatched) {
396                MessageId messageId = node.getMessageId();
397                if (ack.getFirstMessageId() == null
398                        || ack.getFirstMessageId().equals(messageId)) {
399                    inAckRange = true;
400                }
401                if (inAckRange) {
402                    removeList.add(node);
403                    if (ack.getLastMessageId().equals(messageId)) {
404                        break;
405                    }
406                }
407            }
408
409            for (final MessageReference node : removeList) {
410                dispatched.remove(node);
411                getSubscriptionStatistics().getInflightMessageSize().addSize(-node.getSize());
412                getSubscriptionStatistics().getDequeues().increment();
413                ((Destination)node.getRegionDestination()).getDestinationStatistics().getDequeues().increment();
414                ((Destination)node.getRegionDestination()).getDestinationStatistics().getInflight().decrement();
415                if (info.isNetworkSubscription()) {
416                    ((Destination)node.getRegionDestination()).getDestinationStatistics().getForwards().add(ack.getMessageCount());
417                }
418                if (ack.isExpiredAck()) {
419                    destination.getDestinationStatistics().getExpired().add(ack.getMessageCount());
420                }
421            }
422        }
423    }
424
425    private void incrementPrefetchExtension(int amount) {
426        while (true) {
427            int currentExtension = prefetchExtension.get();
428            int newExtension = Math.max(0, currentExtension + amount);
429            if (prefetchExtension.compareAndSet(currentExtension, newExtension)) {
430                break;
431            }
432        }
433    }
434
435    private void decrementPrefetchExtension(int amount) {
436        while (true) {
437            int currentExtension = prefetchExtension.get();
438            int newExtension = Math.max(0, currentExtension - amount);
439            if (prefetchExtension.compareAndSet(currentExtension, newExtension)) {
440                break;
441            }
442        }
443    }
444
445    @Override
446    public int countBeforeFull() {
447        return getPrefetchSize() == 0 ? prefetchExtension.get() : info.getPrefetchSize() + prefetchExtension.get() - getDispatchedQueueSize();
448    }
449
450    @Override
451    public int getPendingQueueSize() {
452        return matched();
453    }
454
455    @Override
456    public long getPendingMessageSize() {
457        synchronized (matchedListMutex) {
458            return matched.messageSize();
459        }
460    }
461
462    @Override
463    public int getDispatchedQueueSize() {
464        return (int)(getSubscriptionStatistics().getDispatched().getCount() -
465                     getSubscriptionStatistics().getDequeues().getCount());
466    }
467
468    public int getMaximumPendingMessages() {
469        return maximumPendingMessages;
470    }
471
472    @Override
473    public long getDispatchedCounter() {
474        return getSubscriptionStatistics().getDispatched().getCount();
475    }
476
477    @Override
478    public long getEnqueueCounter() {
479        return getSubscriptionStatistics().getEnqueues().getCount();
480    }
481
482    @Override
483    public long getDequeueCounter() {
484        return getSubscriptionStatistics().getDequeues().getCount();
485    }
486
487    /**
488     * @return the number of messages discarded due to being a slow consumer
489     */
490    public int discarded() {
491        synchronized (matchedListMutex) {
492            return discarded;
493        }
494    }
495
496    /**
497     * @return the number of matched messages (messages targeted for the
498     *         subscription but not yet able to be dispatched due to the
499     *         prefetch buffer being full).
500     */
501    public int matched() {
502        synchronized (matchedListMutex) {
503            return matched.size();
504        }
505    }
506
507    /**
508     * Sets the maximum number of pending messages that can be matched against
509     * this consumer before old messages are discarded.
510     */
511    public void setMaximumPendingMessages(int maximumPendingMessages) {
512        this.maximumPendingMessages = maximumPendingMessages;
513    }
514
515    public MessageEvictionStrategy getMessageEvictionStrategy() {
516        return messageEvictionStrategy;
517    }
518
519    /**
520     * Sets the eviction strategy used to decide which message to evict when the
521     * slow consumer needs to discard messages
522     */
523    public void setMessageEvictionStrategy(MessageEvictionStrategy messageEvictionStrategy) {
524        this.messageEvictionStrategy = messageEvictionStrategy;
525    }
526
527    public int getMaxProducersToAudit() {
528        return maxProducersToAudit;
529    }
530
531    public synchronized void setMaxProducersToAudit(int maxProducersToAudit) {
532        this.maxProducersToAudit = maxProducersToAudit;
533        if (audit != null) {
534            audit.setMaximumNumberOfProducersToTrack(maxProducersToAudit);
535        }
536    }
537
538    public int getMaxAuditDepth() {
539        return maxAuditDepth;
540    }
541
542    public synchronized void setMaxAuditDepth(int maxAuditDepth) {
543        this.maxAuditDepth = maxAuditDepth;
544        if (audit != null) {
545            audit.setAuditDepth(maxAuditDepth);
546        }
547    }
548
549    public boolean isEnableAudit() {
550        return enableAudit;
551    }
552
553    public synchronized void setEnableAudit(boolean enableAudit) {
554        this.enableAudit = enableAudit;
555        if (enableAudit && audit == null) {
556            audit = new ActiveMQMessageAudit(maxAuditDepth,maxProducersToAudit);
557        }
558    }
559
560    // Implementation methods
561    // -------------------------------------------------------------------------
562    @Override
563    public boolean isFull() {
564        return getPrefetchSize() == 0 ? prefetchExtension.get() == 0 : getDispatchedQueueSize() - prefetchExtension.get() >= info.getPrefetchSize();
565    }
566
567    @Override
568    public int getInFlightSize() {
569        return getDispatchedQueueSize();
570    }
571
572    /**
573     * @return true when 60% or more room is left for dispatching messages
574     */
575    @Override
576    public boolean isLowWaterMark() {
577        return (getDispatchedQueueSize() - prefetchExtension.get()) <= (info.getPrefetchSize() * .4);
578    }
579
580    /**
581     * @return true when 10% or less room is left for dispatching messages
582     */
583    @Override
584    public boolean isHighWaterMark() {
585        return (getDispatchedQueueSize() - prefetchExtension.get()) >= (info.getPrefetchSize() * .9);
586    }
587
588    /**
589     * @param memoryUsageHighWaterMark the memoryUsageHighWaterMark to set
590     */
591    public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark) {
592        this.memoryUsageHighWaterMark = memoryUsageHighWaterMark;
593    }
594
595    /**
596     * @return the memoryUsageHighWaterMark
597     */
598    public int getMemoryUsageHighWaterMark() {
599        return this.memoryUsageHighWaterMark;
600    }
601
602    /**
603     * @return the usageManager
604     */
605    public SystemUsage getUsageManager() {
606        return this.usageManager;
607    }
608
609    /**
610     * @return the matched
611     */
612    public PendingMessageCursor getMatched() {
613        return this.matched;
614    }
615
616    /**
617     * @param matched the matched to set
618     */
619    public void setMatched(PendingMessageCursor matched) {
620        this.matched = matched;
621    }
622
623    /**
624     * inform the MessageConsumer on the client to change it's prefetch
625     *
626     * @param newPrefetch
627     */
628    @Override
629    public void updateConsumerPrefetch(int newPrefetch) {
630        if (context != null && context.getConnection() != null && context.getConnection().isManageable()) {
631            ConsumerControl cc = new ConsumerControl();
632            cc.setConsumerId(info.getConsumerId());
633            cc.setPrefetch(newPrefetch);
634            context.getConnection().dispatchAsync(cc);
635        }
636    }
637
638    private void dispatchMatched() throws IOException {
639        synchronized (matchedListMutex) {
640            if (!matched.isEmpty() && !isFull()) {
641                try {
642                    matched.reset();
643
644                    while (matched.hasNext() && !isFull()) {
645                        MessageReference message = matched.next();
646                        message.decrementReferenceCount();
647                        matched.remove();
648                        // Message may have been sitting in the matched list a while
649                        // waiting for the consumer to ak the message.
650                        if (message.isExpired()) {
651                            discard(message);
652                            continue; // just drop it.
653                        }
654                        dispatch(message);
655                    }
656                } finally {
657                    matched.release();
658                }
659            }
660        }
661    }
662
663    private void dispatch(final MessageReference node) throws IOException {
664        Message message = node != null ? node.getMessage() : null;
665        if (node != null) {
666            node.incrementReferenceCount();
667        }
668        // Make sure we can dispatch a message.
669        MessageDispatch md = new MessageDispatch();
670        md.setMessage(message);
671        md.setConsumerId(info.getConsumerId());
672        if (node != null) {
673            md.setDestination(((Destination)node.getRegionDestination()).getActiveMQDestination());
674            synchronized(dispatchLock) {
675                getSubscriptionStatistics().getDispatched().increment();
676                dispatched.add(node);
677                getSubscriptionStatistics().getInflightMessageSize().addSize(node.getSize());
678            }
679
680            // Keep track if this subscription is receiving messages from a single destination.
681            if (singleDestination) {
682                if (destination == null) {
683                    destination = (Destination)node.getRegionDestination();
684                } else {
685                    if (destination != node.getRegionDestination()) {
686                        singleDestination = false;
687                    }
688                }
689            }
690
691            if (getPrefetchSize() == 0) {
692                decrementPrefetchExtension(1);
693            }
694        }
695
696        if (info.isDispatchAsync()) {
697            if (node != null) {
698                md.setTransmitCallback(new TransmitCallback() {
699
700                    @Override
701                    public void onSuccess() {
702                        Destination regionDestination = (Destination) node.getRegionDestination();
703                        regionDestination.getDestinationStatistics().getDispatched().increment();
704                        regionDestination.getDestinationStatistics().getInflight().increment();
705                        node.decrementReferenceCount();
706                    }
707
708                    @Override
709                    public void onFailure() {
710                        Destination regionDestination = (Destination) node.getRegionDestination();
711                        regionDestination.getDestinationStatistics().getDispatched().increment();
712                        regionDestination.getDestinationStatistics().getInflight().increment();
713                        node.decrementReferenceCount();
714                    }
715                });
716            }
717            context.getConnection().dispatchAsync(md);
718        } else {
719            context.getConnection().dispatchSync(md);
720            if (node != null) {
721                Destination regionDestination = (Destination) node.getRegionDestination();
722                regionDestination.getDestinationStatistics().getDispatched().increment();
723                regionDestination.getDestinationStatistics().getInflight().increment();
724                node.decrementReferenceCount();
725            }
726        }
727    }
728
729    private void discard(MessageReference message) {
730        discarding = true;
731        try {
732            message.decrementReferenceCount();
733            matched.remove(message);
734            discarded++;
735            if (destination != null) {
736                destination.getDestinationStatistics().getDequeues().increment();
737            }
738            LOG.debug("{}, discarding message {}", this, message);
739            Destination dest = (Destination) message.getRegionDestination();
740            if (dest != null) {
741                dest.messageDiscarded(getContext(), this, message);
742            }
743            broker.getRoot().sendToDeadLetterQueue(getContext(), message, this, new Throwable("TopicSubDiscard. ID:" + info.getConsumerId()));
744        } finally {
745            discarding = false;
746        }
747    }
748
749    @Override
750    public String toString() {
751        return "TopicSubscription:" + " consumer=" + info.getConsumerId() + ", destinations=" + destinations.size() + ", dispatched=" + getDispatchedQueueSize() + ", delivered="
752                + getDequeueCounter() + ", matched=" + matched() + ", discarded=" + discarded() + ", prefetchExtension=" + prefetchExtension.get();
753    }
754
755    @Override
756    public void destroy() {
757        this.active=false;
758        synchronized (matchedListMutex) {
759            try {
760                matched.destroy();
761            } catch (Exception e) {
762                LOG.warn("Failed to destroy cursor", e);
763            }
764        }
765        setSlowConsumer(false);
766        synchronized(dispatchLock) {
767            dispatched.clear();
768        }
769    }
770
771    @Override
772    public int getPrefetchSize() {
773        return info.getPrefetchSize();
774    }
775
776    @Override
777    public void setPrefetchSize(int newSize) {
778        info.setPrefetchSize(newSize);
779        try {
780            dispatchMatched();
781        } catch(Exception e) {
782            LOG.trace("Caught exception on dispatch after prefetch size change.");
783        }
784    }
785}