CODEWITHSUNDEEP
javatransactionsapache-camelintegration-testingmessage-queue
Robbo_UK
Robbo_UK

Reputation: 12149

rollback transaction in test with apache camel

I am struggling to make a working junit test that rolls back actions that the occurred during the camel routing.

I have a camel route setup that listens on a directory. It is expecting a csv file. When the csv file appears it then creates new SearchAnalytics data. It adds a new row into a table per each line in the csv file.

The default spring transaction methods that I have put do not seem to apply to actions that occur on the camel routing.

The code below works. However it saves the data permanently and does not rollback the insert. This means that the test will only pass once unless I manually delete the data.

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

my route looks like this

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 test

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

Upvotes: 2

Views: 2691

Answers (1)

KDoyle
KDoyle

Reputation: 123

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

As mentioned in the comment section by Andreas you can add .transacted to the route and ensure that your transaction manager bean is injected in your context file.

Route

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

Context Bean Config

<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

you can use the below annotation at the dao method that is being called in the searchAnalyticsProcesser. A transaction maanger bean will still be required but you can specify it by name in the annotation.

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

Upvotes: 1

Related Questions