/**
 * 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.component.jms;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.jms.ConnectionFactory;

import org.apache.camel.CamelContext;
import org.apache.camel.builder.NotifyBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.test.junit4.CamelTestSupport;
import org.apache.camel.util.StopWatch;
import org.junit.Test;

import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge;

/**
 * @version
 */
public class JmsRequestReplyExclusiveReplyToConcurrentTest extends CamelTestSupport {

    private int size = 100;

    @Test
    public void testJmsRequestReplyExclusiveFixedReplyTo() throws Exception {
        NotifyBuilder builder = new NotifyBuilder(context).from("direct:start").whenDone(size).create();

        StopWatch watch = new StopWatch();
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < size; i++) {
            final Integer num = i;
            executor.submit(new Runnable() {
                @Override
                public void run() {
                    String reply = template.requestBody("direct:start", "" + num, String.class);
                    log.info("Sent {} expecting reply 'Hello {}' got --> {}", new Object[]{num, num, reply});
                    assertNotNull(reply);
                    assertEquals("Hello " + num, reply);
                }
            });
        }

        log.info("Waiting to process {} messages...", size);
        assertTrue(builder.matches(60, TimeUnit.SECONDS));

        long delta = watch.stop();
        log.info("Took {} millis", delta);

        // just sleep a bit before shutting down
        Thread.sleep(1000);

        executor.shutdownNow();
    }

    protected CamelContext createCamelContext() throws Exception {
        CamelContext camelContext = super.createCamelContext();
        ConnectionFactory connectionFactory = CamelJmsTestHelper.createConnectionFactory();
        camelContext.addComponent("activemq", jmsComponentAutoAcknowledge(connectionFactory));
        return camelContext;
    }

    @Override
    protected RouteBuilder createRouteBuilder() throws Exception {
        return new RouteBuilder() {
            @Override
            public void configure() throws Exception {
                from("direct:start")
                    .to("activemq:queue:foo?replyTo=bar&replyToType=Exclusive&concurrentConsumers=5&maxConcurrentConsumers=10&maxMessagesPerTask=100")
                    .to("log:reply")
                    .to("mock:reply");

                from("activemq:queue:foo?concurrentConsumers=5&maxConcurrentConsumers=10&maxMessagesPerTask=100")
                    .transform(body().prepend("Hello "));
            }
        };
    }
}
@@ -838,16 +838,60 @@
 t.jms;%0A%0A
+import java.util.concurrent.CountDownLatch;%0A
 import j
@@ -1497,16 +1497,22 @@
 private 
+final 
 int size
@@ -1518,16 +1518,83 @@
 e = 100;
+%0A    private final CountDownLatch latch = new CountDownLatch(size);
 %0A%0A    @T
@@ -1679,114 +1679,8 @@
 n %7B%0A
-        NotifyBuilder builder = new NotifyBuilder(context).from(%22direct:start%22).whenDone(size).create();%0A%0A
     
@@ -2268,32 +2268,71 @@
  + num, reply);%0A
+                    latch.countDown();%0A
                 
@@ -2425,45 +2425,197 @@
 e);%0A
-        assertTrue(builder.matches(60
+%0A        // if any of the assertions above fails then the latch will not get decremented %0A        assertTrue(%22All assertions outside the main thread above should have passed%22, latch.await(3
 , Ti

fix the potential false positive test behaviour