用apache camel测试回滚事务

Question

我正在努力做一个工作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();
    }
}

Answer 1

将bean添加到上下文（将向此添加javaconfig选项）

正如Andreas在评论部分提到的，您可以添加。事务处理到路由，并确保事务管理器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方法中使用下面的注释。仍然需要事务maanger bean，但您可以在注释中按名称指定它。

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