gistfile1.java

import com.aggregating.StringListAggregator;
import org.apache.activemq.camel.component.ActiveMQComponent;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.LoggingLevel;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.AvailablePortFinder;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.apache.camel.util.jndi.JndiContext;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.util.EntityUtils;
import org.junit.Test;

import javax.naming.Context;
import java.io.IOException;

/**
 * @author stibomar
 * @since 19/07/13 13:25
 */

public class JaxrsRequestWithSplittingBodyAndAggregatingJmsResponsesTest extends CamelTestSupport {

    private static final String PORT_PATH = AvailablePortFinder.getNextAvailable() + "/CxfRsConsumerTest";
    private static final String CXF_RS_ENDPOINT_URI = "cxfrs:http://localhost:" + PORT_PATH +
            "/rest?resourceClasses=com.project.tests.testbean.JaxrsTestingServiceResource" +
            "&bindingStyle=SimpleConsumer";

    private HttpClient httpClient;

    @Override
    public void setUp() throws Exception {
        super.setUp();
        httpClient = new DefaultHttpClient();
    }

    @Override
    public void tearDown() throws Exception {
        super.tearDown();
        httpClient.getConnectionManager().shutdown();
    }

    private String getUriToJmsEndpointWithoutTransferedExchange() {
        return "jms:queue:testingNoExchangeTransfered";
    }

    private String getUriToJmsEndpointWithTransferedExchange() {
        return "jms:queue:testingExchangeTransfered?transferExchange=true";
    }

    @Test
    public void noPropertiesExpectedFromJmsWithoutTransferedExchange() throws IOException {
        HttpGet get = new HttpGet("http://localhost:" + PORT_PATH + "/rest/testing/jmsWithoutExchangeTransfer");
        HttpResponse response = httpClient.execute(get);
        HttpEntity entity = response.getEntity();
        String responseString = EntityUtils.toString(entity, "UTF-8");

        assertEquals(200, response.getStatusLine().getStatusCode());
        assertEquals("[Transformed: exchangeItem1, Transformed: exchangeItem2, Transformed: exchangeItem3]",
                responseString);
    }

    @Test
    public void exchangePropertiesShouldBePreservedFromJmsWithTransferedExchange()
            throws IOException, InterruptedException {
        MockEndpoint mock = getMockEndpoint("mock:jmsTransferedExchange");
        mock.expectedBodiesReceived("Transformed: transferedExchangeItem1", "Transformed: transferedExchangeItem2", "Transformed: transferedExchangeItem3");
        mock.expectedPropertyReceived("transferedProperty", "Please survive through JMS!");

        HttpGet get = new HttpGet("http://localhost:" + PORT_PATH + "/rest/testing/jmsWithExchangeTransfer");
        HttpResponse response = httpClient.execute(get);
        HttpEntity entity = response.getEntity();
        String responseString = EntityUtils.toString(entity, "UTF-8");

        assertEquals(200, response.getStatusLine().getStatusCode());
        assertEquals("[Transformed: transferedExchangeItem1, Transformed: transferedExchangeItem2, Transformed: transferedExchangeItem3]", responseString);

        assertMockEndpointsSatisfied();
    }

    @Override
    protected RouteBuilder createRouteBuilder() {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from(CXF_RS_ENDPOINT_URI)
                    .recipientList(simple("direct:${header.operationName}"));

                from(getUriToJmsEndpointWithoutTransferedExchange())
                    .transform().simple("Transformed: ${body}");

                from("direct:jmsWithoutTransferedExchange")
                    .setBody().constant("exchangeItem1,exchangeItem2,exchangeItem3")
                    .split().tokenize(",").aggregationStrategy(new StringListAggregator())
                        .to(getUriToJmsEndpointWithoutTransferedExchange())
                        .log(LoggingLevel.INFO, "Transformed: ${body}")
                        .end()
                    .convertBodyTo(String.class);


                from(getUriToJmsEndpointWithTransferedExchange())
                        .transform().simple("Transformed: ${body}")
                        .to("mock:jmsTransferedExchange");

                from("direct:jmsWithTransferedExchange")
                    .setBody().constant("transferedExchangeItem1,transferedExchangeItem2,transferedExchangeItem3")
                    .setProperty("transferedProperty", constant("Please survive through JMS!"))
                    .split().tokenize(",").aggregationStrategy(new StringListAggregator())
                        .to(ExchangePattern.InOut, getUriToJmsEndpointWithTransferedExchange())
                        .end()
                    .process(new Processor() {
                        @Override
                        public void process(Exchange exchange) throws Exception {
                            String testedProperty = (String) exchange.getProperty("transferedProperty");
                            assertEquals("Please survive through JMS!", testedProperty);
                        }
                    })
                    .convertBodyTo(String.class);
            }
        };
    }

    @Override
    protected Context createJndiContext() throws Exception {
        JndiContext jndiContext = new JndiContext();
        ActiveMQComponent amq = ActiveMQComponent.activeMQComponent("vm://localhost?broker.persistent=false");
        amq.setCamelContext(context);
        jndiContext.bind("jms", amq);
        return jndiContext;
    }

}