Consumer.java 文件源码

void setConnection(ActiveMQConnection connection) throws OperationsException, JMSException {
    this.session = (ActiveMQSession)connection.createSession(getUseTransaction(), Session.AUTO_ACKNOWLEDGE);

    if(destinationType == KahaDestination.DestinationType.TOPIC) {
        Topic topic = session.createTopic(java.lang.String.valueOf(destinationName));

        if(hasSubscription()) {
            consumer = (ActiveMQMessageConsumer)session.createDurableSubscriber(topic, subscription.getSubscriptionName(), subscription.getMessageSelector(), subscription.getNoLocal());
            showSubscribeAdded();
        }
        else {
            consumer = (ActiveMQMessageConsumer)session.createConsumer(topic);
        }
    }
    else {
        Queue queue = session.createQueue(destinationName);
        consumer = (ActiveMQMessageConsumer)session.createConsumer(queue);
    }
}