/*******************************************************************************
 * Copyright (c) 2017 Red Hat Inc and others.
 *
 * 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:
 *     Red Hat Inc - initial API and implementation
 *******************************************************************************/
package org.eclipse.kapua.client.gateway.mqtt.paho.internal;

import java.util.concurrent.CompletableFuture;

import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;

public final class Listeners {

    private Listeners() {
    }

    public static IMqttActionListener toListener(final CompletableFuture<?> token) {
        return new IMqttActionListener() {

            @Override
            public void onSuccess(final IMqttToken asyncActionToken) {
                token.complete(null);
            }

            @Override
            public void onFailure(final IMqttToken asyncActionToken, final Throwable exception) {
                token.completeExceptionally(exception);
            }
        };
    }
}

@@ -631,16 +631,52 @@
 eFuture;
+%0Aimport java.util.function.Consumer;
 %0A%0Aimport
@@ -903,34 +903,59 @@
 nal 
-CompletableFuture%3C?%3E token
+Runnable success, final Consumer%3CThrowable%3E failure
 ) %7B%0A
@@ -1111,29 +1111,83 @@
     
-token.complete(null);
+if (success != null) %7B%0A                    success.run();%0A                %7D
 %0A   
@@ -1338,71 +1338,299 @@
     
-token.completeExceptionally(exception);%0A            %7D%0A        %7D
+if (failure != null) %7B%0A                    failure.accept(exception);%0A                %7D%0A            %7D%0A        %7D;%0A    %7D%0A%0A    public static IMqttActionListener toListener(final CompletableFuture%3C?%3E future) %7B%0A        return toListener(() -%3E future.complete(null), future::completeExceptionally)
 ;%0A  

Create listener based on functional interfaces