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

   RedeliveryPolicy policy = new RedeliveryPolicy();
   policy.setInitialRedeliveryDelay(0);
   policy.setRedeliveryDelay(500);
   policy.setBackOffMultiplier((short) 2);
   policy.setUseExponentialBackOff(true);

   long delay = policy.getNextRedeliveryDelay(0);
   assertEquals(500, delay);
   delay = policy.getNextRedeliveryDelay(delay);
   assertEquals(500 * 2, delay);
   delay = policy.getNextRedeliveryDelay(delay);
   assertEquals(500 * 4, delay);

   policy.setUseExponentialBackOff(false);
   delay = policy.getNextRedeliveryDelay(delay);
   assertEquals(500, delay);
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
private void startBroker(boolean deleteMessages) throws Exception {
   broker = new BrokerService();
   broker.setSchedulerSupport(true);

   RedeliveryPlugin redeliveryPlugin = new RedeliveryPlugin();

   RedeliveryPolicy brokerRedeliveryPolicy = new RedeliveryPolicy();
   brokerRedeliveryPolicy.setRedeliveryDelay(redeliveryDelayMillis);
   brokerRedeliveryPolicy.setInitialRedeliveryDelay(initialRedeliveryDelayMillis);
   brokerRedeliveryPolicy.setMaximumRedeliveries(maxBrokerRedeliveries);

   RedeliveryPolicyMap redeliveryPolicyMap = new RedeliveryPolicyMap();
   redeliveryPolicyMap.setDefaultEntry(brokerRedeliveryPolicy);
   redeliveryPlugin.setRedeliveryPolicyMap(redeliveryPolicyMap);

   broker.setPlugins(new BrokerPlugin[]{redeliveryPlugin});

   if (deleteMessages) {
      broker.setDeleteAllMessagesOnStartup(true);
   }
   broker.start();
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Before
public void startBroker() throws Exception {
   broker = new BrokerService();
   broker.setDeleteAllMessagesOnStartup(true);
   broker.setPersistent(false);
   broker.setUseJmx(false);
   broker.addConnector("tcp://0.0.0.0:0");
   broker.start();
   broker.waitUntilStarted();

   connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
   connectionFactory = new ActiveMQConnectionFactory(connectionUri);
   connectionFactory.setNonBlockingRedelivery(true);

   RedeliveryPolicy policy = connectionFactory.getRedeliveryPolicy();
   policy.setInitialRedeliveryDelay(TimeUnit.SECONDS.toMillis(2));
   policy.setBackOffMultiplier(-1);
   policy.setRedeliveryDelay(TimeUnit.SECONDS.toMillis(2));
   policy.setMaximumRedeliveryDelay(-1);
   policy.setUseExponentialBackOff(false);
   policy.setMaximumRedeliveries(-1);
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
public RedeliveryPolicy getEntryFor(ActiveMQDestination destination) {
    RedeliveryPolicy answer = (RedeliveryPolicy) chooseValue(destination);
    if (answer == null) {
        answer = getDefaultEntry();
    }
    return answer;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
private void validatePolicyDelay(long limit) {
    final ActiveMQDestination matchAll = new AnyDestination(new ActiveMQDestination[]{new ActiveMQQueue(">"), new ActiveMQTopic(">")});
    for (Object entry : redeliveryPolicyMap.get(matchAll)) {
        RedeliveryPolicy redeliveryPolicy = (RedeliveryPolicy) entry;
        validateLimit(limit, redeliveryPolicy);
    }
    RedeliveryPolicy defaultEntry = redeliveryPolicyMap.getDefaultEntry();
    if (defaultEntry != null) {
        validateLimit(limit, defaultEntry);
    }
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
private void validateLimit(long limit, RedeliveryPolicy redeliveryPolicy) {
    if (redeliveryPolicy.getInitialRedeliveryDelay() < limit) {
        throw new IllegalStateException("RedeliveryPolicy initialRedeliveryDelay must exceed: " + limit + ". " + redeliveryPolicy);
    }
    if (redeliveryPolicy.getRedeliveryDelay() < limit) {
        throw new IllegalStateException("RedeliveryPolicy redeliveryDelay must exceed: " + limit + ". " + redeliveryPolicy);
    }
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
public boolean sendToDeadLetterQueue(ConnectionContext context, MessageReference messageReference, Subscription subscription, Throwable poisonCause) {
    if (messageReference.isExpired()) {
        // there are two uses of  sendToDeadLetterQueue, we are only interested in valid messages
        return super.sendToDeadLetterQueue(context, messageReference, subscription, poisonCause);
    } else {
        try {
            Destination regionDestination = (Destination) messageReference.getRegionDestination();
            final RedeliveryPolicy redeliveryPolicy = redeliveryPolicyMap.getEntryFor(regionDestination.getActiveMQDestination());
            if (redeliveryPolicy != null) {
                final int maximumRedeliveries = redeliveryPolicy.getMaximumRedeliveries();
                int redeliveryCount = messageReference.getRedeliveryCounter();
                if (RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES == maximumRedeliveries || redeliveryCount < maximumRedeliveries) {

                    long delay = ( redeliveryCount == 0 ?
                            redeliveryPolicy.getInitialRedeliveryDelay() :
                            redeliveryPolicy.getNextRedeliveryDelay(getExistingDelay(messageReference)));

                    scheduleRedelivery(context, messageReference, delay, ++redeliveryCount);
                } else if (isSendToDlqIfMaxRetriesExceeded()) {
                    return super.sendToDeadLetterQueue(context, messageReference, subscription, poisonCause);
                } else {
                    LOG.debug("Discarding message that exceeds max redelivery count({}), {}", maximumRedeliveries, messageReference.getMessageId());
                }
            } else if (isFallbackToDeadLetter()) {
                return super.sendToDeadLetterQueue(context, messageReference, subscription, poisonCause);
            } else {
                LOG.debug("Ignoring dlq request for: {}, RedeliveryPolicy not found (and no fallback) for: {}", messageReference.getMessageId(), regionDestination.getActiveMQDestination());
            }

            return false;
        } catch (Exception exception) {
            // abort the ack, will be effective if client use transactions or individual ack with sync send
            RuntimeException toThrow =  new RuntimeException("Failed to schedule redelivery for: " + messageReference.getMessageId(), exception);
            LOG.error(toThrow.toString(), exception);
            throw toThrow;
        }
    }
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
/**
 * @throws Exception
 */
@Test
public void testZeroMaximumNumberOfRedeliveries() throws Exception {

   // Receive a message with the JMS API
   RedeliveryPolicy policy = connection.getRedeliveryPolicy();
   policy.setInitialRedeliveryDelay(100);
   policy.setUseExponentialBackOff(false);
   // let's set the maximum redeliveries to 0
   policy.setMaximumRedeliveries(0);

   connection.start();
   Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
   ActiveMQQueue destination = new ActiveMQQueue("TEST");
   this.makeSureCoreQueueExist("TEST");
   MessageProducer producer = session.createProducer(destination);

   MessageConsumer consumer = session.createConsumer(destination);

   // Send the messages
   producer.send(session.createTextMessage("1st"));
   producer.send(session.createTextMessage("2nd"));
   session.commit();

   TextMessage m;
   m = (TextMessage) consumer.receive(1000);
   assertNotNull(m);
   assertEquals("1st", m.getText());
   session.rollback();

   // the 1st message should not be redelivered since maximumRedeliveries is
   // set to 0
   m = (TextMessage) consumer.receive(1000);
   assertNotNull(m);
   assertEquals("2nd", m.getText());
   session.commit();

}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Test
public void testInitialRedeliveryDelayZero() throws Exception {
   // Receive a message with the JMS API
   RedeliveryPolicy policy = connection.getRedeliveryPolicy();
   policy.setInitialRedeliveryDelay(0);
   policy.setUseExponentialBackOff(false);
   policy.setMaximumRedeliveries(1);

   connection.start();
   Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
   ActiveMQQueue destination = new ActiveMQQueue("TEST");
   this.makeSureCoreQueueExist("TEST");
   MessageProducer producer = session.createProducer(destination);

   MessageConsumer consumer = session.createConsumer(destination);

   // Send the messages
   producer.send(session.createTextMessage("1st"));
   producer.send(session.createTextMessage("2nd"));
   session.commit();

   TextMessage m;
   m = (TextMessage) consumer.receive(100);
   assertNotNull(m);
   assertEquals("1st", m.getText());
   session.rollback();

   m = (TextMessage) consumer.receive(100);
   assertNotNull(m);
   assertEquals("1st", m.getText());

   m = (TextMessage) consumer.receive(100);
   assertNotNull(m);
   assertEquals("2nd", m.getText());
   session.commit();

   session.commit();
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Test
public void testInitialRedeliveryDelayOne() throws Exception {
   // Receive a message with the JMS API
   RedeliveryPolicy policy = connection.getRedeliveryPolicy();
   policy.setInitialRedeliveryDelay(1000);
   policy.setUseExponentialBackOff(false);
   policy.setMaximumRedeliveries(1);

   connection.start();
   Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
   ActiveMQQueue destination = new ActiveMQQueue("TEST");
   this.makeSureCoreQueueExist("TEST");
   MessageProducer producer = session.createProducer(destination);

   MessageConsumer consumer = session.createConsumer(destination);

   // Send the messages
   producer.send(session.createTextMessage("1st"));
   producer.send(session.createTextMessage("2nd"));
   session.commit();

   TextMessage m;
   m = (TextMessage) consumer.receive(100);
   assertNotNull(m);
   assertEquals("1st", m.getText());
   session.rollback();

   m = (TextMessage) consumer.receive(100);
   assertNull(m);

   m = (TextMessage) consumer.receive(2000);
   assertNotNull(m);
   assertEquals("1st", m.getText());

   m = (TextMessage) consumer.receive(100);
   assertNotNull(m);
   assertEquals("2nd", m.getText());
   session.commit();
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
protected RedeliveryPolicy getRedeliveryPolicy() {
   RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
   redeliveryPolicy.setInitialRedeliveryDelay(0);
   redeliveryPolicy.setRedeliveryDelay(1000);
   redeliveryPolicy.setMaximumRedeliveries(3);
   redeliveryPolicy.setBackOffMultiplier((short) 2);
   redeliveryPolicy.setUseExponentialBackOff(true);
   return redeliveryPolicy;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
   ActiveMQConnectionFactory answer = super.createConnectionFactory();
   RedeliveryPolicy policy = new RedeliveryPolicy();
   policy.setMaximumRedeliveries(3);
   policy.setBackOffMultiplier((short) 1);
   policy.setRedeliveryDelay(0);
   policy.setInitialRedeliveryDelay(0);
   policy.setUseExponentialBackOff(false);
   answer.setRedeliveryPolicy(policy);
   return answer;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
   ActiveMQConnectionFactory answer = super.createConnectionFactory();
   RedeliveryPolicy policy = new RedeliveryPolicy();
   policy.setMaximumRedeliveries(3);
   policy.setBackOffMultiplier((short) 1);
   policy.setInitialRedeliveryDelay(10);
   policy.setUseExponentialBackOff(false);
   answer.setRedeliveryPolicy(policy);
   return answer;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
   ActiveMQConnectionFactory connectionFactory = super.createConnectionFactory();
   RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
   redeliveryPolicy.setMaximumRedeliveries(0);
   connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
   return connectionFactory;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
protected ConnectionFactory createConnectionFactory() throws Exception {
   ActiveMQConnectionFactory answer = (ActiveMQConnectionFactory) super.createConnectionFactory();
   RedeliveryPolicy policy = new RedeliveryPolicy();
   policy.setMaximumRedeliveries(3);
   policy.setRedeliveryDelay(0);
   policy.setInitialRedeliveryDelay(0);
   policy.setUseExponentialBackOff(false);
   answer.setRedeliveryPolicy(policy);
   return answer;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
/**
 * Connection factory for the embedded ActiveMQ broker used in tests.
 *
 * @return JMS connection factory bean.
 */
@Bean
@DependsOn("embeddedTestActiveMqBroker")
public ConnectionFactory amqConnectionFactory(
    final RedeliveryPolicy inRedeliveryPolicy) {
    final ActiveMQConnectionFactory theConnectionFactory =
        new ActiveMQConnectionFactory(mJmsBrokerUrl);
    theConnectionFactory.setRedeliveryPolicy(inRedeliveryPolicy);

    return theConnectionFactory;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
/**
 * ActiveMQ redelivery policy.
 *
 * @return ActiveMQ redelivery policy bean.
 */
@Bean
public RedeliveryPolicy amqRedeliveryPolicy() {
    final RedeliveryPolicy theRedeliveryPolicy = new RedeliveryPolicy();
    theRedeliveryPolicy.setMaximumRedeliveries(0);
    return theRedeliveryPolicy;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
private ActiveMQConnection getActiveMQConnection(MessageActivationSpec activationSpec, InvocationHandler invocationHandler) {
    if (AutoConnectionTracker.ConnectionInvocationHandler.class.isInstance(invocationHandler)) {
        final Object handle = Reflections.get(invocationHandler, "handle");
        if (TomEEManagedConnectionProxy.class.isInstance(handle)) {
            final ActiveMQManagedConnection c = ActiveMQManagedConnection.class.cast(Reflections.get(handle, "connection"));
            final ActiveMQConnection physicalConnection = ActiveMQConnection.class.cast(Reflections.get(c, "physicalConnection"));
            final RedeliveryPolicy redeliveryPolicy = activationSpec.redeliveryPolicy();
            if (redeliveryPolicy != null) {
                physicalConnection.setRedeliveryPolicy(redeliveryPolicy);
            }
            return physicalConnection;
        }
    }
    return null;
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
@Override
public void before() throws Throwable {
    log.info("+++ BEFORE on JUnit Rule '" + id(Rule_Mats.class) + "', JMS and MATS:");
    String sysprop_matsTestActiveMq = System.getProperty(SYSPROP_MATS_TEST_ACTIVEMQ);

    // ::: Server (BrokerService)
    // ====================================

    // :? Do we have specific brokerUrl to connect to?
    if (sysprop_matsTestActiveMq == null) {
        // -> No - the system property was not set, hence start the in-vm broker.
        log.info("Setting up in-vm ActiveMQ BrokerService '" + BROKER_NAME + "' (i.e. the MQ server).");
        _amqServer = new BrokerService();
        _amqServer.setBrokerName(BROKER_NAME);
        _amqServer.setUseJmx(false); // No need for JMX registry.
        _amqServer.setPersistent(false); // No need for persistence (prevents KahaDB dirs from being created).
        _amqServer.setAdvisorySupport(false); // No need Advisory Messages.
        _amqServer.setUseShutdownHook(false);

        // :: Set Individual DLQ
        // Hear, hear: http://activemq.2283324.n4.nabble.com/PolicyMap-api-is-really-bad-td4284307.html
        PolicyMap destinationPolicy = new PolicyMap();
        _amqServer.setDestinationPolicy(destinationPolicy);
        PolicyEntry policyEntry = new PolicyEntry();
        policyEntry.setQueue(">");
        destinationPolicy.put(policyEntry.getDestination(), policyEntry);

        IndividualDeadLetterStrategy individualDeadLetterStrategy = new IndividualDeadLetterStrategy();
        individualDeadLetterStrategy.setQueuePrefix("DLQ.");
        policyEntry.setDeadLetterStrategy(individualDeadLetterStrategy);

        _amqServer.start();
    }
    else {
        // -> Yes, there is specified a brokerUrl to connect to, so we
        log.info("SKIPPING setup of in-vm ActiveMQ BrokerService (MQ server), since System Property '"
                + SYSPROP_MATS_TEST_ACTIVEMQ + "' was set (to [" + sysprop_matsTestActiveMq + "]).");
    }

    // ::: Client (ConnectionFactory)
    // ====================================

    // :: Find which broker URL to use
    String brokerUrl;
    if (sysprop_matsTestActiveMq == null) {
        brokerUrl = "vm://" + BROKER_NAME + "?create=false";
    }
    else if (SYSPROP_VALUE_LOCALHOST.equals(sysprop_matsTestActiveMq)) {
        brokerUrl = "tcp://localhost:61616";
    }
    else {
        brokerUrl = sysprop_matsTestActiveMq;
    }
    // :: Connect to the broker
    log.info("Setting up ActiveMQ ConnectionFactory (MQ client), brokerUrl: [" + brokerUrl + "].");
    _amqClient = new ActiveMQConnectionFactory(brokerUrl);
    RedeliveryPolicy redeliveryPolicy = _amqClient.getRedeliveryPolicy();
    // :: Only try redelivery once, since the unit tests does not need any more to prove that they work.
    redeliveryPolicy.setInitialRedeliveryDelay(500);
    redeliveryPolicy.setUseExponentialBackOff(false);
    redeliveryPolicy.setMaximumRedeliveries(1);

    // ::: MatsFactory
    // ====================================

    log.info("Setting up JmsMatsFactory.");
    _matsStringSerializer = new MatsDefaultJsonSerializer();
    // Allow for override in specialization classes, in particular the one with DB.
    _matsFactory = createMatsFactory(_matsStringSerializer, _amqClient);
    // For all test scenarios, it makes no sense to have a concurrency more than 1, unless explicitly testing that.
    _matsFactory.getFactoryConfig().setConcurrency(1);
    log.info("--- BEFORE done! JUnit Rule '" + id(Rule_Mats.class) + "', JMS and MATS.");
}
 

import org.apache.activemq.RedeliveryPolicy; //导入依赖的package包/类
public RedeliveryPolicy getDefaultEntry() {
    return defaultEntry;
}