package org.loklak.stream;

import net.sf.xenqtt.client.AsyncClientListener;
import net.sf.xenqtt.client.AsyncMqttClient;
import net.sf.xenqtt.client.MqttClient;
import net.sf.xenqtt.client.PublishMessage;
import net.sf.xenqtt.client.Subscription;
import net.sf.xenqtt.message.ConnectReturnCode;
import net.sf.xenqtt.message.QoS;
import org.loklak.data.DAO;

public class MQTTPublisher {

    private AsyncMqttClient mqttClient;

    public MQTTPublisher(String address) {
        AsyncClientListener listener = new AsyncClientListener() {

            @Override
            public void publishReceived(MqttClient client, PublishMessage message) {
                DAO.severe("Received a message when no subscriptions were active. Please check your broker.");
            }

            @Override
            public void disconnected(MqttClient client, Throwable cause, boolean reconnecting) {
                if (cause != null) {
                    DAO.severe("Disconnected from the broker due to an exception.", cause);
                } else {
                    DAO.log("Disconnected from the broker.");
                }
                if (reconnecting) {
                    DAO.log("Attempting to reconnect to the broker.");
                }
            }

            @Override
            public void connected(MqttClient client, ConnectReturnCode returnCode) {
                DAO.log("Connected to client " + client + " with return code " + returnCode);
            }

            @Override
            public void subscribed(MqttClient client, Subscription[] requestedSubscriptions, Subscription[] grantedSubscriptions, boolean requestsGranted) {
                DAO.log("Subscribed to " + grantedSubscriptions.length + "/" + requestedSubscriptions.length + " subscriptions.");
            }

            @Override
            public void unsubscribed(MqttClient client, String[] topics) {
                DAO.log("Unsubscribed from " + topics.length + " topics.");
            }

            @Override
            public void published(MqttClient client, PublishMessage message) {
            }

        };

        this.mqttClient = new AsyncMqttClient(address, listener, 5);
        this.mqttClient.connect("loklak_server", false);
    }

    public void publish(String channel, String message) {
        this.mqttClient.publish(new PublishMessage(channel, QoS.AT_LEAST_ONCE, message));
    }

    public void publish(String channel, String messages[]) {
        for (String message : messages) {
            this.publish(channel, message);
        }
    }
}

@@ -2443,32 +2443,34 @@
  channel, String
+%5B%5D
  messages%5B%5D) %7B%0A 
@@ -2462,18 +2462,16 @@
 messages
-%5B%5D
 ) %7B%0A    
@@ -2495,32 +2495,363 @@
 essage : message
+s) %7B%0A            this.publish(channel, message);%0A        %7D%0A    %7D%0A%0A    public void publish(String%5B%5D channels, String message) %7B%0A        for (String channel : channels) %7B%0A            this.publish(channel, message);%0A        %7D%0A    %7D%0A%0A    public void publish(String%5B%5D channels, String%5B%5D messages) %7B%0A        for (String channel : channel
 s) %7B%0A           
@@ -2872,32 +2872,33 @@
 channel, message
+s
 );%0A        %7D%0A   

Fix #1406: Add methods to publish multiple message on multiple channels (#1407)