| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.activemq.broker.region; |
| |
| import java.io.IOException; |
| import java.util.ArrayList; |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.CancellationException; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.CopyOnWriteArrayList; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import org.apache.activemq.advisory.AdvisorySupport; |
| import org.apache.activemq.broker.BrokerService; |
| import org.apache.activemq.broker.ConnectionContext; |
| import org.apache.activemq.broker.ProducerBrokerExchange; |
| import org.apache.activemq.broker.region.policy.DispatchPolicy; |
| import org.apache.activemq.broker.region.policy.LastImageSubscriptionRecoveryPolicy; |
| import org.apache.activemq.broker.region.policy.RetainedMessageSubscriptionRecoveryPolicy; |
| import org.apache.activemq.broker.region.policy.SimpleDispatchPolicy; |
| import org.apache.activemq.broker.region.policy.SubscriptionRecoveryPolicy; |
| import org.apache.activemq.broker.util.InsertionCountList; |
| import org.apache.activemq.command.ActiveMQDestination; |
| import org.apache.activemq.command.ConsumerInfo; |
| import org.apache.activemq.command.ExceptionResponse; |
| import org.apache.activemq.command.Message; |
| import org.apache.activemq.command.MessageAck; |
| import org.apache.activemq.command.MessageId; |
| import org.apache.activemq.command.ProducerAck; |
| import org.apache.activemq.command.ProducerInfo; |
| import org.apache.activemq.command.Response; |
| import org.apache.activemq.command.SubscriptionInfo; |
| import org.apache.activemq.filter.MessageEvaluationContext; |
| import org.apache.activemq.filter.NonCachedMessageEvaluationContext; |
| import org.apache.activemq.store.MessageRecoveryListener; |
| import org.apache.activemq.store.NoLocalSubscriptionAware; |
| import org.apache.activemq.store.PersistenceAdapter; |
| import org.apache.activemq.store.TopicMessageStore; |
| import org.apache.activemq.thread.Task; |
| import org.apache.activemq.thread.TaskRunner; |
| import org.apache.activemq.thread.TaskRunnerFactory; |
| import org.apache.activemq.transaction.Synchronization; |
| import org.apache.activemq.util.SubscriptionKey; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.jms.JMSException; |
| |
| import static org.apache.activemq.transaction.Transaction.IN_USE_STATE; |
| |
| /** |
| * The Topic is a destination that sends a copy of a message to every active |
| * Subscription registered. |
| */ |
| public class Topic extends BaseDestination implements Task { |
| protected static final Logger LOG = LoggerFactory.getLogger(Topic.class); |
| private final TopicMessageStore topicStore; |
| protected final CopyOnWriteArrayList<Subscription> consumers = new CopyOnWriteArrayList<Subscription>(); |
| private final ReentrantReadWriteLock dispatchLock = new ReentrantReadWriteLock(); |
| private DispatchPolicy dispatchPolicy = new SimpleDispatchPolicy(); |
| private SubscriptionRecoveryPolicy subscriptionRecoveryPolicy; |
| private final ConcurrentMap<SubscriptionKey, DurableTopicSubscription> durableSubscribers = new ConcurrentHashMap<SubscriptionKey, DurableTopicSubscription>(); |
| private final TaskRunner taskRunner; |
| private final LinkedList<Runnable> messagesWaitingForSpace = new LinkedList<Runnable>(); |
| private final Runnable sendMessagesWaitingForSpaceTask = new Runnable() { |
| @Override |
| public void run() { |
| try { |
| Topic.this.taskRunner.wakeup(); |
| } catch (InterruptedException e) { |
| } |
| } |
| }; |
| |
| public Topic(BrokerService brokerService, ActiveMQDestination destination, TopicMessageStore store, |
| DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception { |
| super(brokerService, store, destination, parentStats); |
| this.topicStore = store; |
| subscriptionRecoveryPolicy = new RetainedMessageSubscriptionRecoveryPolicy(null); |
| this.taskRunner = taskFactory.createTaskRunner(this, "Topic " + destination.getPhysicalName()); |
| } |
| |
| @Override |
| public void initialize() throws Exception { |
| super.initialize(); |
| // set non default subscription recovery policy (override policyEntries) |
| if (AdvisorySupport.isMasterBrokerAdvisoryTopic(destination)) { |
| subscriptionRecoveryPolicy = new LastImageSubscriptionRecoveryPolicy(); |
| setAlwaysRetroactive(true); |
| } |
| if (store != null) { |
| // AMQ-2586: Better to leave this stat at zero than to give the user |
| // misleading metrics. |
| // int messageCount = store.getMessageCount(); |
| // destinationStatistics.getMessages().setCount(messageCount); |
| store.start(); |
| } |
| } |
| |
| @Override |
| public List<Subscription> getConsumers() { |
| synchronized (consumers) { |
| return new ArrayList<Subscription>(consumers); |
| } |
| } |
| |
| public boolean lock(MessageReference node, LockOwner sub) { |
| return true; |
| } |
| |
| @Override |
| public void addSubscription(ConnectionContext context, final Subscription sub) throws Exception { |
| if (!sub.getConsumerInfo().isDurable()) { |
| |
| // Do a retroactive recovery if needed. |
| if (sub.getConsumerInfo().isRetroactive() || isAlwaysRetroactive()) { |
| |
| // synchronize with dispatch method so that no new messages are sent |
| // while we are recovering a subscription to avoid out of order messages. |
| dispatchLock.writeLock().lock(); |
| try { |
| boolean applyRecovery = false; |
| synchronized (consumers) { |
| if (!consumers.contains(sub)){ |
| sub.add(context, this); |
| consumers.add(sub); |
| applyRecovery=true; |
| super.addSubscription(context, sub); |
| } |
| } |
| if (applyRecovery){ |
| subscriptionRecoveryPolicy.recover(context, this, sub); |
| } |
| } finally { |
| dispatchLock.writeLock().unlock(); |
| } |
| |
| } else { |
| synchronized (consumers) { |
| if (!consumers.contains(sub)){ |
| sub.add(context, this); |
| consumers.add(sub); |
| super.addSubscription(context, sub); |
| } |
| } |
| } |
| } else { |
| DurableTopicSubscription dsub = (DurableTopicSubscription) sub; |
| super.addSubscription(context, sub); |
| sub.add(context, this); |
| if(dsub.isActive()) { |
| synchronized (consumers) { |
| boolean hasSubscription = false; |
| |
| if (consumers.size() == 0) { |
| hasSubscription = false; |
| } else { |
| for (Subscription currentSub : consumers) { |
| if (currentSub.getConsumerInfo().isDurable()) { |
| DurableTopicSubscription dcurrentSub = (DurableTopicSubscription) currentSub; |
| if (dcurrentSub.getSubscriptionKey().equals(dsub.getSubscriptionKey())) { |
| hasSubscription = true; |
| break; |
| } |
| } |
| } |
| } |
| |
| if (!hasSubscription) { |
| consumers.add(sub); |
| } |
| } |
| } |
| durableSubscribers.put(dsub.getSubscriptionKey(), dsub); |
| } |
| } |
| |
| @Override |
| public void removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId) throws Exception { |
| if (!sub.getConsumerInfo().isDurable()) { |
| boolean removed = false; |
| synchronized (consumers) { |
| removed = consumers.remove(sub); |
| } |
| if (removed) { |
| super.removeSubscription(context, sub, lastDeliveredSequenceId); |
| } |
| } |
| sub.remove(context, this); |
| } |
| |
| public void deleteSubscription(ConnectionContext context, SubscriptionKey key) throws Exception { |
| if (topicStore != null) { |
| topicStore.deleteSubscription(key.clientId, key.subscriptionName); |
| DurableTopicSubscription removed = durableSubscribers.remove(key); |
| if (removed != null) { |
| destinationStatistics.getConsumers().decrement(); |
| // deactivate and remove |
| removed.deactivate(false, 0l); |
| consumers.remove(removed); |
| } |
| } |
| } |
| |
| private boolean hasDurableSubChanged(SubscriptionInfo info1, ConsumerInfo info2) throws IOException { |
| if (hasSelectorChanged(info1, info2)) { |
| return true; |
| } |
| |
| return hasNoLocalChanged(info1, info2); |
| } |
| |
| private boolean hasNoLocalChanged(SubscriptionInfo info1, ConsumerInfo info2) throws IOException { |
| //Not all persistence adapters store the noLocal value for a subscription |
| PersistenceAdapter adapter = broker.getBrokerService().getPersistenceAdapter(); |
| if (adapter instanceof NoLocalSubscriptionAware) { |
| if (info1.isNoLocal() ^ info2.isNoLocal()) { |
| return true; |
| } |
| } |
| |
| return false; |
| } |
| |
| private boolean hasSelectorChanged(SubscriptionInfo info1, ConsumerInfo info2) { |
| if (info1.getSelector() != null ^ info2.getSelector() != null) { |
| return true; |
| } |
| |
| if (info1.getSelector() != null && !info1.getSelector().equals(info2.getSelector())) { |
| return true; |
| } |
| |
| return false; |
| } |
| |
| public void activate(ConnectionContext context, final DurableTopicSubscription subscription) throws Exception { |
| // synchronize with dispatch method so that no new messages are sent |
| // while we are recovering a subscription to avoid out of order messages. |
| dispatchLock.writeLock().lock(); |
| try { |
| |
| if (topicStore == null) { |
| return; |
| } |
| |
| // Recover the durable subscription. |
| String clientId = subscription.getSubscriptionKey().getClientId(); |
| String subscriptionName = subscription.getSubscriptionKey().getSubscriptionName(); |
| SubscriptionInfo info = topicStore.lookupSubscription(clientId, subscriptionName); |
| if (info != null) { |
| // Check to see if selector changed. |
| if (hasDurableSubChanged(info, subscription.getConsumerInfo())) { |
| // Need to delete the subscription |
| topicStore.deleteSubscription(clientId, subscriptionName); |
| info = null; |
| // Force a rebuild of the selector chain for the subscription otherwise |
| // the stored subscription is updated but the selector expression is not |
| // and the subscription will not behave according to the new configuration. |
| subscription.setSelector(subscription.getConsumerInfo().getSelector()); |
| synchronized (consumers) { |
| consumers.remove(subscription); |
| } |
| } else { |
| synchronized (consumers) { |
| if (!consumers.contains(subscription)) { |
| consumers.add(subscription); |
| } |
| } |
| } |
| } |
| |
| // Do we need to create the subscription? |
| if (info == null) { |
| info = new SubscriptionInfo(); |
| info.setClientId(clientId); |
| info.setSelector(subscription.getConsumerInfo().getSelector()); |
| info.setSubscriptionName(subscriptionName); |
| info.setDestination(getActiveMQDestination()); |
| info.setNoLocal(subscription.getConsumerInfo().isNoLocal()); |
| // This destination is an actual destination id. |
| info.setSubscribedDestination(subscription.getConsumerInfo().getDestination()); |
| // This destination might be a pattern |
| synchronized (consumers) { |
| consumers.add(subscription); |
| topicStore.addSubscription(info, subscription.getConsumerInfo().isRetroactive()); |
| } |
| } |
| |
| final MessageEvaluationContext msgContext = new NonCachedMessageEvaluationContext(); |
| msgContext.setDestination(destination); |
| if (subscription.isRecoveryRequired()) { |
| topicStore.recoverSubscription(clientId, subscriptionName, new MessageRecoveryListener() { |
| @Override |
| public boolean recoverMessage(Message message) throws Exception { |
| message.setRegionDestination(Topic.this); |
| try { |
| msgContext.setMessageReference(message); |
| if (subscription.matches(message, msgContext)) { |
| subscription.add(message); |
| } |
| } catch (IOException e) { |
| LOG.error("Failed to recover this message {}", message, e); |
| } |
| return true; |
| } |
| |
| @Override |
| public boolean recoverMessageReference(MessageId messageReference) throws Exception { |
| throw new RuntimeException("Should not be called."); |
| } |
| |
| @Override |
| public boolean hasSpace() { |
| return true; |
| } |
| |
| @Override |
| public boolean isDuplicate(MessageId id) { |
| return false; |
| } |
| }); |
| } |
| } finally { |
| dispatchLock.writeLock().unlock(); |
| } |
| } |
| |
| public void deactivate(ConnectionContext context, DurableTopicSubscription sub, List<MessageReference> dispatched) throws Exception { |
| synchronized (consumers) { |
| consumers.remove(sub); |
| } |
| sub.remove(context, this, dispatched); |
| } |
| |
| public void recoverRetroactiveMessages(ConnectionContext context, Subscription subscription) throws Exception { |
| if (subscription.getConsumerInfo().isRetroactive()) { |
| subscriptionRecoveryPolicy.recover(context, this, subscription); |
| } |
| } |
| |
| @Override |
| public void send(final ProducerBrokerExchange producerExchange, final Message message) throws Exception { |
| final ConnectionContext context = producerExchange.getConnectionContext(); |
| |
| final ProducerInfo producerInfo = producerExchange.getProducerState().getInfo(); |
| producerExchange.incrementSend(); |
| final boolean sendProducerAck = !message.isResponseRequired() && producerInfo.getWindowSize() > 0 |
| && !context.isInRecoveryMode(); |
| |
| message.setRegionDestination(this); |
| |
| // There is delay between the client sending it and it arriving at the |
| // destination.. it may have expired. |
| if (message.isExpired()) { |
| broker.messageExpired(context, message, null); |
| getDestinationStatistics().getExpired().increment(); |
| if (sendProducerAck) { |
| ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); |
| context.getConnection().dispatchAsync(ack); |
| } |
| return; |
| } |
| |
| if (memoryUsage.isFull()) { |
| isFull(context, memoryUsage); |
| fastProducer(context, producerInfo); |
| |
| if (isProducerFlowControl() && context.isProducerFlowControl()) { |
| |
| if (isFlowControlLogRequired()) { |
| 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.", |
| getActiveMQDestination().getQualifiedName(), memoryUsage.getLimit()); |
| } |
| |
| if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) { |
| throw new javax.jms.ResourceAllocationException("Usage Manager memory limit (" |
| + memoryUsage.getLimit() + ") reached. Rejecting send for producer (" + message.getProducerId() |
| + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." |
| + " See http://activemq.apache.org/producer-flow-control.html for more info"); |
| } |
| |
| // We can avoid blocking due to low usage if the producer is sending a sync message or |
| // if it is using a producer window |
| if (producerInfo.getWindowSize() > 0 || message.isResponseRequired()) { |
| synchronized (messagesWaitingForSpace) { |
| messagesWaitingForSpace.add(new Runnable() { |
| @Override |
| public void run() { |
| try { |
| |
| // While waiting for space to free up... |
| // the transaction may be done |
| if (message.isInTransaction()) { |
| if (context.getTransaction() == null || context.getTransaction().getState() > IN_USE_STATE) { |
| throw new JMSException("Send transaction completed while waiting for space"); |
| } |
| } |
| |
| // the message may have expired. |
| if (message.isExpired()) { |
| broker.messageExpired(context, message, null); |
| getDestinationStatistics().getExpired().increment(); |
| } else { |
| doMessageSend(producerExchange, message); |
| } |
| |
| if (sendProducerAck) { |
| ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message |
| .getSize()); |
| context.getConnection().dispatchAsync(ack); |
| } else { |
| Response response = new Response(); |
| response.setCorrelationId(message.getCommandId()); |
| context.getConnection().dispatchAsync(response); |
| } |
| |
| } catch (Exception e) { |
| if (!sendProducerAck && !context.isInRecoveryMode()) { |
| ExceptionResponse response = new ExceptionResponse(e); |
| response.setCorrelationId(message.getCommandId()); |
| context.getConnection().dispatchAsync(response); |
| } |
| } |
| } |
| }); |
| |
| registerCallbackForNotFullNotification(); |
| context.setDontSendReponse(true); |
| return; |
| } |
| |
| } else { |
| // Producer flow control cannot be used, so we have do the flow control |
| // at the broker by blocking this thread until there is space available. |
| |
| if (memoryUsage.isFull()) { |
| if (context.isInTransaction()) { |
| |
| int count = 0; |
| while (!memoryUsage.waitForSpace(1000)) { |
| if (context.getStopping().get()) { |
| throw new IOException("Connection closed, send aborted."); |
| } |
| if (count > 2 && context.isInTransaction()) { |
| count = 0; |
| int size = context.getTransaction().size(); |
| LOG.warn("Waiting for space to send transacted message - transaction elements = {} need more space to commit. Message = {}", size, message); |
| } |
| count++; |
| } |
| } else { |
| waitForSpace( |
| context, |
| producerExchange, |
| memoryUsage, |
| "Usage Manager Memory Usage limit reached. Stopping producer (" |
| + message.getProducerId() |
| + ") to prevent flooding " |
| + getActiveMQDestination().getQualifiedName() |
| + "." |
| + " See http://activemq.apache.org/producer-flow-control.html for more info"); |
| } |
| } |
| |
| // The usage manager could have delayed us by the time |
| // we unblock the message could have expired.. |
| if (message.isExpired()) { |
| getDestinationStatistics().getExpired().increment(); |
| LOG.debug("Expired message: {}", message); |
| return; |
| } |
| } |
| } |
| } |
| |
| doMessageSend(producerExchange, message); |
| messageDelivered(context, message); |
| if (sendProducerAck) { |
| ProducerAck ack = new ProducerAck(producerInfo.getProducerId(), message.getSize()); |
| context.getConnection().dispatchAsync(ack); |
| } |
| } |
| |
| /** |
| * do send the message - this needs to be synchronized to ensure messages |
| * are stored AND dispatched in the right order |
| * |
| * @param producerExchange |
| * @param message |
| * @throws IOException |
| * @throws Exception |
| */ |
| synchronized void doMessageSend(final ProducerBrokerExchange producerExchange, final Message message) |
| throws IOException, Exception { |
| final ConnectionContext context = producerExchange.getConnectionContext(); |
| message.getMessageId().setBrokerSequenceId(getDestinationSequenceId()); |
| Future<Object> result = null; |
| |
| if (topicStore != null && message.isPersistent() && !canOptimizeOutPersistence()) { |
| if (systemUsage.getStoreUsage().isFull(getStoreUsageHighWaterMark())) { |
| final String logMessage = "Persistent store is Full, " + getStoreUsageHighWaterMark() + "% of " |
| + systemUsage.getStoreUsage().getLimit() + ". Stopping producer (" + message.getProducerId() |
| + ") to prevent flooding " + getActiveMQDestination().getQualifiedName() + "." |
| + " See http://activemq.apache.org/producer-flow-control.html for more info"; |
| if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) { |
| throw new javax.jms.ResourceAllocationException(logMessage); |
| } |
| |
| waitForSpace(context,producerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage); |
| } |
| result = topicStore.asyncAddTopicMessage(context, message,isOptimizeStorage()); |
| |
| //Moved the reduceMemoryfootprint clearing to the dispatch method |
| } |
| |
| message.incrementReferenceCount(); |
| |
| if (context.isInTransaction() && (context.getTransaction() != null)) { |
| context.getTransaction().addSynchronization(new Synchronization() { |
| @Override |
| public void afterCommit() throws Exception { |
| // It could take while before we receive the commit |
| // operation.. by that time the message could have |
| // expired.. |
| if (message.isExpired()) { |
| if (broker.isExpired(message)) { |
| getDestinationStatistics().getExpired().increment(); |
| broker.messageExpired(context, message, null); |
| } |
| message.decrementReferenceCount(); |
| return; |
| } |
| try { |
| dispatch(context, message); |
| } finally { |
| message.decrementReferenceCount(); |
| } |
| } |
| |
| @Override |
| public void afterRollback() throws Exception { |
| message.decrementReferenceCount(); |
| } |
| }); |
| |
| } else { |
| try { |
| dispatch(context, message); |
| } finally { |
| message.decrementReferenceCount(); |
| } |
| } |
| |
| if (result != null && !result.isCancelled()) { |
| try { |
| result.get(); |
| } catch (CancellationException e) { |
| // ignore - the task has been cancelled if the message |
| // has already been deleted |
| } |
| } |
| } |
| |
| private boolean canOptimizeOutPersistence() { |
| return durableSubscribers.size() == 0; |
| } |
| |
| @Override |
| public String toString() { |
| return "Topic: destination=" + destination.getPhysicalName() + ", subscriptions=" + consumers.size(); |
| } |
| |
| @Override |
| public void acknowledge(ConnectionContext context, Subscription sub, final MessageAck ack, |
| final MessageReference node) throws IOException { |
| if (topicStore != null && node.isPersistent()) { |
| DurableTopicSubscription dsub = (DurableTopicSubscription) sub; |
| SubscriptionKey key = dsub.getSubscriptionKey(); |
| topicStore.acknowledge(context, key.getClientId(), key.getSubscriptionName(), node.getMessageId(), |
| convertToNonRangedAck(ack, node)); |
| } |
| messageConsumed(context, node); |
| } |
| |
| @Override |
| public void gc() { |
| } |
| |
| public Message loadMessage(MessageId messageId) throws IOException { |
| return topicStore != null ? topicStore.getMessage(messageId) : null; |
| } |
| |
| @Override |
| public void start() throws Exception { |
| if (started.compareAndSet(false, true)) { |
| this.subscriptionRecoveryPolicy.start(); |
| if (memoryUsage != null) { |
| memoryUsage.start(); |
| } |
| |
| if (getExpireMessagesPeriod() > 0 && !AdvisorySupport.isAdvisoryTopic(getActiveMQDestination())) { |
| scheduler.executePeriodically(expireMessagesTask, getExpireMessagesPeriod()); |
| } |
| } |
| } |
| |
| @Override |
| public void stop() throws Exception { |
| if (started.compareAndSet(true, false)) { |
| if (taskRunner != null) { |
| taskRunner.shutdown(); |
| } |
| this.subscriptionRecoveryPolicy.stop(); |
| if (memoryUsage != null) { |
| memoryUsage.stop(); |
| } |
| if (this.topicStore != null) { |
| this.topicStore.stop(); |
| } |
| |
| scheduler.cancel(expireMessagesTask); |
| } |
| } |
| |
| @Override |
| public Message[] browse() { |
| final List<Message> result = new ArrayList<Message>(); |
| doBrowse(result, getMaxBrowsePageSize()); |
| return result.toArray(new Message[result.size()]); |
| } |
| |
| private void doBrowse(final List<Message> browseList, final int max) { |
| try { |
| if (topicStore != null) { |
| final List<Message> toExpire = new ArrayList<Message>(); |
| topicStore.recover(new MessageRecoveryListener() { |
| @Override |
| public boolean recoverMessage(Message message) throws Exception { |
| if (message.isExpired()) { |
| toExpire.add(message); |
| } |
| browseList.add(message); |
| return true; |
| } |
| |
| @Override |
| public boolean recoverMessageReference(MessageId messageReference) throws Exception { |
| return true; |
| } |
| |
| @Override |
| public boolean hasSpace() { |
| return browseList.size() < max; |
| } |
| |
| @Override |
| public boolean isDuplicate(MessageId id) { |
| return false; |
| } |
| }); |
| final ConnectionContext connectionContext = createConnectionContext(); |
| for (Message message : toExpire) { |
| for (DurableTopicSubscription sub : durableSubscribers.values()) { |
| if (!sub.isActive()) { |
| message.setRegionDestination(this); |
| messageExpired(connectionContext, sub, message); |
| } |
| } |
| } |
| Message[] msgs = subscriptionRecoveryPolicy.browse(getActiveMQDestination()); |
| if (msgs != null) { |
| for (int i = 0; i < msgs.length && browseList.size() < max; i++) { |
| browseList.add(msgs[i]); |
| } |
| } |
| } |
| } catch (Throwable e) { |
| LOG.warn("Failed to browse Topic: {}", getActiveMQDestination().getPhysicalName(), e); |
| } |
| } |
| |
| @Override |
| public boolean iterate() { |
| synchronized (messagesWaitingForSpace) { |
| while (!memoryUsage.isFull() && !messagesWaitingForSpace.isEmpty()) { |
| Runnable op = messagesWaitingForSpace.removeFirst(); |
| op.run(); |
| } |
| |
| if (!messagesWaitingForSpace.isEmpty()) { |
| registerCallbackForNotFullNotification(); |
| } |
| } |
| return false; |
| } |
| |
| private void registerCallbackForNotFullNotification() { |
| // If the usage manager is not full, then the task will not |
| // get called.. |
| if (!memoryUsage.notifyCallbackWhenNotFull(sendMessagesWaitingForSpaceTask)) { |
| // so call it directly here. |
| sendMessagesWaitingForSpaceTask.run(); |
| } |
| } |
| |
| // Properties |
| // ------------------------------------------------------------------------- |
| |
| public DispatchPolicy getDispatchPolicy() { |
| return dispatchPolicy; |
| } |
| |
| public void setDispatchPolicy(DispatchPolicy dispatchPolicy) { |
| this.dispatchPolicy = dispatchPolicy; |
| } |
| |
| public SubscriptionRecoveryPolicy getSubscriptionRecoveryPolicy() { |
| return subscriptionRecoveryPolicy; |
| } |
| |
| public void setSubscriptionRecoveryPolicy(SubscriptionRecoveryPolicy recoveryPolicy) { |
| if (this.subscriptionRecoveryPolicy != null && this.subscriptionRecoveryPolicy instanceof RetainedMessageSubscriptionRecoveryPolicy) { |
| // allow users to combine retained message policy with other ActiveMQ policies |
| RetainedMessageSubscriptionRecoveryPolicy policy = (RetainedMessageSubscriptionRecoveryPolicy) this.subscriptionRecoveryPolicy; |
| policy.setWrapped(recoveryPolicy); |
| } else { |
| this.subscriptionRecoveryPolicy = recoveryPolicy; |
| } |
| } |
| |
| // Implementation methods |
| // ------------------------------------------------------------------------- |
| |
| @Override |
| public final void wakeup() { |
| } |
| |
| protected void dispatch(final ConnectionContext context, Message message) throws Exception { |
| // AMQ-2586: Better to leave this stat at zero than to give the user |
| // misleading metrics. |
| // destinationStatistics.getMessages().increment(); |
| destinationStatistics.getEnqueues().increment(); |
| destinationStatistics.getMessageSize().addSize(message.getSize()); |
| MessageEvaluationContext msgContext = null; |
| |
| dispatchLock.readLock().lock(); |
| try { |
| if (!subscriptionRecoveryPolicy.add(context, message)) { |
| return; |
| } |
| synchronized (consumers) { |
| if (consumers.isEmpty()) { |
| onMessageWithNoConsumers(context, message); |
| return; |
| } |
| } |
| |
| // Clear memory before dispatch - need to clear here because the call to |
| //subscriptionRecoveryPolicy.add() will unmarshall the state |
| if (isReduceMemoryFootprint() && message.isMarshalled()) { |
| message.clearUnMarshalledState(); |
| } |
| |
| msgContext = context.getMessageEvaluationContext(); |
| msgContext.setDestination(destination); |
| msgContext.setMessageReference(message); |
| if (!dispatchPolicy.dispatch(message, msgContext, consumers)) { |
| onMessageWithNoConsumers(context, message); |
| } |
| |
| } finally { |
| dispatchLock.readLock().unlock(); |
| if (msgContext != null) { |
| msgContext.clear(); |
| } |
| } |
| } |
| |
| private final Runnable expireMessagesTask = new Runnable() { |
| @Override |
| public void run() { |
| List<Message> browsedMessages = new InsertionCountList<Message>(); |
| doBrowse(browsedMessages, getMaxExpirePageSize()); |
| } |
| }; |
| |
| @Override |
| public void messageExpired(ConnectionContext context, Subscription subs, MessageReference reference) { |
| broker.messageExpired(context, reference, subs); |
| // AMQ-2586: Better to leave this stat at zero than to give the user |
| // misleading metrics. |
| // destinationStatistics.getMessages().decrement(); |
| destinationStatistics.getExpired().increment(); |
| MessageAck ack = new MessageAck(); |
| ack.setAckType(MessageAck.STANDARD_ACK_TYPE); |
| ack.setDestination(destination); |
| ack.setMessageID(reference.getMessageId()); |
| try { |
| if (subs instanceof DurableTopicSubscription) { |
| ((DurableTopicSubscription)subs).removePending(reference); |
| } |
| acknowledge(context, subs, ack, reference); |
| } catch (Exception e) { |
| LOG.error("Failed to remove expired Message from the store ", e); |
| } |
| } |
| |
| @Override |
| protected Logger getLog() { |
| return LOG; |
| } |
| |
| protected boolean isOptimizeStorage(){ |
| boolean result = false; |
| |
| if (isDoOptimzeMessageStorage() && durableSubscribers.isEmpty()==false){ |
| result = true; |
| for (DurableTopicSubscription s : durableSubscribers.values()) { |
| if (s.isActive()== false){ |
| result = false; |
| break; |
| } |
| if (s.getPrefetchSize()==0){ |
| result = false; |
| break; |
| } |
| if (s.isSlowConsumer()){ |
| result = false; |
| break; |
| } |
| if (s.getInFlightUsage() > getOptimizeMessageStoreInFlightLimit()){ |
| result = false; |
| break; |
| } |
| } |
| } |
| return result; |
| } |
| |
| /** |
| * force a reread of the store - after transaction recovery completion |
| */ |
| @Override |
| public void clearPendingMessages() { |
| dispatchLock.readLock().lock(); |
| try { |
| for (DurableTopicSubscription durableTopicSubscription : durableSubscribers.values()) { |
| clearPendingAndDispatch(durableTopicSubscription); |
| } |
| } finally { |
| dispatchLock.readLock().unlock(); |
| } |
| } |
| |
| private void clearPendingAndDispatch(DurableTopicSubscription durableTopicSubscription) { |
| synchronized (durableTopicSubscription.pendingLock) { |
| durableTopicSubscription.pending.clear(); |
| try { |
| durableTopicSubscription.dispatchPending(); |
| } catch (IOException exception) { |
| LOG.warn("After clear of pending, failed to dispatch to: {}, for: {}, pending: {}, exception: {}", |
| durableTopicSubscription, |
| destination, |
| durableTopicSubscription.pending, |
| exception); |
| } |
| } |
| } |
| |
| public Map<SubscriptionKey, DurableTopicSubscription> getDurableTopicSubs() { |
| return durableSubscribers; |
| } |
| } |
