/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.storm.mqtt;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.StormTopology;
import backtype.storm.testing.IntegrationTest;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.ITuple;
import org.apache.activemq.broker.BrokerService;
import org.apache.storm.mqtt.bolt.MqttBolt;
import org.apache.storm.mqtt.common.MqttOptions;
import org.apache.storm.mqtt.common.MqttPublisher;
import org.apache.storm.mqtt.mappers.StringMessageMapper;
import org.apache.storm.mqtt.spout.MqttSpout;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.net.URI;
import java.util.Arrays;

@Category(IntegrationTest.class)
public class StormMqttIntegrationTest implements Serializable{
    private static final Logger LOG = LoggerFactory.getLogger(StormMqttIntegrationTest.class);
    private static BrokerService broker;
    static boolean spoutActivated = false;

    public static class TestSpout extends MqttSpout{
        public TestSpout(MqttMessageMapper type, MqttOptions options){
            super(type, options);
        }

        @Override
        public void activate() {
            super.activate();
            LOG.info("Spout activated.");
            spoutActivated = true;
        }
    }


    @AfterClass
    public static void cleanup() throws Exception {
        broker.stop();
    }

    @BeforeClass
    public static void start() throws Exception {
        LOG.warn("Starting broker...");
        broker = new BrokerService();
        broker.addConnector("mqtt://localhost:1883");
        broker.setDataDirectory("target");
        broker.start();
        LOG.debug("MQTT broker started");
    }


    @Test
    public void testMqttTopology() throws Exception {
        MQTT client = new MQTT();
        client.setTracer(new MqttLogger());
        URI uri = URI.create("tcp://localhost:1883");
        client.setHost(uri);

        client.setClientId("MQTTSubscriber");
        client.setCleanSession(false);
        BlockingConnection connection = client.blockingConnection();
        connection.connect();
        Topic[] topics = {new Topic("/integration-result", QoS.AT_LEAST_ONCE)};
        byte[] qoses = connection.subscribe(topics);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("test", new Config(), buildMqttTopology());

        System.out.println("topology started");
        while(!spoutActivated) {
            Thread.sleep(500);
        }

        // publish a retained message to the broker
        MqttOptions options = new MqttOptions();
        options.setCleanConnection(false);
        MqttPublisher publisher = new MqttPublisher(options, true);
        publisher.connectMqtt("MqttPublisher");
        publisher.publish(new MqttMessage("/mqtt-topology", "test".getBytes()));

        LOG.info("published message");

        Message message = connection.receive();
        LOG.info("Message recieved on topic: " + message.getTopic());
        LOG.info("Payload: " + new String(message.getPayload()));
        message.ack();

        Assert.assertArrayEquals(message.getPayload(), "hello mqtt".getBytes());
        Assert.assertEquals(message.getTopic(), "/integration-result");
        cluster.shutdown();
    }

    public StormTopology buildMqttTopology(){
        TopologyBuilder builder = new TopologyBuilder();

        MqttOptions options = new MqttOptions();
        options.setTopics(Arrays.asList("/mqtt-topology"));
        options.setCleanConnection(false);
        TestSpout spout = new TestSpout(new StringMessageMapper(), options);

        MqttBolt bolt = new MqttBolt(options, new MqttTupleMapper() {
            @Override
            public MqttMessage toMessage(ITuple tuple) {
                LOG.info("Received: " + tuple);
                return new MqttMessage("/integration-result", "hello mqtt".getBytes());
            }
        });

        builder.setSpout("mqtt-spout", spout);
        builder.setBolt("mqtt-bolt", bolt).shuffleGrouping("mqtt-spout");

        return builder.createTopology();
    }

}

@@ -2154,16 +2154,219 @@
 false;%0A%0A
+    private static final String TEST_TOPIC = %22/mqtt-topology%22;%0A    private static final String RESULT_TOPIC = %22/integration-result%22;%0A    private static final String RESULT_PAYLOAD = %22Storm MQTT Spout%22;%0A%0A
     publ
@@ -4212,32 +4212,26 @@
 Message(
-%22/mqtt-topology%22
+TEST_TOPIC
 , %22test%22
@@ -4549,28 +4549,30 @@
 load(), 
-%22hello mqtt%22
+RESULT_PAYLOAD
 .getByte
@@ -4625,37 +4625,28 @@
 opic(), 
-%22/integration-result%22
+RESULT_TOPIC
 );%0A     
@@ -4868,32 +4868,26 @@
 .asList(
-%22/mqtt-topology%22
+TEST_TOPIC
 ));%0A    
@@ -5243,43 +5243,36 @@
 age(
-%22/integration-result%22, %22hello mqtt%22
+RESULT_TOPIC, RESULT_PAYLOAD
 .get

use constants in integration test instead of duplicate strings