mirror of https://github.com/apache/activemq.git
test to verify camel transaction rollback bubbling back to amq rollback and dql processing
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1143930 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
060f97dd87
commit
d2bc650212
|
@ -0,0 +1,95 @@
|
||||||
|
/**
|
||||||
|
* Licensed to the Apache Software Foundation (ASF) under one or more
|
||||||
|
* contributor license agreements. See the NOTICE file distributed with
|
||||||
|
* this work for additional information regarding copyright ownership.
|
||||||
|
* The ASF licenses this file to You under the Apache License, Version 2.0
|
||||||
|
* (the "License"); you may not use this file except in compliance with
|
||||||
|
* the License. You may obtain a copy of the License at
|
||||||
|
*
|
||||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
*
|
||||||
|
* Unless required by applicable law or agreed to in writing, software
|
||||||
|
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
* See the License for the specific language governing permissions and
|
||||||
|
* limitations under the License.
|
||||||
|
*/
|
||||||
|
package org.apache.activemq.camel;
|
||||||
|
|
||||||
|
import javax.jms.Connection;
|
||||||
|
import javax.jms.MessageProducer;
|
||||||
|
import javax.jms.Session;
|
||||||
|
import javax.jms.TextMessage;
|
||||||
|
import org.apache.activemq.ActiveMQConnectionFactory;
|
||||||
|
import org.apache.activemq.broker.BrokerService;
|
||||||
|
import org.apache.activemq.command.ActiveMQQueue;
|
||||||
|
import org.apache.activemq.util.Wait;
|
||||||
|
import org.apache.camel.spring.SpringTestSupport;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.context.support.AbstractXmlApplicationContext;
|
||||||
|
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||||
|
|
||||||
|
public class DlqTest extends SpringTestSupport {
|
||||||
|
private static final Logger LOG = LoggerFactory.getLogger(DlqTest.class);
|
||||||
|
BrokerService broker = null;
|
||||||
|
int messageCount;
|
||||||
|
|
||||||
|
public void testSendToDlq() throws Exception {
|
||||||
|
sendJMSMessageToKickOffRoute();
|
||||||
|
|
||||||
|
LOG.info("Wait for dlq message...");
|
||||||
|
|
||||||
|
assertTrue(Wait.waitFor(new Wait.Condition() {
|
||||||
|
@Override
|
||||||
|
public boolean isSatisified() throws Exception {
|
||||||
|
return broker.getAdminView().getTotalEnqueueCount() == 2;
|
||||||
|
}
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
private void sendJMSMessageToKickOffRoute() throws Exception {
|
||||||
|
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://testDlq");
|
||||||
|
factory.setWatchTopicAdvisories(false);
|
||||||
|
Connection connection = factory.createConnection();
|
||||||
|
connection.start();
|
||||||
|
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
|
||||||
|
MessageProducer producer = session.createProducer(new ActiveMQQueue("fidEtpOrders"));
|
||||||
|
TextMessage message = session.createTextMessage("Some Text, messageCount:" + messageCount++);
|
||||||
|
message.setJMSCorrelationID("pleaseCorrelate");
|
||||||
|
producer.send(message);
|
||||||
|
connection.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
private BrokerService createBroker(boolean deleteAllMessages) throws Exception {
|
||||||
|
BrokerService brokerService = new BrokerService();
|
||||||
|
brokerService.setDeleteAllMessagesOnStartup(deleteAllMessages);
|
||||||
|
brokerService.setBrokerName("testDlq");
|
||||||
|
brokerService.setAdvisorySupport(false);
|
||||||
|
brokerService.setDataDirectory("target/data");
|
||||||
|
return brokerService;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
protected AbstractXmlApplicationContext createApplicationContext() {
|
||||||
|
|
||||||
|
deleteDirectory("target/data");
|
||||||
|
|
||||||
|
// make broker available to recovery processing on app context start
|
||||||
|
try {
|
||||||
|
broker = createBroker(true);
|
||||||
|
broker.start();
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new RuntimeException("Failed to start broker", e);
|
||||||
|
}
|
||||||
|
|
||||||
|
return new ClassPathXmlApplicationContext("org/apache/activemq/camel/dlq.xml");
|
||||||
|
}
|
||||||
|
|
||||||
|
public static class CanError {
|
||||||
|
public String enrich(String body) throws Exception {
|
||||||
|
LOG.info("Got body: " + body);
|
||||||
|
throw new RuntimeException("won't enrich today!");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
|
@ -0,0 +1,54 @@
|
||||||
|
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||||
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xmlns:context="http://www.springframework.org/schema/context"
|
||||||
|
xsi:schemaLocation="
|
||||||
|
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
|
||||||
|
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd
|
||||||
|
http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
|
||||||
|
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
|
||||||
|
">
|
||||||
|
|
||||||
|
<bean id="transactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
|
||||||
|
<property name="connectionFactory" ref="pooledConnectionFactory"/>
|
||||||
|
</bean>
|
||||||
|
|
||||||
|
<bean id="localJMS" class="org.apache.activemq.camel.component.ActiveMQComponent">
|
||||||
|
<property name="connectionFactory" ref="pooledConnectionFactory"/>
|
||||||
|
<property name="transactionManager" ref="transactionManager"/>
|
||||||
|
<property name="transacted" value="true"/>
|
||||||
|
</bean>
|
||||||
|
|
||||||
|
<camelContext id="camel" xmlns="http://camel.apache.org/schema/spring">
|
||||||
|
<route id="orders">
|
||||||
|
<from uri="localJMS:fidEtpOrders" />
|
||||||
|
<transacted/>
|
||||||
|
<!-- <log loggingLevel="INFO" message="from activemq:fidessaTrades: ${body}" /> -->
|
||||||
|
<bean ref="canError" method="enrich" />
|
||||||
|
<wireTap uri="localJMS:ordersTap" />
|
||||||
|
<to uri="localJMS:fidessaOrders" />
|
||||||
|
</route>
|
||||||
|
</camelContext>
|
||||||
|
|
||||||
|
<bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
|
||||||
|
<property name="brokerURL" value="vm://testDlq" />
|
||||||
|
<property name="redeliveryPolicy" ref="redeliveryPolicy"/>
|
||||||
|
</bean>
|
||||||
|
|
||||||
|
<bean id="redeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
|
||||||
|
<property name="maximumRedeliveries" value="1"/>
|
||||||
|
</bean>
|
||||||
|
|
||||||
|
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
|
||||||
|
<property name="maxConnections" value="8" />
|
||||||
|
<property name="connectionFactory" ref="activemqConnectionFactory" />
|
||||||
|
</bean>
|
||||||
|
|
||||||
|
<!-- only for jta - not jms tm
|
||||||
|
bean id="resourceManager" class="org.apache.activemq.pool.ActiveMQResourceManager" init-method="recoverResource">
|
||||||
|
<property name="transactionManager" ref="transactionManager" />
|
||||||
|
<property name="connectionFactory" ref="activemqConnectionFactory" />
|
||||||
|
<property name="resourceName" value="activemq.default" />
|
||||||
|
</bean -->
|
||||||
|
|
||||||
|
<bean id="canError" class="org.apache.activemq.camel.DlqTest$CanError"/>
|
||||||
|
</beans>
|
|
@ -575,7 +575,7 @@ public class ActiveMQSession implements Session, QueueSession, TopicSession, Sta
|
||||||
throw new javax.jms.IllegalStateException("Not a transacted session");
|
throw new javax.jms.IllegalStateException("Not a transacted session");
|
||||||
}
|
}
|
||||||
if (LOG.isDebugEnabled()) {
|
if (LOG.isDebugEnabled()) {
|
||||||
LOG.debug(getSessionId() + " Transaction Rollback");
|
LOG.debug(getSessionId() + " Transaction Rollback, txid:" + transactionContext.getTransactionId());
|
||||||
}
|
}
|
||||||
transactionContext.rollback();
|
transactionContext.rollback();
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue