/**
 * Copyright (c) 2016 Bosch Software Innovations GmbH.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Bosch Software Innovations GmbH - initial creation
 *
 */

package org.eclipse.hono.tests.jms;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
public class CorrelationHelper<T, R> {

    private static final Logger LOGGER = LoggerFactory.getLogger(CorrelationHelper.class);

    private final Map<String, Consumer<T>>  handlers = new HashMap<>();

    public CompletableFuture<R> add(final String correlationId, final Function<T, R> consumer) {

        final CompletableFuture<R> future = new CompletableFuture<>();

        final Consumer<T> wrappedConsumer =  t -> {
            try {
                future.complete(consumer.apply(t));
            } catch (final Exception e) {
                future.completeExceptionally(e);
            }
        };

        handlers.put(correlationId, wrappedConsumer);

        return future;
    }

    public void handle(final String correlationId, final T message) {
        final Consumer<T> consumer = handlers.remove(correlationId);
        if (consumer != null ){
            consumer.accept(message);
        } else {
            LOGGER.debug("No pending request found for {}.", correlationId);
        }
    }
}

@@ -427,34 +427,8 @@
 s;%0A%0A
-import java.util.HashMap;%0A
 impo
@@ -445,16 +445,16 @@
 il.Map;%0A
+
 import j
@@ -492,16 +492,63 @@
 Future;%0A
+import java.util.concurrent.ConcurrentHashMap;%0A
 import j
@@ -676,16 +676,151 @@
 %0A%0A/**%0A *
+ @param T the type of response this helper keeps consumers for%0A * @param R the type of result the consumers held by this helper provide
 %0A */%0Apub
@@ -944,24 +944,24 @@
 er.class);%0A%0A
-
     private 
@@ -1007,16 +1007,26 @@
 s = new 
+Concurrent
 HashMap%3C
@@ -1843,11 +1843,74 @@
  %7D%0A    %7D
+%0A%0A    public int size() %7B%0A        return handlers.size();%0A    %7D
 %0A%7D%0A

Add getter for size of message handler map.