使用 apache camel 测试回滚事务

Question

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

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

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

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

Given my example code how do I make it roll back the transaction?

我的路线是这样的

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();
    }
}

Answer 1

Add bean to Context (will add javaconfig option to this)

正如 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>

Alternatively add transaction to dao

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

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

使用 apache camel 测试回滚事务

rollback transaction in test with apache camel

java

integration-testing

transactions

message-queue

apache-camel