<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.camel.itest.sql;

import java.net.ConnectException;
import java.util.concurrent.atomic.AtomicInteger;

import javax.sql.DataSource;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.NotifyBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.test.junit4.CamelSpringTestSupport;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;

/**
 * Jms with JDBC idempotent consumer test.
 */
public class FromJmsToJdbcIdempotentConsumerToJmsTest extends CamelSpringTestSupport {

    protected JdbcTemplate jdbcTemplate;
    protected DataSource dataSource;

    @Override
    protected AbstractApplicationContext createApplicationContext() {
        return new ClassPathXmlApplicationContext("org/apache/camel/itest/sql/FromJmsToJdbcIdempotentConsumerToJmsTest.xml");
    }

    @Override
    @Before
    public void setUp() throws Exception {
        super.setUp();

        dataSource = context.getRegistry().lookup(getDatasourceName(), DataSource.class);
        jdbcTemplate = new JdbcTemplate(dataSource);
        jdbcTemplate.afterPropertiesSet();
    }

    protected String getDatasourceName() {
        return "myNonXADataSource";
    }

    @Test
    public void testJmsToJdbcJmsCommit() throws Exception {
        checkInitialState();

        // use a notify to know when the message is done
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

        // use mock during testing as well
        getMockEndpoint("mock:a").expectedMessageCount(1);
        getMockEndpoint("mock:b").expectedMessageCount(1);

        template.sendBodyAndHeader("activemq:queue:inbox", "A", "uid", 123);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 1 message", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(1, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        Object out = consumer.receiveBody("activemq:queue:outbox", 3000);
        assertEquals("DONE-A", out);
    }

    @Test
    public void testJmsToJdbcJmsRollbackAtA() throws Exception {
        checkInitialState();

        // use a notify to know that after 1+6 (1 original + 6 redelivery) attempts from AcitveMQ
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(7).create();

        // TODO: occasionally we get only 6 instead of 7 expected exchanges which's most probably an issue in ActiveMQ itself
        getMockEndpoint("mock:a").expectedMessageCount(7);
        // force exception to occur at mock a
        getMockEndpoint("mock:a").whenAnyExchangeReceived(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                throw new ConnectException("Forced cannot connect to database");
            }
        });
        getMockEndpoint("mock:b").expectedMessageCount(0);

        template.sendBodyAndHeader("activemq:queue:inbox", "A", "uid", 123);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 7 message", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(0, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        assertNull(consumer.receiveBody("activemq:queue:outbox", 3000));

        // the message should have been moved to the AMQ DLQ queue
        assertEquals("A", consumer.receiveBody("activemq:queue:ActiveMQ.DLQ", 3000));
    }

    @Test
    public void testJmsToJdbcJmsRollbackAtB() throws Exception {
        checkInitialState();

        // use a notify to know that after 1+6 (1 original + 6 redelivery) attempts from AcitveMQ
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(7).create();

        getMockEndpoint("mock:a").expectedMessageCount(7);
        getMockEndpoint("mock:b").expectedMessageCount(7);
        // force exception to occur at mock b
        getMockEndpoint("mock:b").whenAnyExchangeReceived(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                throw new ConnectException("Forced cannot send to AMQ queue");
            }
        });

        template.sendBodyAndHeader("activemq:queue:inbox", "B", "uid", 456);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 7 messages", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(0, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        assertNull(consumer.receiveBody("activemq:queue:outbox", 3000));

        // the message should have been moved to the AMQ DLQ queue
        assertEquals("B", consumer.receiveBody("activemq:queue:ActiveMQ.DLQ", 3000));<commit_msg>CAMEL-5856: Added comment about the occasional 6 instead of 7 expected messages.<commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.camel.itest.sql;

import java.net.ConnectException;
import java.util.concurrent.atomic.AtomicInteger;

import javax.sql.DataSource;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.NotifyBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.spi.IdempotentRepository;
import org.apache.camel.test.junit4.CamelSpringTestSupport;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;

/**
 * Jms with JDBC idempotent consumer test.
 */
public class FromJmsToJdbcIdempotentConsumerToJmsTest extends CamelSpringTestSupport {

    protected JdbcTemplate jdbcTemplate;
    protected DataSource dataSource;

    @Override
    protected AbstractApplicationContext createApplicationContext() {
        return new ClassPathXmlApplicationContext("org/apache/camel/itest/sql/FromJmsToJdbcIdempotentConsumerToJmsTest.xml");
    }

    @Override
    @Before
    public void setUp() throws Exception {
        super.setUp();

        dataSource = context.getRegistry().lookup(getDatasourceName(), DataSource.class);
        jdbcTemplate = new JdbcTemplate(dataSource);
        jdbcTemplate.afterPropertiesSet();
    }

    protected String getDatasourceName() {
        return "myNonXADataSource";
    }

    @Test
    public void testJmsToJdbcJmsCommit() throws Exception {
        checkInitialState();

        // use a notify to know when the message is done
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

        // use mock during testing as well
        getMockEndpoint("mock:a").expectedMessageCount(1);
        getMockEndpoint("mock:b").expectedMessageCount(1);

        template.sendBodyAndHeader("activemq:queue:inbox", "A", "uid", 123);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 1 message", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(1, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        Object out = consumer.receiveBody("activemq:queue:outbox", 3000);
        assertEquals("DONE-A", out);
    }

    @Test
    public void testJmsToJdbcJmsRollbackAtA() throws Exception {
        checkInitialState();

        // use a notify to know that after 1+6 (1 original + 6 redelivery) attempts from AcitveMQ
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(7).create();

        // TODO: occasionally we get only 6 instead of 7 expected exchanges which's most probably an issue in ActiveMQ itself
        getMockEndpoint("mock:a").expectedMessageCount(7);
        // force exception to occur at mock a
        getMockEndpoint("mock:a").whenAnyExchangeReceived(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                throw new ConnectException("Forced cannot connect to database");
            }
        });
        getMockEndpoint("mock:b").expectedMessageCount(0);

        template.sendBodyAndHeader("activemq:queue:inbox", "A", "uid", 123);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 7 message", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(0, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        assertNull(consumer.receiveBody("activemq:queue:outbox", 3000));

        // the message should have been moved to the AMQ DLQ queue
        assertEquals("A", consumer.receiveBody("activemq:queue:ActiveMQ.DLQ", 3000));
    }

    @Test
    public void testJmsToJdbcJmsRollbackAtB() throws Exception {
        checkInitialState();

        // use a notify to know that after 1+6 (1 original + 6 redelivery) attempts from AcitveMQ
        NotifyBuilder notify = new NotifyBuilder(context).whenDone(7).create();

        // TODO: occasionally we get only 6 instead of 7 expected exchanges which's most probably an issue in ActiveMQ itself
        getMockEndpoint("mock:a").expectedMessageCount(7);
        getMockEndpoint("mock:b").expectedMessageCount(7);
        // force exception to occur at mock b
        getMockEndpoint("mock:b").whenAnyExchangeReceived(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
                throw new ConnectException("Forced cannot send to AMQ queue");
            }
        });

        template.sendBodyAndHeader("activemq:queue:inbox", "B", "uid", 456);

        // assert mock and wait for the message to be done
        assertMockEndpointsSatisfied();
        assertTrue("Should complete 7 messages", notify.matchesMockWaitTime());

        // check that there is a message in the database and JMS queue
        assertEquals(0, jdbcTemplate.queryForInt("select count(*) from CAMEL_MESSAGEPROCESSED"));
        assertNull(consumer.receiveBody("activemq:queue:outbox", 3000));

        // the message should have been moved to the AMQ DLQ queue
        assertEquals("B", consumer.receiveBody("activemq:queue:ActiveMQ.DLQ", 3000));
{'fraction_non_alphanumeric': 0.09007780540627255, 'fraction_numerical': 0.010106681639528355, 'mean_word_length': 3.504335260115607, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5837291', 'n_tokens_mistral': 3357, 'n_tokens_neox': 3151, 'n_words': 1007}