import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Inject
public TestLifeCycle(JmsConnection jmsConnection, Destination resultsDestination,
    SuiteExecutor suiteExecutor,
    SuiteIndexWrapper suite)
    throws JMSException {
  this.suiteExecutor = suiteExecutor;
  this.suite = suite;
  if (resultsDestination instanceof ActiveMQDestination) {
    ActiveMQTopic advisoryTopic = AdvisorySupport.getConsumerAdvisoryTopic(resultsDestination);
    session = jmsConnection.getJmsSession();
    topicConsumer = session.createConsumer(advisoryTopic);
    topicConsumer.setMessageListener(this);
  } else {
    topicConsumer = null;
    session = null;
  }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
protected void sendStats(ConnectionContext context, ActiveMQMapMessage msg, ActiveMQDestination replyTo)
        throws Exception {
    msg.setPersistent(false);
    msg.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE);
    msg.setMessageId(new MessageId(this.advisoryProducerId, this.messageIdGenerator.getNextSequenceId()));
    msg.setDestination(replyTo);
    msg.setResponseRequired(false);
    msg.setProducerId(this.advisoryProducerId);
    boolean originalFlowControl = context.isProducerFlowControl();
    final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
    producerExchange.setConnectionContext(context);
    producerExchange.setMutable(true);
    producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
    try {
        context.setProducerFlowControl(false);
        this.next.send(producerExchange, msg);
    } finally {
        context.setProducerFlowControl(originalFlowControl);
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
private void fireFailedForwardAdvisory(MessageDispatch messageDispatch, Throwable error) {
    if (configuration.isAdvisoryForFailedForward()) {
        AdvisoryBroker advisoryBroker = null;
        try {
            advisoryBroker = (AdvisoryBroker) brokerService.getBroker().getAdaptor(AdvisoryBroker.class);

            if (advisoryBroker != null) {
                ConnectionContext context = new ConnectionContext();
                context.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT);
                context.setBroker(brokerService.getBroker());

                ActiveMQMessage advisoryMessage = new ActiveMQMessage();
                advisoryMessage.setStringProperty("cause", error.getLocalizedMessage());
                advisoryBroker.fireAdvisory(context, AdvisorySupport.getNetworkBridgeForwardFailureAdvisoryTopic(), messageDispatch.getMessage(), null,
                    advisoryMessage);

            }
        } catch (Exception e) {
            LOG.warn("failed to fire forward failure advisory, cause: {}", e);
            LOG.debug("detail", e);
        }
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
protected void configureDemandSubscription(ConsumerInfo info, DemandSubscription sub) throws IOException {
    if (AdvisorySupport.isConsumerAdvisoryTopic(info.getDestination())) {
        sub.getLocalInfo().setDispatchAsync(true);
    } else {
        sub.getLocalInfo().setDispatchAsync(configuration.isDispatchAsync());
    }
    sub.getLocalInfo().setPrefetchSize(configuration.getPrefetchSize());
    subscriptionMapByLocalId.put(sub.getLocalInfo().getConsumerId(), sub);
    subscriptionMapByRemoteId.put(sub.getRemoteInfo().getConsumerId(), sub);

    sub.setNetworkBridgeFilter(createNetworkBridgeFilter(info));
    if (!info.isDurable()) {
        // This works for now since we use a VM connection to the local broker.
        // may need to change if we ever subscribe to a remote broker.
        sub.getLocalInfo().setAdditionalPredicate(sub.getNetworkBridgeFilter());
    } else {
        sub.setLocalDurableSubscriber(new SubscriptionInfo(info.getClientId(), info.getSubscriptionName()));
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
/**
 * Add a destination
 *
 * @param context
 * @param destination
 * @throws Exception
 */
@Override
public synchronized void add(ConnectionContext context, Destination destination) throws Exception {
    if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination.getActiveMQDestination())) {
        TopicStorePrefetch tsp = new TopicStorePrefetch(this.subscription,(Topic)destination, clientId, subscriberName);
        tsp.setMaxBatchSize(destination.getMaxPageSize());
        tsp.setSystemUsage(systemUsage);
        tsp.setMessageAudit(getMessageAudit());
        tsp.setEnableAudit(isEnableAudit());
        tsp.setMemoryUsageHighWaterMark(getMemoryUsageHighWaterMark());
        tsp.setUseCache(isUseCache());
        tsp.setCacheEnabled(isUseCache() && tsp.isEmpty());
        topics.put(destination, tsp);
        storePrefetches.add(tsp);
        if (isStarted()) {
            tsp.start();
        }
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
protected FrameTranslator findTranslator(String header, ActiveMQDestination destination) {
    FrameTranslator translator = frameTranslator;
    try {
        if (header != null) {
            translator = (FrameTranslator) FRAME_TRANSLATOR_FINDER
                    .newInstance(header);
        } else {
            if (destination != null && AdvisorySupport.isAdvisoryTopic(destination)) {
                translator = new JmsFrameTranslator();
            }
        }
    } catch (Exception ignore) {
        // if anything goes wrong use the default translator
    }

    if (translator instanceof BrokerContextAware) {
        ((BrokerContextAware)translator).setBrokerContext(brokerContext);
    }

    return translator;
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Test
public void testDynamicallyIncludedDestinationsForDuplex() throws Exception {
   // Once the bridge is set up, we should see the filter used for the duplex end of the bridge
   // only subscribe to the specific destinations included in the <dynamicallyIncludedDestinations> list
   // so let's test that the filter is correct, let's also test the subscription on the localbroker
   // is correct

   // the bridge on the remote broker has the correct filter
   TransportConnection bridgeConnection = getDuplexBridgeConnectionFromRemote();
   assertNotNull(bridgeConnection);
   DemandForwardingBridge duplexBridge = getDuplexBridgeFromConnection(bridgeConnection);
   assertNotNull(duplexBridge);
   NetworkBridgeConfiguration configuration = getConfigurationFromNetworkBridge(duplexBridge);
   assertNotNull(configuration);
   assertFalse("This destinationFilter does not include ONLY the destinations specified in dynamicallyIncludedDestinations", configuration.getDestinationFilter().equals(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX + ">"));
   assertEquals("There are other patterns in the destinationFilter that shouldn't be there", "ActiveMQ.Advisory.Consumer.Queue.include.test.foo,ActiveMQ.Advisory.Consumer.Topic.include.test.bar", configuration.getDestinationFilter());
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public void testConcurrentConnection() throws Exception {

      StubConnection connection1 = createConnection();
      StubConnection connection2 = createConnection();

      // reuse same connection info
      ConnectionInfo connectionInfo = createConnectionInfo();
      connection1.request(connectionInfo);
      connection2.request(connectionInfo);

      // second one should win out, verify using consumer on default session (watchAdvisories)
      ConsumerId consumerId = new ConsumerId(new SessionId(connectionInfo.getConnectionId(), -1), 1);
      ConsumerInfo consumerInfo = new ConsumerInfo(consumerId);
      consumerInfo.setDestination(AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC);

      connection2.request(consumerInfo);
   }
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public void testPrefetchValueOne() throws Exception {

      ActiveMQTopic consumerDestination = new ActiveMQTopic(TOPIC_NAME + "?consumer.prefetchSize=1");
      consumer = session.createConsumer(consumerDestination);

      //add a consumer to the slow consumer advisory topic.
      ActiveMQTopic slowConsumerAdvisoryTopic = AdvisorySupport.getSlowConsumerAdvisoryTopic(destination);
      MessageConsumer slowConsumerAdvisory = session.createConsumer(slowConsumerAdvisoryTopic);

      //publish 2 messages
      Message txtMessage = session.createTextMessage("Sample Text Message");
      for (int i = 0; i < 2; i++) {
         producer.send(txtMessage);
      }

      //consume 2 messages
      for (int i = 0; i < 2; i++) {
         Message receivedMsg = consumer.receive(100);
         Assert.assertNotNull("received msg " + i + " should not be null", receivedMsg);
      }

      //check for "slow consumer" advisory message
      Message slowAdvisoryMessage = slowConsumerAdvisory.receive(100);
      Assert.assertNull("should not have received a slow consumer advisory message", slowAdvisoryMessage);

   }
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Override
public void tempQueueDeleted(SimpleString bindingName) {
   ActiveMQDestination dest = new ActiveMQTempQueue(bindingName.toString());
   state.removeTempDestination(dest);

   if (!AdvisorySupport.isAdvisoryTopic(dest)) {
      AMQConnectionContext context = getContext();
      DestinationInfo advInfo = new DestinationInfo(context.getConnectionId(), DestinationInfo.REMOVE_OPERATION_TYPE, dest);

      ActiveMQTopic topic = AdvisorySupport.getDestinationAdvisoryTopic(dest);
      try {
         protocolManager.fireAdvisory(context, topic, advInfo);
      } catch (Exception e) {
         logger.warn("Failed to fire advisory on " + topic, e);
      }
   }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public static void main(String... args) throws Exception {
	Advisory advisory = new Advisory();
	Session session = advisory.getSession();
   	for (String stock : args) {
   		
   		ActiveMQDestination destination = (ActiveMQDestination)session.createTopic("STOCKS." + stock);
   		
   		Destination consumerTopic = AdvisorySupport.getConsumerAdvisoryTopic(destination);
   		System.out.println("Subscribing to advisory " + consumerTopic);
   		MessageConsumer consumerAdvisory = session.createConsumer(consumerTopic);
   		consumerAdvisory.setMessageListener(new ConsumerAdvisoryListener());
   		
   		Destination noConsumerTopic = AdvisorySupport.getNoTopicConsumersAdvisoryTopic(destination);
   		System.out.println("Subscribing to advisory " + noConsumerTopic);
   		MessageConsumer noConsumerAdvisory = session.createConsumer(noConsumerTopic);
   		noConsumerAdvisory.setMessageListener(new NoConsumerAdvisoryListener());
   		
   	}
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public static void main(String... args) throws Exception {
	Test advisory = new Test();
	Session session = advisory.getSession();
   	for (String job : advisory.jobs) {
   		
   		ActiveMQDestination destination = (ActiveMQDestination)session.createQueue("JOBS." + job);
   		
   		Destination consumerTopic = AdvisorySupport.getConsumerAdvisoryTopic(destination);
   		System.out.println("Subscribing to advisory " + consumerTopic);
   		MessageConsumer consumerAdvisory = session.createConsumer(consumerTopic);
   		consumerAdvisory.setMessageListener(new ConsumerAdvisoryListener());
   		
   		Destination noConsumerTopic = AdvisorySupport.getNoQueueConsumersAdvisoryTopic(destination);
   		System.out.println("Subscribing to advisory " + noConsumerTopic);
   		MessageConsumer noConsumerAdvisory = session.createConsumer(noConsumerTopic);
   		noConsumerAdvisory.setMessageListener(new NoConsumerAdvisoryListener());
   		
   	}
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
protected Set<GroupPrincipal> getACLs(ActiveMQDestination destination, String roleBase, String roleAttribute) {
    try {
        context = open();
    } catch (NamingException e) {
        LOG.error(e.toString());
        return new HashSet<GroupPrincipal>();
    }



    String destinationBase = "";
    SearchControls constraints = new SearchControls();
    if (AdvisorySupport.isAdvisoryTopic(destination) && useAdvisorySearchBase) {
        destinationBase = advisorySearchBase;
    } else {
        if ((destination.getDestinationType() & ActiveMQDestination.QUEUE_TYPE) == ActiveMQDestination.QUEUE_TYPE) {
            destinationBase = queueSearchMatchingFormat.format(new String[]{destination.getPhysicalName()});
            if (queueSearchSubtreeBool) {
                constraints.setSearchScope(SearchControls.SUBTREE_SCOPE);
            } else {
                constraints.setSearchScope(SearchControls.ONELEVEL_SCOPE);
            }
        }
        if ((destination.getDestinationType() & ActiveMQDestination.TOPIC_TYPE) == ActiveMQDestination.TOPIC_TYPE) {
            destinationBase = topicSearchMatchingFormat.format(new String[]{destination.getPhysicalName()});
            if (topicSearchSubtreeBool) {
                constraints.setSearchScope(SearchControls.SUBTREE_SCOPE);
            } else {
                constraints.setSearchScope(SearchControls.ONELEVEL_SCOPE);
            }
        }
    }

    constraints.setReturningAttributes(new String[] {roleAttribute});

    return getACLs(destinationBase, constraints, roleBase, roleAttribute);
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
/**
 * @return the destinationFilter
 */
public String getDestinationFilter() {
    if (this.destinationFilter == null) {
        if (dynamicallyIncludedDestinations != null && !dynamicallyIncludedDestinations.isEmpty()) {
            StringBuffer filter = new StringBuffer();
            String delimiter = "";
            for (ActiveMQDestination destination : dynamicallyIncludedDestinations) {
                if (!destination.isTemporary()) {
                    filter.append(delimiter);
                    filter.append(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX);
                    filter.append(destination.getDestinationTypeAsString());
                    filter.append(".");
                    filter.append(destination.getPhysicalName());
                    delimiter = ",";
                }
            }
            return filter.toString();
        }   else {
            return AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX + ">";
        }
    } else {
        // prepend consumer advisory prefix
        // to keep backward compatibility
        if (!this.destinationFilter.startsWith(AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX)) {
             return AdvisorySupport.CONSUMER_ADVISORY_TOPIC_PREFIX + this.destinationFilter;
        } else {
            return this.destinationFilter;
        }
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Override
public Response processAddProducer(ProducerInfo info) throws Exception {
    SessionId sessionId = info.getProducerId().getParentId();
    ConnectionId connectionId = sessionId.getParentId();
    TransportConnectionState cs = lookupConnectionState(connectionId);
    if (cs == null) {
        throw new IllegalStateException("Cannot add a producer to a connection that had not been registered: "
                + connectionId);
    }
    SessionState ss = cs.getSessionState(sessionId);
    if (ss == null) {
        throw new IllegalStateException("Cannot add a producer to a session that had not been registered: "
                + sessionId);
    }
    // Avoid replaying dup commands
    if (!ss.getProducerIds().contains(info.getProducerId())) {
        ActiveMQDestination destination = info.getDestination();
        if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination)) {
            if (getProducerCount(connectionId) >= connector.getMaximumProducersAllowedPerConnection()){
                throw new IllegalStateException("Can't add producer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumProducersAllowedPerConnection());
            }
        }
        broker.addProducer(cs.getContext(), info);
        try {
            ss.addProducer(info);
        } catch (IllegalStateException e) {
            broker.removeProducer(cs.getContext(), info);
        }

    }
    return null;
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Override
public Response processAddConsumer(ConsumerInfo info) throws Exception {
    SessionId sessionId = info.getConsumerId().getParentId();
    ConnectionId connectionId = sessionId.getParentId();
    TransportConnectionState cs = lookupConnectionState(connectionId);
    if (cs == null) {
        throw new IllegalStateException("Cannot add a consumer to a connection that had not been registered: "
                + connectionId);
    }
    SessionState ss = cs.getSessionState(sessionId);
    if (ss == null) {
        throw new IllegalStateException(broker.getBrokerName()
                + " Cannot add a consumer to a session that had not been registered: " + sessionId);
    }
    // Avoid replaying dup commands
    if (!ss.getConsumerIds().contains(info.getConsumerId())) {
        ActiveMQDestination destination = info.getDestination();
        if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination)) {
            if (getConsumerCount(connectionId) >= connector.getMaximumConsumersAllowedPerConnection()){
                throw new IllegalStateException("Can't add consumer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumConsumersAllowedPerConnection());
            }
        }

        broker.addConsumer(cs.getContext(), info);
        try {
            ss.addConsumer(info);
            addConsumerBrokerExchange(info.getConsumerId());
        } catch (IllegalStateException e) {
            broker.removeConsumer(cs.getContext(), info);
        }

    }
    return null;
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public Topic(BrokerService brokerService, ActiveMQDestination destination, TopicMessageStore store,
        DestinationStatistics parentStats, TaskRunnerFactory taskFactory) throws Exception {
    super(brokerService, store, destination, parentStats);
    this.topicStore = store;
    // set default subscription recovery policy
    if (AdvisorySupport.isMasterBrokerAdvisoryTopic(destination)) {
        subscriptionRecoveryPolicy = new LastImageSubscriptionRecoveryPolicy();
        setAlwaysRetroactive(true);
    } else {
        subscriptionRecoveryPolicy = new NoSubscriptionRecoveryPolicy();
    }
    this.taskRunner = taskFactory.createTaskRunner(this, "Topic  " + destination.getPhysicalName());
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
@Override
public void removeDestination(Destination dest) {
    ActiveMQDestination destination = dest.getActiveMQDestination();
    if (!destination.isTemporary()) {
        if (destination.isQueue()) {
            persistenceAdapter.removeQueueMessageStore((ActiveMQQueue) destination);
        }
        else if (!AdvisorySupport.isAdvisoryTopic(destination)) {
            persistenceAdapter.removeTopicMessageStore((ActiveMQTopic) destination);
        }
    }
}
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
protected void sendScheduledJob(ConnectionContext context, Job job, ActiveMQDestination replyTo) throws Exception {

        org.apache.activemq.util.ByteSequence packet = new org.apache.activemq.util.ByteSequence(job.getPayload());
        try {
            Message msg = (Message) this.wireFormat.unmarshal(packet);
            msg.setOriginalTransactionId(null);
            msg.setPersistent(false);
            msg.setType(AdvisorySupport.ADIVSORY_MESSAGE_TYPE);
            msg.setMessageId(new MessageId(this.producerId, this.messageIdGenerator.getNextSequenceId()));
            msg.setDestination(replyTo);
            msg.setResponseRequired(false);
            msg.setProducerId(this.producerId);

            // Add the jobId as a property
            msg.setProperty("scheduledJobId", job.getJobId());

            final boolean originalFlowControl = context.isProducerFlowControl();
            final ProducerBrokerExchange producerExchange = new ProducerBrokerExchange();
            producerExchange.setConnectionContext(context);
            producerExchange.setMutable(true);
            producerExchange.setProducerState(new ProducerState(new ProducerInfo()));
            try {
                context.setProducerFlowControl(false);
                this.next.send(producerExchange, msg);
            } finally {
                context.setProducerFlowControl(originalFlowControl);
            }
        } catch (Exception e) {
            LOG.error("Failed to send scheduled message {}", job.getJobId(), e);
        }
    }
 

import org.apache.activemq.advisory.AdvisorySupport; //导入依赖的package包/类
public AdvisoryConsumer(ActiveMQConnection connection, ConsumerId consumerId) throws JMSException {
    this.connection = connection;
    info = new ConsumerInfo(consumerId);
    info.setDestination(AdvisorySupport.TEMP_DESTINATION_COMPOSITE_ADVISORY_TOPIC);
    info.setPrefetchSize(1000);
    info.setNoLocal(true);
    info.setDispatchAsync(true);

    this.connection.addDispatcher(info.getConsumerId(), this);
    this.connection.syncSendPacket(this.info);
}