/**
 * 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;

import static org.eclipse.hono.TelemetryDataReadStream.DEVICE_BUMLUX_TEMP_4711;

import java.io.IOException;
import java.net.InetAddress;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.qpid.proton.message.Message;
import org.eclipse.hono.authorization.Permission;
import org.eclipse.hono.authorization.impl.InMemoryAuthorizationService;
import org.eclipse.hono.impl.ProtonSenderWriteStream;
import org.eclipse.hono.server.HonoServer;
import org.eclipse.hono.telemetry.TelemetryConstants;
import org.eclipse.hono.telemetry.impl.MessageDiscardingTelemetryAdapter;
import org.eclipse.hono.util.Constants;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.streams.Pump;
import io.vertx.core.streams.ReadStream;
import io.vertx.core.streams.WriteStream;
import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
import io.vertx.proton.ProtonConnection;
import io.vertx.proton.ProtonQoS;
import io.vertx.proton.ProtonSender;

/**
 * Integration tests for Hono Server.
 *
 */
@RunWith(VertxUnitRunner.class)
public class HonoServerTest {

    /**
     * 
     */
    private static final String TENANT_ID_BOSCH = "BOSCH";
    private static final Logger    LOG                     = LoggerFactory.getLogger(HonoServerTest.class);
    static Vertx                   vertx;
    static HonoServer           server;
    ProtonConnection               connection;
    ProtonSender                   protonSender;

    @BeforeClass
    public static void init(final TestContext ctx) {

        // Create the Vert.x instance
        vertx = Vertx.vertx();
        server = new HonoServer();
        server.setPort(0);
        server.setHost("0.0.0.0");

        vertx.deployVerticle(server, ctx.asyncAssertSuccess());
        vertx.deployVerticle(MessageDiscardingTelemetryAdapter.class.getName(), ctx.asyncAssertSuccess());

        final InMemoryAuthorizationService authorizationService = new InMemoryAuthorizationService();
        authorizationService.addPermission(Constants.DEFAULT_SUBJECT, "telemetry/" + TENANT_ID_BOSCH, Permission.WRITE);
        authorizationService.addPermission(Constants.DEFAULT_SUBJECT, "telemetry/" + DEVICE_BUMLUX_TEMP_4711, Permission.WRITE);
        vertx.deployVerticle(authorizationService, ctx.asyncAssertSuccess());

    }

    @Before
    public void setup(final TestContext ctx) throws Exception {

        Async connectionOpened = ctx.async();
        String host = InetAddress.getLoopbackAddress().getHostAddress();
        int port = server.getPort();
        // Create the Vert.x AMQP client instance
        TestSupport.connect(ctx, vertx, host, port, connectedCon -> {
            LOG.debug("client connected to server, now opening connection");
            connectedCon.openHandler(result -> {
                ctx.assertTrue(result.succeeded());
                ProtonConnection openedCon = result.result();
                LOG.debug("connection to [container: {}] open", openedCon.getRemoteContainer());
                connection = openedCon;
                connectionOpened.complete();
            }).open();
        });
    }

    @After
    public void disconnect(final TestContext ctx) {
        connection.close();
    }

    @AfterClass
    public static void shutdown(final TestContext ctx) throws IOException {
        vertx.close(ctx.asyncAssertSuccess(done -> LOG.info("Vertx has been shut down")));
    }

    @Test
    public void testTelemetryUpload(final TestContext ctx) {
        LOG.debug("starting telemetry upload test");
        final int messagesToBeSent = 100;
        final Async sent = ctx.async(messagesToBeSent);
        int timeout = 2000; // milliseconds
        final AtomicLong start = new AtomicLong();

        protonSender = connection.createSender(TelemetryConstants.NODE_ADDRESS_TELEMETRY_PREFIX + TENANT_ID_BOSCH);
        protonSender
                .setQoS(ProtonQoS.AT_MOST_ONCE)
                .openHandler(senderOpen -> {
                    ctx.assertTrue(senderOpen.succeeded());
                    LOG.debug("outbound link created, now starting to send messages");
                    start.set(System.currentTimeMillis());
                    uploadTelemetryData(messagesToBeSent, protonSender, allProduced -> {
                        LOG.debug("all {} telemetry messages have been produced", messagesToBeSent);
                    }, sent);
                }).open();

        sent.await(timeout);
        LOG.info("messages ack'd after {} milliseconds: {}", System.currentTimeMillis() - start.get(),
                messagesToBeSent - sent.count());
        protonSender.close();
    }

    private void uploadTelemetryData(final int count, final ProtonSender sender, final Handler<Void> allProducedHandler,
            final Async sentMessages) {
        vertx.executeBlocking(future -> {
            ReadStream<Message> rs = new TelemetryDataReadStream(vertx, count, TENANT_ID_BOSCH);
            WriteStream<Message> ws = new ProtonSenderWriteStream(sender, delivered -> sentMessages.countDown());
            rs.endHandler(done -> future.complete());
            LOG.debug("pumping test telemetry data to Hono server");
            Pump.pump(rs, ws).start();
        }, done -> allProducedHandler.handle(null));
    }
}

@@ -1827,16 +1827,19 @@
  String 
+MY_
 TENANT_I
@@ -1843,23 +1843,21 @@
 T_ID
-_BOSCH = %22BOSCH
+ = %22MY_TENANT
 %22;%0A 
@@ -1891,25 +1891,11 @@
 ger 
-   LOG           
+LOG
     
@@ -2725,16 +2725,19 @@
 try/%22 + 
+MY_
 TENANT_I
@@ -2737,22 +2737,16 @@
 ENANT_ID
-_BOSCH
 , Permis
@@ -4494,16 +4494,19 @@
 REFIX + 
+MY_
 TENANT_I
@@ -4498,38 +4498,32 @@
 X + MY_TENANT_ID
-_BOSCH
 );%0A        proto
@@ -5578,16 +5578,19 @@
  count, 
+MY_
 TENANT_I
@@ -5594,14 +5594,8 @@
 T_ID
-_BOSCH
 );%0A 

change default tenant in tests