使用 apache camel 测试回滚事务答案

【问题标题】：rollback transaction in test with apache camel使用 apache camel 测试回滚事务
【发布时间】：2015-12-31 11:53:59
【问题描述】：

我正在努力进行一个有效的 junit 测试，以回滚在骆驼路由期间发生的操作。

我有一个监听目录的骆驼路由设置。它需要一个 csv 文件。当 csv 文件出现时，它会创建新的 SearchAnalytics 数据。它为 csv 文件中的每一行添加一个新行到表中。

我设置的默认 spring 事务方法似乎不适用于骆驼路由上发生的操作。

下面的代码有效。但是，它会永久保存数据并且不会回滚插入。这意味着除非我手动删除数据，否则测试只会通过一次。

根据我的示例代码，我该如何让它回滚事务？

我的路线是这样的

from("ftp://some__remote__ftp_dir_path")
  .routeId("searchAnalyticsImport")             
  .choice()
    .when(simple("${in.header.CamelFileName} contains '.csv'"))
    .split().method("csvSplitter", "iterator").streaming() // reads the csv file returns data objects
    .processRef("searchAnalyticsProcesser")  // this some dao saves
    .to(Queues.SOME_REQUEST)
.end();

Junit 测试

@TransactionConfiguration(defaultRollback = true, transactionManager = "transactionManager")
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { TestAppConfig.class})
public class searchAnalyticsImportTest  {

    @EndpointInject(uri = "mock:sippmatcher.requestqueue?preserveMessageQos=true")
    private MockEndpoint mockEndpointRequest;

    @Before
    public void setup() throws Exception {

        camelContext.getRouteDefinition("searchAnalyticsImport").adviceWith(camelContext, new AdviceWithRouteBuilder() {
            @Override
            public void configure() throws Exception {

                replaceFromWith("file://"+this.getClass().getResource("path to folder etc...")+"?noop=true");

                interceptSendToEndpoint(Queues.SOME_REQUEST)
                        .skipSendToOriginalEndpoint()
                        .to(mockEndpointRequest);
            }
        });
    }

    @Test
    public void simpleTest() throws Exception{

        // there are 2 results in the test csv file.. need to poll the results till it completes
        PollWithTimeout.run("keep polling until route has been statisfied", 15000, new PollWithTimeout.Attempt() {
            @Override
            public boolean complete() {
                Date dateTime1MinuteAgo = new DateTime().minusMinutes(1).toDate();

                Integer newSearchCount = searchAnalysiticDao.findBySearchStartedAfter(dateTime1MinuteAgo).size();

                System.out.println("Recently added count: " + newSearchCount);
                return (newSearchCount == 2);
            }
        });

        mockEndpointRequest.expectedMessageCount(2);
        mockEndpointRequest.assertIsSatisfied();
    }
}

【问题讨论】：

据我所知，您的路线不是事务性的。您是否尝试过交易 (camel.apache.org/transactional-client.html)？
如何修改路由以包含交易？
查看我之前的回答（camel docs）。应该像在路由定义的开头添加 .transacted() 一样简单。也许您还必须提供您的事务管理器的名称..
我已经尝试将 transaction() 添加到路由中。 java config中如何指定事务管理器的名称？
查看我发布的链接。定义一个事务策略，如：然后你可以像这样使用它：transacted("PROPAGATION_REQUIRED") 在你的路由定义中......

标签： java transactions apache-camel integration-testing message-queue

【解决方案1】：

将 bean 添加到 Context（将添加 javaconfig 选项）

正如 Andreas 在评论部分中提到的，您可以将 .transacted 添加到路由并确保您的事务管理器 bean 被注入到您的上下文文件中。

路线

from("ftp://some__remote__ftp_dir_path")
  .routeId("searchAnalyticsImport") 
  .end()
  .transacted("PROPAGATION_REQUIRED")            
  etc....

上下文 Bean 配置

<bean id="jmsTransactionManager"
      class="org.springframework.jms.connection.JmsTransactionManager">
    <property name="connectionFactory" ref="pooledConnectionFactory" />
    <property name="defaultTimeout" value="30"/>
</bean>

<bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    <property name="transactionManager" ref="jmsTransactionManager" />
    <property name="propagationBehaviorName" value="PROPAGATION_REQUIRED" />
</bean>

或者将事务添加到 dao

您可以在 searchAnalyticsProcesser 中调用的 dao 方法中使用以下注释。事务管理器 bean 仍然是必需的，但您可以在注解中按名称指定它。

@Transactional(
    propagation = Propagation.REQUIRED,
    readOnly = false,
    value="transactionManager",
    rollbackFor = {
       Exception.class
      })
public void insertStuff()

【讨论】：