/**
 * 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.activemq.transport.mqtt;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.TransportConnector;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

public class PahoMQTNioTTest extends PahoMQTTTest {

    private static final Logger LOG = LoggerFactory.getLogger(PahoMQTNioTTest.class);

    @Override
    protected String getProtocolScheme() {
        return "mqtt+nio";
    }

    @Test(timeout=300000)
    public void testLotsOfClients() throws Exception {

        final int CLIENTS = 100;
        addMQTTConnector();
        TransportConnector openwireTransport = brokerService.addConnector("tcp://localhost:0");
        brokerService.start();

        ActiveMQConnection activeMQConnection = (ActiveMQConnection) new ActiveMQConnectionFactory(openwireTransport.getConnectUri()).createConnection();
        activeMQConnection.start();
        Session s = activeMQConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer consumer = s.createConsumer(s.createTopic("test"));

        final AtomicInteger receiveCounter = new AtomicInteger();
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                receiveCounter.incrementAndGet();
            }
        });

        final AtomicReference<Throwable> asyncError = new AtomicReference<Throwable>();
        final CountDownLatch connectedDoneLatch = new CountDownLatch(CLIENTS);
        final CountDownLatch disconnectDoneLatch = new CountDownLatch(CLIENTS);
        final CountDownLatch sendBarrier = new CountDownLatch(1);
        for( int i=0; i < CLIENTS; i++ ) {
            Thread.sleep(10);
            new Thread(null, null, "client:"+i) {
                @Override
                public void run() {
                    try {
                        MqttClient client = new MqttClient("tcp://localhost:" + mqttConnector.getConnectUri().getPort(), Thread.currentThread().getName(), new MemoryPersistence());
                        client.connect();
                        connectedDoneLatch.countDown();
                        sendBarrier.await();
                        client.publish("test", "hello".getBytes(), 1, false);
                        client.disconnect();
                        client.close();
                    } catch (Throwable e) {
                        e.printStackTrace();
                        asyncError.set(e);
                    } finally {
                        disconnectDoneLatch.countDown();
                    }
                }
            }.start();
        }

        connectedDoneLatch.await();
        assertNull("Async error: "+asyncError.get(),asyncError.get());
        System.out.println("All clients connected...");
        sendBarrier.countDown();

        // We should eventually get all the messages.
        within(30, TimeUnit.SECONDS, new Task() {
            @Override
            public void run() throws Exception {
                assertTrue(receiveCounter.get() == CLIENTS);
            }
        });

        disconnectDoneLatch.await();
        assertNull("Async error: "+asyncError.get(),asyncError.get());

    }


}
@@ -1843,11 +1843,106 @@
 S = 
-100
+Integer.getInteger(%22PahoMQTNioTTest.CLIENTS%22, 100);%0A        LOG.info(%22Using: %22+CLIENTS+%22 clients%22)
 ;%0A  
@@ -3567,32 +3567,135 @@
 arrier.await();%0A
+                        for( int i=0; i %3C 10; i++) %7B%0A                           Thread.sleep(1000);%0A   
                 
@@ -3752,24 +3752,50 @@
  1, false);%0A
+                        %7D%0A
             
@@ -4281,86 +4281,113 @@
     
-System.out.println(%22All clients connected...%22);%0A        sendBarrier.countDown(
+sendBarrier.countDown();%0A%0A        LOG.info(%22All clients connected... waiting to receive sent messages...%22
 );%0A%0A
@@ -4615,24 +4615,27 @@
 ) == CLIENTS
+*10
 );%0A         
@@ -4644,32 +4644,77 @@
  %7D%0A        %7D);%0A%0A
+        LOG.info(%22All messages received.%22);%0A%0A
         disconne
@@ -4815,8 +4815,9 @@
    %7D%0A%0A%0A%7D
+%0A

Improve the Paho over NIO tests, allow scaling the test using a -DPahoMQTNioTTest.CLIENTS=xxxx mvn build arg.