import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
/**
 * Add new message consumer.
 *
 * @param context
 * @param info
 * @return
 * @throws Exception
 * @see org.apache.activemq.broker.BrokerFilter#addConsumer(org.apache.activemq.broker.ConnectionContext, org.apache.activemq.command.ConsumerInfo)
 */
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {

    ActiveMQDestination dest = info.getDestination();
    Connection conn = context.getConnection();
    if (dest != null) {
        String destName = info.getDestination().getPhysicalName();
        String clientId = context.getClientId();
        String allowedDest = userMap.get(clientId);

        logger.info(">>> Got Consumer Add request { Destination: " + destName
                + ", Remote Address: " + conn.getRemoteAddress()
                + ", ClientID: " + clientId
                + " }");
        if (allowedDest != null && (allowedDest.equals("*") || allowedDest.equals(destName) || destName.startsWith("ActiveMQ"))) {
            logger.info(">>> Subscription allowed");
        } else {
            logger.error(">>> Destination not allowed. Subscription denied!");
            throw new CmsAuthException(">>> Subscription denied!");
        }
    } else {
        logger.error("<<< Got Consumer Add request from Remote Address:" + conn.getRemoteAddress() + ". But destination is NULL.");
    }
    return super.addConsumer(context, info);
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
private boolean duplicateSuppressionIsRequired(DemandSubscription candidate) {
    final ConsumerInfo consumerInfo = candidate.getRemoteInfo();
    boolean suppress = false;

    if (consumerInfo.getDestination().isQueue() && !configuration.isSuppressDuplicateQueueSubscriptions() || consumerInfo.getDestination().isTopic()
        && !configuration.isSuppressDuplicateTopicSubscriptions()) {
        return suppress;
    }

    List<ConsumerId> candidateConsumers = consumerInfo.getNetworkConsumerIds();
    Collection<Subscription> currentSubs = getRegionSubscriptions(consumerInfo.getDestination());
    for (Subscription sub : currentSubs) {
        List<ConsumerId> networkConsumers = sub.getConsumerInfo().getNetworkConsumerIds();
        if (!networkConsumers.isEmpty()) {
            if (matchFound(candidateConsumers, networkConsumers)) {
                if (isInActiveDurableSub(sub)) {
                    suppress = false;
                } else {
                    suppress = hasLowerPriority(sub, candidate.getLocalInfo());
                }
                break;
            }
        }
    }
    return suppress;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
private boolean hasLowerPriority(Subscription existingSub, ConsumerInfo candidateInfo) {
    boolean suppress = false;

    if (existingSub.getConsumerInfo().getPriority() >= candidateInfo.getPriority()) {
        LOG.debug("{} Ignoring duplicate subscription from {}, sub: {} is duplicate by network subscription with equal or higher network priority: {}, networkConsumerIds: {}", new Object[]{
                configuration.getBrokerName(), remoteBrokerName, candidateInfo, existingSub, existingSub.getConsumerInfo().getNetworkConsumerIds()
        });
        suppress = true;
    } else {
        // remove the existing lower priority duplicate and allow this candidate
        try {
            removeDuplicateSubscription(existingSub);

            LOG.debug("{} Replacing duplicate subscription {} with sub from {}, which has a higher priority, new sub: {}, networkConsumerIds: {}", new Object[]{
                    configuration.getBrokerName(), existingSub.getConsumerInfo(), remoteBrokerName, candidateInfo, candidateInfo.getNetworkConsumerIds()
            });
        } catch (IOException e) {
            LOG.error("Failed to remove duplicated sub as a result of sub with higher priority, sub: {}", existingSub, e);
        }
    }
    return suppress;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public void messageExpired(ConnectionContext context, MessageReference messageReference, Subscription subscription) {
    super.messageExpired(context, messageReference, subscription);
    try {
        if(!messageReference.isAdvisory()) {
            ActiveMQTopic topic = AdvisorySupport.getExpiredMessageTopic(messageReference.getMessage().getDestination());
            Message payload = messageReference.getMessage().copy();
            payload.clearBody();
            ActiveMQMessage advisoryMessage = new ActiveMQMessage();
            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_MESSAGE_ID, payload.getMessageId().toString());
            fireAdvisory(context, topic, payload, null, advisoryMessage);
        }
    } catch (Exception e) {
        handleFireFailure("expired", e);
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public void messageDiscarded(ConnectionContext context, Subscription sub, MessageReference messageReference) {
    super.messageDiscarded(context, sub, messageReference);
    try {
        if (!messageReference.isAdvisory()) {
            ActiveMQTopic topic = AdvisorySupport.getMessageDiscardedAdvisoryTopic(messageReference.getMessage().getDestination());
            Message payload = messageReference.getMessage().copy();
            payload.clearBody();
            ActiveMQMessage advisoryMessage = new ActiveMQMessage();
            if (sub instanceof TopicSubscription) {
                advisoryMessage.setIntProperty(AdvisorySupport.MSG_PROPERTY_DISCARDED_COUNT, ((TopicSubscription)sub).discarded());
            }
            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_ID, sub.getConsumerInfo().getConsumerId().toString());
            fireAdvisory(context, topic, payload, null, advisoryMessage);
        }
    } catch (Exception e) {
        handleFireFailure("discarded", e);
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public boolean sendToDeadLetterQueue(ConnectionContext context, MessageReference messageReference,
                                     Subscription subscription, Throwable poisonCause) {
    boolean wasDLQd = super.sendToDeadLetterQueue(context, messageReference, subscription, poisonCause);
    if (wasDLQd) {
        try {
            if(!messageReference.isAdvisory()) {
                ActiveMQTopic topic = AdvisorySupport.getMessageDLQdAdvisoryTopic(messageReference.getMessage().getDestination());
                Message payload = messageReference.getMessage().copy();
                payload.clearBody();
                fireAdvisory(context, topic,payload);
            }
        } catch (Exception e) {
            handleFireFailure("add to DLQ", e);
        }
    }

    return wasDLQd;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public ObjectName createDurableSubscriber(String clientId, String subscriberName, String topicName,
                                          String selector) throws Exception {
    ConnectionContext context = new ConnectionContext();
    context.setBroker(safeGetBroker());
    context.setClientId(clientId);
    ConsumerInfo info = new ConsumerInfo();
    ConsumerId consumerId = new ConsumerId();
    consumerId.setConnectionId(clientId);
    consumerId.setSessionId(sessionIdCounter.incrementAndGet());
    consumerId.setValue(0);
    info.setConsumerId(consumerId);
    info.setDestination(new ActiveMQTopic(topicName));
    info.setSubscriptionName(subscriberName);
    info.setSelector(selector);
    Subscription subscription = safeGetBroker().addConsumer(context, info);
    safeGetBroker().removeConsumer(context, info);
    if (subscription != null) {
        return subscription.getObjectName();
    }
    return null;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Test
public void testRemoveLowerPriorityDup() throws Exception {

   List<Subscription> consumers = new ArrayList<>();

   for (int i = 0; i < 3; i++) {
      ConsumerInfo instance = info.copy();
      instance.setPriority((byte) i);
      consumers.add(new TopicSubscription(brokerService.getBroker(), context, instance, usageManager));
   }
   underTest.dispatch(node, null, consumers);

   long count = 0;
   for (Subscription consumer : consumers) {
      count += consumer.getEnqueueCounter();
   }
   assertEquals("only one sub got message", 1, count);
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
public boolean dispatch(MessageReference node, MessageEvaluationContext msgContext, List<Subscription> consumers)
        throws Exception {

    int count = 0;
    for (Subscription sub : consumers) {
        // Don't deliver to browsers
        if (sub.getConsumerInfo().isBrowser()) {
            continue;
        }
        // Only dispatch to interested subscriptions
        if (!sub.matches(node, msgContext)) {
            sub.unmatched(node);
            continue;
        }

        sub.add(node);
        count++;
    }

    return count > 0;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
/**
 * @param node
 * @param msgContext
 * @param consumers
 * @return true if dispatched
 * @throws Exception
 * @see org.apache.activemq.broker.region.policy.DispatchPolicy#dispatch(org.apache.activemq.broker.region.MessageReference,
 *      org.apache.activemq.filter.MessageEvaluationContext, java.util.List)
 */
public boolean dispatch(MessageReference node, MessageEvaluationContext msgContext, List consumers) throws Exception {
    // Big synch here so that only 1 message gets dispatched at a time.
    // Ensures
    // Everyone sees the same order.
    synchronized (consumers) {
        int count = 0;
        for (Iterator iter = consumers.iterator(); iter.hasNext();) {
            Subscription sub = (Subscription)iter.next();

            // Only dispatch to interested subscriptions
            if (!sub.matches(node, msgContext)) {
                sub.unmatched(node);
                continue;
            }

            sub.add(node);
            count++;
        }
        return count > 0;
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
public void configurePrefetch(Subscription subscription) {

        final int currentPrefetch = subscription.getConsumerInfo().getPrefetchSize();
        if (subscription instanceof QueueBrowserSubscription) {
            if (currentPrefetch == ActiveMQPrefetchPolicy.DEFAULT_QUEUE_BROWSER_PREFETCH) {
                ((QueueBrowserSubscription) subscription).setPrefetchSize(getQueueBrowserPrefetch());
            }
        } else if (subscription instanceof QueueSubscription) {
            if (currentPrefetch == ActiveMQPrefetchPolicy.DEFAULT_QUEUE_PREFETCH) {
                ((QueueSubscription) subscription).setPrefetchSize(getQueuePrefetch());
            }
        } else if (subscription instanceof DurableTopicSubscription) {
            if (currentPrefetch == ActiveMQPrefetchPolicy.DEFAULT_DURABLE_TOPIC_PREFETCH ||
                    subscription.getConsumerInfo().getPrefetchSize() == ActiveMQPrefetchPolicy.DEFAULT_OPTIMIZE_DURABLE_TOPIC_PREFETCH) {
                ((DurableTopicSubscription)subscription).setPrefetchSize(getDurableTopicPrefetch());
            }
        } else if (subscription instanceof TopicSubscription) {
            if (currentPrefetch == ActiveMQPrefetchPolicy.DEFAULT_TOPIC_PREFETCH) {
                ((TopicSubscription) subscription).setPrefetchSize(getTopicPrefetch());
            }
        }
        if (currentPrefetch != 0 && subscription.getPrefetchSize() == 0) {
            // tell the sub so that it can issue a pull request
            subscription.updateConsumerPrefetch(0);
        }
    }
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public void slowConsumer(ConnectionContext context, Subscription subs) {
    if (maxSlowCount < 0 && maxSlowDuration < 0) {
        // nothing to do
        LOG.info("no limits set, slowConsumer strategy has nothing to do");
        return;
    }

    if (taskStarted.compareAndSet(false, true)) {
        scheduler.executePeriodically(this, checkPeriod);
    }

    if (!slowConsumers.containsKey(subs)) {
        slowConsumers.put(subs, new SlowConsumerEntry(context));
    } else if (maxSlowCount > 0) {
        slowConsumers.get(subs).slow();
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
/**
 * @throws Exception If failed.
 */
public void testQueueMultipleThreads() throws Exception {
    Destination dest = new ActiveMQQueue(QUEUE_NAME);

    // produce messages into the queue
    produceObjectMessages(dest, false);

    try (IgniteDataStreamer<String, String> dataStreamer = grid().dataStreamer(DEFAULT_CACHE_NAME)) {
        JmsStreamer<ObjectMessage, String, String> jmsStreamer = newJmsStreamer(ObjectMessage.class, dataStreamer);
        jmsStreamer.setDestination(dest);
        jmsStreamer.setThreads(5);

        // subscribe to cache PUT events and return a countdown latch starting at CACHE_ENTRY_COUNT
        CountDownLatch latch = subscribeToPutEvents(CACHE_ENTRY_COUNT);

        // start the streamer
        jmsStreamer.start();

        DestinationStatistics qStats = broker.getBroker().getDestinationMap().get(dest).getDestinationStatistics();
        assertEquals(5, qStats.getConsumers().getCount());

        // all cache PUT events received in 10 seconds
        latch.await(10, TimeUnit.SECONDS);

        // assert that all consumers received messages - given that the prefetch is 1
        for (Subscription subscription : broker.getBroker().getDestinationMap().get(dest).getConsumers())
            assertTrue(subscription.getDequeueCounter() > 0);

        assertAllCacheEntriesLoaded();

        jmsStreamer.stop();
    }

}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
    final SecurityContext securityContext = checkSecurityContext(context);

    Set<?> allowedACLs = null;
    if (!info.getDestination().isTemporary()) {
        allowedACLs = authorizationMap.getReadACLs(info.getDestination());
    } else {
        allowedACLs = authorizationMap.getTempDestinationReadACLs();
    }

    if (!securityContext.isBrokerContext() && allowedACLs != null && !securityContext.isInOneOf(allowedACLs) ) {
        throw new SecurityException("User " + securityContext.getUserName() + " is not authorized to read from: " + info.getDestination());
    }
    securityContext.getAuthorizedReadDests().put(info.getDestination(), info.getDestination());

    /*
     * Need to think about this a little more. We could do per message
     * security checking to implement finer grained security checking. For
     * example a user can only see messages with price>1000 . Perhaps this
     * should just be another additional broker filter that installs this
     * type of feature. If we did want to do that, then we would install a
     * predicate. We should be careful since there may be an existing
     * predicate already assigned and the consumer info may be sent to a
     * remote broker, so it also needs to support being marshaled.
     * info.setAdditionalPredicate(new BooleanExpression() { public boolean
     * matches(MessageEvaluationContext message) throws JMSException { if(
     * !subject.getAuthorizedReadDests().contains(message.getDestination()) ) {
     * Set allowedACLs =
     * authorizationMap.getReadACLs(message.getDestination());
     * if(allowedACLs!=null && !subject.isInOneOf(allowedACLs)) return
     * false; subject.getAuthorizedReadDests().put(message.getDestination(),
     * message.getDestination()); } return true; } public Object
     * evaluate(MessageEvaluationContext message) throws JMSException {
     * return matches(message) ? Boolean.TRUE : Boolean.FALSE; } });
     */

    return super.addConsumer(context, info);
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public boolean sendToDeadLetterQueue(ConnectionContext ctx, MessageReference msgRef, Subscription subscription, Throwable poisonCause) {
    log.trace("Discarding DLQ BrokerFilter[pass through] - skipping message: {}", (msgRef != null ? msgRef.getMessage() : null));
    boolean dropped = true;
    Message msg = null;
    ActiveMQDestination dest = null;
    String destName = null;
    msg = msgRef.getMessage();
    dest = msg.getDestination();
    destName = dest.getPhysicalName();

    if (dest == null || destName == null) {
        // do nothing, no need to forward it
        skipMessage("NULL DESTINATION", msgRef);
    } else if (dropAll) {
        // do nothing
        skipMessage("dropAll", msgRef);
    } else if (dropTemporaryTopics && dest.isTemporary() && dest.isTopic()) {
        // do nothing
        skipMessage("dropTemporaryTopics", msgRef);
    } else if (dropTemporaryQueues && dest.isTemporary() && dest.isQueue()) {
        // do nothing
        skipMessage("dropTemporaryQueues", msgRef);
    } else if (destFilter != null && matches(destName)) {
        // do nothing
        skipMessage("dropOnly", msgRef);
    } else {
        dropped = false;
        return next.sendToDeadLetterQueue(ctx, msgRef, subscription, poisonCause);
    }

    if (dropped && getReportInterval() > 0) {
        if ((++dropCount) % getReportInterval() == 0) {
            log.info("Total of {} messages were discarded, since their destination was the dead letter queue", dropCount);
        }
    }

    return false;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public Subscription addConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
    LOG.debug("Caching consumer selector [{}] on a {}", info.getSelector(), info.getDestination().getQualifiedName());
    String selector = info.getSelector();

    // As ConcurrentHashMap doesn't support null values, use always true expression
    if (selector == null) {
        selector = "TRUE";
    }

    subSelectorCache.put(info.getDestination().getQualifiedName(), selector);

    return super.addConsumer(context, info);
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
private boolean hasNoLocalConsumers(final Message message, final MessageEvaluationContext mec) {
    Destination regionDestination = (Destination) mec.getMessageReference().getRegionDestination();
    List<Subscription> consumers = regionDestination.getConsumers();
    for (Subscription sub : consumers) {
        if (!sub.getConsumerInfo().isNetworkSubscription() && !sub.getConsumerInfo().isBrowser()) {
            LOG.trace("Not replaying [{}] for [{}] to origin due to existing local consumer: {}", new Object[]{
                    message.getMessageId(), message.getDestination(), sub.getConsumerInfo()
            });
            return false;
        }
    }
    return true;
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
private void removeDuplicateSubscription(Subscription existingSub) throws IOException {
    for (NetworkConnector connector : brokerService.getNetworkConnectors()) {
        if (connector.removeDemandSubscription(existingSub.getConsumerInfo().getConsumerId())) {
            break;
        }
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public void slowConsumer(ConnectionContext context, Destination destination,Subscription subs) {
    super.slowConsumer(context, destination,subs);
    try {
        if (!AdvisorySupport.isAdvisoryTopic(destination.getActiveMQDestination())) {
            ActiveMQTopic topic = AdvisorySupport.getSlowConsumerAdvisoryTopic(destination.getActiveMQDestination());
            ActiveMQMessage advisoryMessage = new ActiveMQMessage();
            advisoryMessage.setStringProperty(AdvisorySupport.MSG_PROPERTY_CONSUMER_ID, subs.getConsumerInfo().getConsumerId().toString());
            fireAdvisory(context, topic, subs.getConsumerInfo(), null, advisoryMessage);
        }
    } catch (Exception e) {
        handleFireFailure("slow consumer", e);
    }
}
 

import org.apache.activemq.broker.region.Subscription; //导入依赖的package包/类
@Override
public boolean sendToDeadLetterQueue(ConnectionContext context,
                                     MessageReference messageReference,
                                     Subscription subscription,
                                     Throwable poisonCause) {
   throw new RuntimeException("Don't call me!");
}