/**
 * 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.telemetry.impl;

import java.util.Objects;

import org.eclipse.hono.telemetry.SenderFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.proton.ProtonConnection;
import io.vertx.proton.ProtonQoS;
import io.vertx.proton.ProtonSender;

/**
 * A default {@code SenderFactory} for creating {@code ProtonSender} from a given connection.
 */
@Component
public class SenderFactoryImpl implements SenderFactory {

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

    @Override
    public void createSender(
            final ProtonConnection connection,
            final String address,
            final Handler<ProtonSender> sendQueueDrainHandler,
            final Future<ProtonSender> result) {

        Objects.requireNonNull(connection);
        Objects.requireNonNull(address);
        Objects.requireNonNull(result);

        ProtonSender sender = connection.createSender(address);
        sender.setQoS(ProtonQoS.AT_MOST_ONCE);
        sender.sendQueueDrainHandler(sendQueueDrainHandler);
        sender.openHandler(openAttempt -> {
            if (openAttempt.succeeded()) {
                LOG.debug(
                        "sender [{}] for downstream container [{}] open",
                        address,
                        connection.getRemoteContainer());
                result.complete(openAttempt.result());
            } else {
                LOG.warn("could not open sender for downstream container [{}]",
                        connection.getRemoteContainer());
                result.fail(openAttempt.cause());
            }
        });
        sender.closeHandler(closed -> {
            if (closed.succeeded()) {
                LOG.debug(
                        "sender [{}] for downstream container [{}] closed",
                        address,
                        connection.getRemoteContainer());
            }
        });
        sender.open();
    }

}

@@ -608,16 +608,50 @@
 onent;%0A%0A
+import io.vertx.core.AsyncResult;%0A
 import i
@@ -671,16 +671,16 @@
 Future;%0A
-
 import i
@@ -812,16 +812,54 @@
 nSender;
+%0Aimport io.vertx.proton.ProtonSession;
 %0A%0A/**%0A *
@@ -1489,37 +1489,958 @@
     
-ProtonSe
+if (connection.isDisconnected()) %7B%0A            result.fail(%22no connection to downstream container%22);%0A        %7D else %7B%0A            newSession(connection, remoteOpen -%3E %7B%0A                if (remoteOpen.succeeded()) %7B%0A                    newSender(connection, remoteOpen.result(), address, sendQueueDrainHandler, result);%0A                %7D else %7B%0A                    result.fail(remoteOpen.cause());%0A                %7D%0A            %7D);%0A        %7D%0A    %7D%0A%0A    private void newSession(final ProtonConnection con, final Handler%3CAsyncResult%3CProtonSession%3E%3E sessionOpenHa
 nd
+l
 er
- sender = connect
+) %7B%0A        con.createSession().openHandler(sessionOpenHandler).open();%0A    %7D%0A%0A    private void newSender(%0A            final ProtonConnection downstreamConnection,%0A            final ProtonSession session,%0A            final String address,%0A            final Handler%3CProtonSender%3E sendQueueDrainHandler,%0A            final Future%3CProtonSender%3E result) %7B%0A%0A        ProtonSender sender = sess
 ion.
@@ -2807,33 +2807,43 @@
                 
-c
+downstreamC
 onnection.getRem
@@ -2959,12 +2959,13 @@
 LOG.
-warn
+debug
 (%22co
@@ -3032,33 +3032,43 @@
                 
-c
+downstreamC
 onnection.getRem
@@ -3073,32 +3073,53 @@
 emoteContainer()
+, openAttempt.cause()
 );%0A             
@@ -3419,17 +3419,27 @@
         
-c
+downstreamC
 onnectio
@@ -3512,17 +3512,16 @@
 open();%0A
+
     %7D%0A
-%0A
 %7D%0A

Use separate session for each downstream sender.