import org.apache.activemq.command.ConsumerInfo; //导入依赖的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.command.ConsumerInfo; //导入依赖的package包/类
@Test(priority=5)
public void addConsumerTest(){
	//set up a mock for ConsumerInfo
	ActiveMQDestination activeMQDestinationMQ =  ActiveMQDestination.createDestination("mockMQDestionation",  (byte) 1 );		
	activeMQDestinationMQ.setPhysicalName(MQ_PHYSICAL_NAME);
	
	ConsumerInfo consumerInfoActiveMQ = mock(ConsumerInfo.class);
	when(consumerInfoActiveMQ.getDestination()).thenReturn(activeMQDestinationMQ);
	
	//set up mock for ProducerInfo
	producerInfo = mock(ProducerInfo.class);
	when(producerInfo.getDestination()).thenReturn(activeMQDestination);	
	
	try {
		this.oneopsAuthBroker.addConsumer(connectionContextMock, consumerInfoActiveMQ);
	} catch (Exception e) {
		logger.warn("caught exception, make sure Broker is mocked",e);
		throw new RuntimeException(e);
	}
	
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
/**
 * @param transport
 * @param sessionState
 * @throws IOException
 */
protected void restoreConsumers(Transport transport, SessionState sessionState) throws IOException {
    // Restore the session's consumers but possibly in pull only (prefetch 0 state) till recovery complete
    final ConnectionState connectionState = connectionStates.get(sessionState.getInfo().getSessionId().getParentId());
    final boolean connectionInterruptionProcessingComplete = connectionState.isConnectionInterruptProcessingComplete();
    for (ConsumerState consumerState : sessionState.getConsumerStates()) {   
        ConsumerInfo infoToSend = consumerState.getInfo();
        if (!connectionInterruptionProcessingComplete && infoToSend.getPrefetchSize() > 0) {
            infoToSend = consumerState.getInfo().copy();
            connectionState.getRecoveringPullConsumers().put(infoToSend.getConsumerId(), consumerState.getInfo());
            infoToSend.setPrefetchSize(0);
            if (LOG.isDebugEnabled()) {
                LOG.debug("restore consumer: " + infoToSend.getConsumerId() + " in pull mode pending recovery, overriding prefetch: " + consumerState.getInfo().getPrefetchSize());
            }
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("restore consumer: " + infoToSend.getConsumerId());
        }
        transport.oneway(infoToSend);
    }
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
public Response processAddConsumer(ConsumerInfo info) {
    if (info != null) {
        SessionId sessionId = info.getConsumerId().getParentId();
        if (sessionId != null) {
            ConnectionId connectionId = sessionId.getParentId();
            if (connectionId != null) {
                ConnectionState cs = connectionStates.get(connectionId);
                if (cs != null) {
                    SessionState ss = cs.getSessionState(sessionId);
                    if (ss != null) {
                        ss.addConsumer(info);
                    }
                }
            }
        }
    }
    return TRACKED_RESPONSE_MARKER;
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
QoS onSubscribe(SUBSCRIBE command, Topic topic) throws MQTTProtocolException {
    ActiveMQDestination destination = new ActiveMQTopic(convertMQTTToActiveMQ(topic.name().toString()));

    ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId());
    ConsumerInfo consumerInfo = new ConsumerInfo(id);
    consumerInfo.setDestination(destination);
    consumerInfo.setPrefetchSize(getActiveMQSubscriptionPrefetch());
    consumerInfo.setDispatchAsync(true);
    if (!connect.cleanSession() && (connect.clientId() != null)) {
        //by default subscribers are persistent
        consumerInfo.setSubscriptionName(
            connect.clientId().toString() + topic.name().toString());
    }
    MQTTSubscription mqttSubscription = new MQTTSubscription(this, topic.qos(), consumerInfo);

    subscriptionsByConsumerId.put(id, mqttSubscription);
    mqttSubscriptionByTopic.put(topic.name(), mqttSubscription);

    sendToActiveMQ(consumerInfo, null);
    return topic.qos();
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
public void testAddConsumerThenSend() throws Exception {

      // Start a producer and consumer
      StubConnection connection = createConnection();
      ConnectionInfo connectionInfo = createConnectionInfo();
      SessionInfo sessionInfo = createSessionInfo(connectionInfo);
      ProducerInfo producerInfo = createProducerInfo(sessionInfo);
      connection.send(connectionInfo);
      connection.send(sessionInfo);
      connection.send(producerInfo);

      destination = createDestinationInfo(connection, connectionInfo, destinationType);

      ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
      connection.send(consumerInfo);

      connection.send(createMessage(producerInfo, destination, deliveryMode));

      // Make sure the message was delivered.
      Message m = receiveMessage(connection);
      assertNotNull(m);
   }
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
/**
 * Write the booleans that this object uses to a BooleanStream
 */
public void looseMarshal(OpenWireFormat wireFormat, Object o, DataOutput dataOut) throws IOException {

    ConsumerInfo info = (ConsumerInfo)o;

    super.looseMarshal(wireFormat, o, dataOut);
    looseMarshalCachedObject(wireFormat, (DataStructure)info.getConsumerId(), dataOut);
    dataOut.writeBoolean(info.isBrowser());
    looseMarshalCachedObject(wireFormat, (DataStructure)info.getDestination(), dataOut);
    dataOut.writeInt(info.getPrefetchSize());
    dataOut.writeInt(info.getMaximumPendingMessageLimit());
    dataOut.writeBoolean(info.isDispatchAsync());
    looseMarshalString(info.getSelector(), dataOut);
    looseMarshalString(info.getSubscriptionName(), dataOut);
    dataOut.writeBoolean(info.isNoLocal());
    dataOut.writeBoolean(info.isExclusive());
    dataOut.writeBoolean(info.isRetroactive());
    dataOut.writeByte(info.getPriority());
    looseMarshalObjectArray(wireFormat, info.getBrokerPath(), dataOut);
    looseMarshalNestedObject(wireFormat, (DataStructure)info.getAdditionalPredicate(), dataOut);
    dataOut.writeBoolean(info.isNetworkSubscription());
    dataOut.writeBoolean(info.isOptimizedAcknowledge());
    dataOut.writeBoolean(info.isNoRangeAcks());

}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
private void consumeExpected() throws Exception {
   // Setup the consumer and receive the message.
   StubConnection connection = createConnection();
   ConnectionInfo connectionInfo = createConnectionInfo();
   SessionInfo sessionInfo = createSessionInfo(connectionInfo);
   connection.send(connectionInfo);
   connection.send(sessionInfo);
   ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
   connection.send(consumerInfo);

   Message m = receiveMessage(connection);
   assertNotNull("Should have received message " + expected.get(0) + " by now!", m);
   assertEquals(expected.get(0), m.getMessageId().toString());
   MessageAck ack = createAck(consumerInfo, m, 1, MessageAck.STANDARD_ACK_TYPE);
   connection.send(ack);

   assertNoMessagesLeft(connection);
   connection.request(closeConnectionInfo(connectionInfo));
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的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.command.ConsumerInfo; //导入依赖的package包/类
/**
 * Write the booleans that this object uses to a BooleanStream
 */
public int tightMarshal1(OpenWireFormat wireFormat, Object o, BooleanStream bs) throws IOException {

    ConsumerInfo info = (ConsumerInfo)o;

    int rc = super.tightMarshal1(wireFormat, o, bs);
    rc += tightMarshalCachedObject1(wireFormat, (DataStructure)info.getConsumerId(), bs);
    bs.writeBoolean(info.isBrowser());
    rc += tightMarshalCachedObject1(wireFormat, (DataStructure)info.getDestination(), bs);
    bs.writeBoolean(info.isDispatchAsync());
    rc += tightMarshalString1(info.getSelector(), bs);
    rc += tightMarshalString1(info.getSubscriptionName(), bs);
    bs.writeBoolean(info.isNoLocal());
    bs.writeBoolean(info.isExclusive());
    bs.writeBoolean(info.isRetroactive());
    rc += tightMarshalObjectArray1(wireFormat, info.getBrokerPath(), bs);
    rc += tightMarshalNestedObject1(wireFormat, (DataStructure)info.getAdditionalPredicate(), bs);
    bs.writeBoolean(info.isNetworkSubscription());
    bs.writeBoolean(info.isOptimizedAcknowledge());
    bs.writeBoolean(info.isNoRangeAcks());

    return rc + 9;
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
private static BooleanExpression parseSelector(ConsumerInfo info) throws InvalidSelectorException {
    BooleanExpression rc = null;
    if (info.getSelector() != null) {
        rc = SelectorParser.parse(info.getSelector());
    }
    if (info.isNoLocal()) {
        if (rc == null) {
            rc = new NoLocalExpression(info.getConsumerId().getConnectionId());
        } else {
            rc = LogicExpression.createAND(new NoLocalExpression(info.getConsumerId().getConnectionId()), rc);
        }
    }
    if (info.getAdditionalPredicate() != null) {
        if (rc == null) {
            rc = info.getAdditionalPredicate();
        } else {
            rc = LogicExpression.createAND(info.getAdditionalPredicate(), rc);
        }
    }
    return rc;
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
public void testQueueBrowseMessages() throws Exception {

      // Start a producer and consumer
      StubConnection connection = createConnection();
      ConnectionInfo connectionInfo = createConnectionInfo();
      SessionInfo sessionInfo = createSessionInfo(connectionInfo);
      ProducerInfo producerInfo = createProducerInfo(sessionInfo);
      connection.send(connectionInfo);
      connection.send(sessionInfo);
      connection.send(producerInfo);

      destination = createDestinationInfo(connection, connectionInfo, destinationType);

      connection.send(createMessage(producerInfo, destination, deliveryMode));
      connection.send(createMessage(producerInfo, destination, deliveryMode));
      connection.send(createMessage(producerInfo, destination, deliveryMode));
      connection.send(createMessage(producerInfo, destination, deliveryMode));

      // Use selector to skip first message.
      ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
      consumerInfo.setBrowser(true);
      connection.send(consumerInfo);

      for (int i = 0; i < 4; i++) {
         Message m = receiveMessage(connection);
         assertNotNull(m);
         connection.send(createAck(consumerInfo, m, 1, MessageAck.DELIVERED_ACK_TYPE));
      }

      assertNoMessagesLeft(connection);
   }
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的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.command.ConsumerInfo; //导入依赖的package包/类
@Override
public NetworkBridgeFilter create(ConsumerInfo info, BrokerId[] remoteBrokerPath, int messageTTL, int consumerTTL) {
    ConditionalNetworkBridgeFilter filter = new ConditionalNetworkBridgeFilter();
    filter.setNetworkBrokerId(remoteBrokerPath[0]);
    filter.setMessageTTL(messageTTL);
    filter.setConsumerTTL(consumerTTL);
    filter.setAllowReplayWhenNoConsumers(isReplayWhenNoConsumers());
    filter.setRateLimit(getRateLimit());
    filter.setRateDuration(getRateDuration());
    filter.setReplayDelay(getReplayDelay());
    return filter;
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
/**
 * @throws IOException
 */
final void startBridge() throws IOException {
    connectionInfo = new ConnectionInfo();
    connectionInfo.setConnectionId(new ConnectionId(ID_GENERATOR.generateId()));
    connectionInfo.setClientId(clientId);
    localBroker.oneway(connectionInfo);
    remoteBroker.oneway(connectionInfo);

    sessionInfo = new SessionInfo(connectionInfo, 1);
    localBroker.oneway(sessionInfo);
    remoteBroker.oneway(sessionInfo);

    queueConsumerInfo = new ConsumerInfo(sessionInfo, 1);
    queueConsumerInfo.setDispatchAsync(dispatchAsync);
    queueConsumerInfo.setDestination(new ActiveMQQueue(destinationFilter));
    queueConsumerInfo.setPrefetchSize(prefetchSize);
    queueConsumerInfo.setPriority(ConsumerInfo.NETWORK_CONSUMER_PRIORITY);
    localBroker.oneway(queueConsumerInfo);

    producerInfo = new ProducerInfo(sessionInfo, 1);
    producerInfo.setResponseRequired(false);
    remoteBroker.oneway(producerInfo);

    if (connectionInfo.getClientId() != null) {
        topicConsumerInfo = new ConsumerInfo(sessionInfo, 2);
        topicConsumerInfo.setDispatchAsync(dispatchAsync);
        topicConsumerInfo.setSubscriptionName("topic-bridge");
        topicConsumerInfo.setRetroactive(true);
        topicConsumerInfo.setDestination(new ActiveMQTopic(destinationFilter));
        topicConsumerInfo.setPrefetchSize(prefetchSize);
        topicConsumerInfo.setPriority(ConsumerInfo.NETWORK_CONSUMER_PRIORITY);
        localBroker.oneway(topicConsumerInfo);
    }

    LOG.info("Network connection between {} and {} has been established.", localBroker, remoteBroker);
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
public void testConsumerPrefetchAndDeliveredAck() throws Exception {

      // Start a producer and consumer
      StubConnection connection = createConnection();
      ConnectionInfo connectionInfo = createConnectionInfo();
      SessionInfo sessionInfo = createSessionInfo(connectionInfo);
      ProducerInfo producerInfo = createProducerInfo(sessionInfo);
      connection.send(connectionInfo);
      connection.send(sessionInfo);
      connection.send(producerInfo);

      destination = createDestinationInfo(connection, connectionInfo, destinationType);

      ConsumerInfo consumerInfo = createConsumerInfo(sessionInfo, destination);
      consumerInfo.setPrefetchSize(1);
      connection.request(consumerInfo);

      // Send 3 messages to the broker.
      connection.send(createMessage(producerInfo, destination, deliveryMode));
      connection.send(createMessage(producerInfo, destination, deliveryMode));
      connection.request(createMessage(producerInfo, destination, deliveryMode));

      // Make sure only 1 message was delivered.
      Message m1 = receiveMessage(connection);
      assertNotNull(m1);

      assertNoMessagesLeft(connection);

      // Acknowledge the first message. This should cause the next message to
      // get dispatched.
      connection.request(createAck(consumerInfo, m1, 1, MessageAck.DELIVERED_ACK_TYPE));

      Message m2 = receiveMessage(connection);
      assertNotNull(m2);
      connection.request(createAck(consumerInfo, m2, 1, MessageAck.DELIVERED_ACK_TYPE));

      Message m3 = receiveMessage(connection);
      assertNotNull(m3);
      connection.request(createAck(consumerInfo, m3, 1, MessageAck.DELIVERED_ACK_TYPE));
   }
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的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);
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
public void testRequestResponse() throws Exception {
   ConsumerInfo expected = new ConsumerInfo();
   expected.setSelector("Edam");
   expected.setResponseRequired(true);
   LOG.info("About to send: " + expected);
   Response response = (Response) producer.request(expected, 2000);

   LOG.info("Received: " + response);
   assertNotNull("Received a response", response);
   assertTrue("Should not be an exception", !response.isException());
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
/**
 * Create a ConnectionConsumer
 * 
 * @param theConnection
 * @param theSessionPool
 * @param theConsumerInfo
 * @throws JMSException
 */
protected ActiveMQConnectionConsumer(ActiveMQConnection theConnection, ServerSessionPool theSessionPool, ConsumerInfo theConsumerInfo) throws JMSException {
    this.connection = theConnection;
    this.sessionPool = theSessionPool;
    this.consumerInfo = theConsumerInfo;

    this.connection.addConnectionConsumer(this);
    this.connection.addDispatcher(consumerInfo.getConsumerId(), this);
    this.connection.syncSendPacket(this.consumerInfo);
}
 

import org.apache.activemq.command.ConsumerInfo; //导入依赖的package包/类
@Override
public void removeConsumer(ConnectionContext context, ConsumerInfo info) throws Exception {
    super.removeConsumer(context, info);

    // Don't advise advisory topics.
    ActiveMQDestination dest = info.getDestination();
    if (!AdvisorySupport.isAdvisoryTopic(dest)) {
        ActiveMQTopic topic = AdvisorySupport.getConsumerAdvisoryTopic(dest);
        consumers.remove(info);
        if (!dest.isTemporary() || destinations.containsKey(dest)) {
            fireConsumerAdvisory(context,dest, topic, info.createRemoveCommand());
        }
    }
}