Small improvements regarding test execution.

Author: Kai Hudalla

server/src/main/java/org/eclipse/hono/server/HonoServer.java

@@ -119,7 +119,10 @@ ProtonServerOptions createServerOptions() {
     @Override
     public void stop(Future<Void> shutdownHandler) {
         if (server != null) {
-            server.close(done -> shutdownHandler.complete());
+            server.close(done -> {
+                LOG.info("HonoServer has been shut down");
+                shutdownHandler.complete();
+            });
         }
     }
 

server/src/main/java/org/eclipse/hono/telemetry/impl/MessageDiscardingTelemetryAdapter.java

@@ -13,6 +13,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.function.Consumer;
 
 import org.apache.qpid.proton.message.Message;
 import org.slf4j.Logger;
@@ -34,20 +35,30 @@ public final class MessageDiscardingTelemetryAdapter extends BaseTelemetryAdapte
     private final long pauseThreshold;
     private final long pausePeriod;
     private Map<String, LinkStatus> statusMap = new HashMap<>();
+    private Consumer<Message> messageConsumer;
 
     public MessageDiscardingTelemetryAdapter() {
-        this(0, 0);
+        this(0, 0, null);
+    }
+
+    /**
+     * 
+     * @param consumer a consumer that is invoked for every message received.
+     */
+    public MessageDiscardingTelemetryAdapter(final Consumer<Message> consumer) {
+        this(0, 0, consumer);
     }
 
     /**
      * @param pauseThreshold the number of messages after which the sender is paused. If set to 0 (zero) the sender will
      *                       never be paused.
      * @param pausePeriod the number of milliseconds after which the sender is resumed.
      */
-    public MessageDiscardingTelemetryAdapter(final long pauseThreshold, final long pausePeriod) {
+    public MessageDiscardingTelemetryAdapter(final long pauseThreshold, final long pausePeriod, final Consumer<Message> consumer) {
         super(0, 1);
         this.pauseThreshold = pauseThreshold;
         this.pausePeriod = pausePeriod;
+        this.messageConsumer = consumer;
     }
 
     @Override
@@ -60,6 +71,9 @@ public void processTelemetryData(final Message data, final String linkId) {
         }
         LOG.debug("processing telemetry data [id: {}, to: {}, content-type: {}]", data.getMessageId(), data.getAddress(),
                 data.getContentType());
+        if (messageConsumer != null) {
+            messageConsumer.accept(data);
+        }
         status.onMsgReceived();
     }
 

server/src/test/java/org/eclipse/hono/server/HonoServerIntegrationTest.java

@@ -56,9 +56,17 @@ public class HonoServerIntegrationTest {
     private static final String NAME_HONO_CONNECTION_FACTORY = "hono";
     private static final Logger LOG          = LoggerFactory.getLogger(HonoServerIntegrationTest.class);
     private static final String BIND_ADDRESS = InetAddress.getLoopbackAddress().getHostAddress();
+    private Connection connection;
+    private Vertx vertx;
 
     @After
-    public void disconnect(final TestContext ctx) {
+    public void disconnect(final TestContext ctx) throws JMSException {
+        if (connection != null) {
+            connection.close();
+        }
+        if (vertx != null) {
+            vertx.close(ctx.asyncAssertSuccess());
+        }
     }
 
     private static HonoServer createServer(final Endpoint telemetryEndpoint) {
@@ -72,14 +80,18 @@ private static HonoServer createServer(final Endpoint telemetryEndpoint) {
     @Test
     public void testTelemetryUpload(final TestContext ctx) throws Exception {
 
-        Vertx vertx = Vertx.vertx();
+        vertx = Vertx.vertx();
         LOG.debug("starting telemetry upload test");
         int count = 110;
+        final Async messagesReceived = ctx.async(count);
         final Async deployed = ctx.async();
 
         TelemetryEndpoint telemetryEndpoint = new TelemetryEndpoint(vertx, false);
         HonoServer server = createServer(telemetryEndpoint);
-        vertx.deployVerticle(new MessageDiscardingTelemetryAdapter());
+        vertx.deployVerticle(new MessageDiscardingTelemetryAdapter(msg -> {
+            messagesReceived.countDown();
+            LOG.debug("Received message [id: {}]", msg.getMessageId());
+        }));
         vertx.deployVerticle(InMemoryAuthorizationService.class.getName());
         vertx.deployVerticle(server, res -> {
             ctx.assertTrue(res.succeeded());
@@ -91,7 +103,7 @@ public void testTelemetryUpload(final TestContext ctx) throws Exception {
         ConnectionFactory factory = (ConnectionFactory) context.lookup(NAME_HONO_CONNECTION_FACTORY);
         Destination telemetryAddress = (Destination) context.lookup(TelemetryConstants.TELEMETRY_ENDPOINT);
 
-        Connection connection = factory.createConnection();
+        connection = factory.createConnection();
         connection.setExceptionListener(new ExceptionListener() {
 
             @Override
@@ -105,7 +117,6 @@ public void onException(JMSException exception) {
         MessageProducer messageProducer = session.createProducer(telemetryAddress);
         messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
 
-        long start = System.currentTimeMillis();
         for (int i = 1; i <= count; i++) {
             BytesMessage message = createTelemetryMessage(session, i);
             messageProducer.send(message);
@@ -114,12 +125,7 @@ public void onException(JMSException exception) {
                 LOG.debug("Sent message {}", i);
             }
         }
-
-        long finish = System.currentTimeMillis();
-        long taken = finish - start;
-        LOG.info("Sent {} messages in {}ms", count, taken);
-
-        connection.close();
+        messagesReceived.awaitSuccess(2000);
     }
 
     private static Context createInitialContext(final HonoServer server) throws NamingException {

tests/src/test/java/org/eclipse/hono/tests/HonoTestSupport.java

@@ -13,6 +13,7 @@
 
 import org.eclipse.hono.Application;
 import org.eclipse.hono.client.TelemetryClient;
+import org.junit.After;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -54,13 +55,23 @@ public class HonoTestSupport {
     @Value(value = "${hono.telemetry.downstream.port}")
     private int                 downstreamPort;
 
+    @After
+    public void closeClients() {
+        if (receiver != null) {
+            receiver.shutdown();
+        }
+        if (sender != null) {
+            sender.shutdown();
+        }
+    }
+
     @Test
     public void testTelemetry() throws Exception {
         final CountDownLatch received = new CountDownLatch(MSG_COUNT);
         receiver = new TelemetryClient(downstreamHostName, downstreamPort, TENANT_ID);
 
         receiver.createReceiver(message -> {
-            LOGGER.info("Received " + message);
+            LOGGER.debug("Received message: {}", message);
             received.countDown();
         }, "telemetry" + pathSeparator + "%s").setHandler(r -> createSender());