RunawaySalesforceIntegrationTest.java

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.camel.AggregationStrategy;
import org.apache.camel.Exchange;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.component.salesforce.AbstractSalesforceTestBase;
import org.apache.camel.component.salesforce.dto.generated.QueryRecordsContact;
import org.junit.jupiter.api.Test;

import static org.junit.jupiter.api.Assertions.assertSame;

public class RunawaySalesforceIntegrationTest extends AbstractSalesforceTestBase {

    @Test
    public void testAggregateQuery() throws Exception {
        MockEndpoint mock = getMockEndpoint("mock:messageTracker");
        AtomicInteger index = new AtomicInteger(0);
        // messages should alternate between a String and a QueryRecordsContact instance
        mock.whenAnyExchangeReceived(exchange -> {
            if (index.getAndIncrement() % 2 == 0) {
                assertSame(String.class, exchange.getIn().getBody().getClass());
            }
            else {
                assertSame(QueryRecordsContact.class, exchange.getIn().getBody().getClass());
            }
        });
        List<String> queries = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            queries.add("SELECT Id FROM Contact LIMIT 1");
        }
        mock.expectedMessageCount(queries.size() * 2);
        template().sendBody("direct:aggregateQuery", queries);
        mock.assertIsSatisfied();
    }

    @Test
    public void testPlainQuery() throws Exception {
        MockEndpoint mock = getMockEndpoint("mock:messageTracker");
        AtomicInteger index = new AtomicInteger(0);
        // messages should alternate between a String and a QueryRecordsContact instance
        mock.whenAnyExchangeReceived(exchange -> {
            if (index.getAndIncrement() % 2 == 0) {
                assertSame(String.class, exchange.getIn().getBody().getClass());
            }
            else {
                assertSame(QueryRecordsContact.class, exchange.getIn().getBody().getClass());
            }
        });
        List<String> queries = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            queries.add("SELECT Id FROM Contact LIMIT 1");
        }
        mock.expectedMessageCount(queries.size() * 2);
        template().sendBody("direct:plainQuery", queries);
        mock.assertIsSatisfied();
    }

    @Override
    protected RouteBuilder doCreateRouteBuilder() throws Exception {

        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {

                from("direct:aggregateQuery")
                    .split(body())
                        .aggregate(constant(1), new AggregationStrategy() {
                            @Override
                            public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
                                return newExchange;
                            }
                        }).completionPredicate(constant(true))
                        .log("sending")
                            .to("mock:messageTracker")
                            .to("salesforce:query?sObjectClass=" + QueryRecordsContact.class.getName())
                            .log("result")
                            .to("mock:messageTracker");

                from("direct:plainQuery")
                    .split(body())
                        .log("sending")
                        .to("mock:messageTracker")
                        .to("salesforce:query?sObjectClass=" + QueryRecordsContact.class.getName())
                        .log("result")
                        .to("mock:messageTracker");
            }
        };
    }
}