/**
 * Copyright (c) 2016, 2018 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.example;

import javax.annotation.PostConstruct;

import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Section;
import org.apache.qpid.proton.message.Message;
import org.eclipse.hono.client.HonoClient;
import org.eclipse.hono.client.MessageConsumer;
import org.eclipse.hono.util.MessageHelper;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Component;

import io.vertx.core.Future;
import io.vertx.proton.ProtonConnection;

/**
 * Example of a event/telemetry receiver that connects to the Hono Server, waits for incoming messages and logs the message
 * payload if anything is received.
 */
@Component
@Profile("receiver")
public class ExampleReceiver extends AbstractExampleClient {

    private static final String PROFILE_TELEMETRY = "telemetry";
    private static final String PROFILE_EVENT = "event";

    @PostConstruct
    private void start() {

        client.connect(getClientOptions(), this::onDisconnect)
            .compose(connectedClient -> createConsumer(connectedClient))
            .setHandler(startup -> {
                if (startup.succeeded()) {
                    String consumerType = activeProfiles.contains(PROFILE_EVENT) ? PROFILE_EVENT : PROFILE_TELEMETRY;
                    LOG.info("Receiver [tenant: {}, type: {}] created successfully, hit ctrl-c to exit", tenantId, consumerType);
                } else {
                    LOG.error("Error occurred during initialization of receiver: {}", startup.cause().getMessage());
                    vertx.close();
                }
            });

    }

    private Future<MessageConsumer> createConsumer(final HonoClient connectedClient) {

        if (activeProfiles.contains(PROFILE_EVENT)) {
            return connectedClient.createEventConsumer(tenantId,
                    (msg) -> handleMessage(PROFILE_EVENT, msg), closeHandler -> {
                        LOG.info("close handler of event consumer is called");
                        vertx.setTimer(DEFAULT_CONNECT_TIMEOUT_MILLIS, reconnect -> {
                            LOG.info("attempting to re-open the EventConsumer link ...");
                            createConsumer(connectedClient);
                        });
                    });
        } else {
            // default is telemetry consumer
            return connectedClient.createTelemetryConsumer(tenantId,
                    msg -> handleMessage(PROFILE_TELEMETRY, msg), closeHandler -> {
                        LOG.info("close handler of event consumer is called");
                        vertx.setTimer(DEFAULT_CONNECT_TIMEOUT_MILLIS, reconnect -> {
                            LOG.info("attempting to re-open the EventConsumer link ...");
                            createConsumer(connectedClient);
                        });
                    });
        }
    }

    private void onDisconnect(final ProtonConnection con) {

        // give Vert.x some time to clean up NetClient
        vertx.setTimer(DEFAULT_CONNECT_TIMEOUT_MILLIS, reconnect -> {
            LOG.info("attempting to re-connect to Hono ...");
            client.connect(getClientOptions(), this::onDisconnect).compose(connectedClient -> createConsumer(connectedClient));
        });
    }

    private void handleMessage(final String endpoint, final Message msg) {
        final String deviceId = MessageHelper.getDeviceId(msg);
        final Section body = msg.getBody();
        String content = null;
        if (body instanceof Data) {
            content = ((Data) msg.getBody()).getValue().toString();
        } else if (body instanceof AmqpValue) {
            content = ((AmqpValue) msg.getBody()).getValue().toString();
        }

        LOG.info("received {} message [device: {}, content-type: {}]: {}", endpoint, deviceId, msg.getContentType(), content);

        if (msg.getApplicationProperties() != null) {
            LOG.info("... with application properties: {}", msg.getApplicationProperties().getValue());
        }
    }
}

@@ -462,24 +462,54 @@
 Construct;%0A%0A
+import io.vertx.core.Handler;%0A
 import org.a
@@ -2217,25 +2217,24 @@
 ient) %7B%0A
-%0A
         
 if (acti
@@ -2229,209 +2229,52 @@
     
-if (activeProfiles.contains(PROFILE_EVENT)) %7B%0A            return connectedClient.createEventConsumer(tenantId,%0A                    (msg) -%3E handleMessage(PROFILE_EVENT, msg), closeHandler -%3E %7B%0A        
+Handler%3CVoid%3E closeHandler = closeHook -%3E %7B%0A
     
@@ -2273,36 +2273,32 @@
 %3E %7B%0A            
-    
 LOG.info(%22close 
@@ -2300,38 +2300,32 @@
 lose handler of 
-event 
 consumer is call
@@ -2326,36 +2326,24 @@
 s called%22);%0A
-            
             
@@ -2416,36 +2416,24 @@
             
-  
-          
 LOG.info(%22at
@@ -2448,38 +2448,33 @@
  to re-open the 
-EventC
+c
 onsumer link ...
@@ -2489,36 +2489,24 @@
             
-      
-      
 createConsum
@@ -2542,109 +2542,76 @@
     
-            %7D);%0A                    %7D);%0A        %7D else %7B%0A            // default is telemetry consumer
+%7D);%0A        %7D;%0A        if (activeProfiles.contains(PROFILE_EVENT)) %7B
 %0A   
@@ -2652,17 +2652,13 @@
 eate
-Telemetry
+Event
 Cons
@@ -2671,36 +2671,16 @@
 enantId,
-%0A                   
  msg -%3E 
@@ -2701,25 +2701,21 @@
 PROFILE_
-TELEMETRY
+EVENT
 , msg), 
@@ -2726,21 +2726,18 @@
 eHandler
- -%3E %7B
+);
 %0A       
@@ -2729,34 +2729,41 @@
 ndler);%0A        
-  
+%7D else %7B%0A
               LO
@@ -2762,351 +2762,168 @@
     
-  LOG.info(%22close handler of event consumer is called%22);%0A                        vertx.setTimer(DEFAULT_CONNECT_TIMEOUT_MILLIS, reconnect -%3E %7B%0A                            LOG.info(%22attempting to re-open the EventConsumer link ...%22);%0A                            createConsumer(connectedClient);%0A                        %7D);%0A                    %7D
+// default is telemetry consumer%0A            return connectedClient.createTelemetryConsumer(tenantId, msg -%3E handleMessage(PROFILE_TELEMETRY, msg), closeHandler
 );%0A 

Fix wrong log message and simplify ExampleReceiver#createConsumer.