/*
*  Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*  WSO2 Inc. 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.wso2.carbon.event.output.adapter.mqtt;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.wso2.carbon.event.output.adapter.core.OutputEventAdapter;
import org.wso2.carbon.event.output.adapter.core.OutputEventAdapterConfiguration;
import org.wso2.carbon.event.output.adapter.core.exception.OutputEventAdapterException;
import org.wso2.carbon.event.output.adapter.core.exception.TestConnectionNotSupportedException;
import org.wso2.carbon.event.output.adapter.mqtt.internal.util.MQTTAdapterPublisher;
import org.wso2.carbon.event.output.adapter.mqtt.internal.util.MQTTBrokerConnectionConfiguration;
import org.wso2.carbon.event.output.adapter.mqtt.internal.util.MQTTEventAdapterConstants;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class MQTTEventAdapter implements OutputEventAdapter {

    private static final Log log = LogFactory.getLog(MQTTEventAdapter.class);
    private OutputEventAdapterConfiguration eventAdapterConfiguration;
    private Map<String, String> globalProperties;
    private ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<String, MQTTAdapterPublisher>>>
            publisherMap = new ConcurrentHashMap<String,
            ConcurrentHashMap<String, ConcurrentHashMap<String, MQTTAdapterPublisher>>>();

    public MQTTEventAdapter(OutputEventAdapterConfiguration eventAdapterConfiguration,
                            Map<String, String> globalProperties) {
        this.eventAdapterConfiguration = eventAdapterConfiguration;
        this.globalProperties = globalProperties;
    }


    @Override
    public void init() throws OutputEventAdapterException {
        //not required
    }

    @Override
    public void testConnect() throws TestConnectionNotSupportedException {
        throw new TestConnectionNotSupportedException("not-available");
    }

    @Override
    public void connect() {
        //not required
    }

    @Override
    public void publish(Object message, Map<String, String> dynamicProperties) {

        ConcurrentHashMap<String, ConcurrentHashMap<String, MQTTAdapterPublisher>> clientIdSpecificEventSenderMap =
                publisherMap.get(eventAdapterConfiguration.getName());
        if (null == clientIdSpecificEventSenderMap) {
            clientIdSpecificEventSenderMap =
                    new ConcurrentHashMap<String, ConcurrentHashMap<String, MQTTAdapterPublisher>>();
            if (null != publisherMap.putIfAbsent(eventAdapterConfiguration.getName(), clientIdSpecificEventSenderMap)) {
                clientIdSpecificEventSenderMap = publisherMap.get(eventAdapterConfiguration.getName());
            }
        }

        String clientId = dynamicProperties.get(MQTTEventAdapterConstants.ADAPTER_MESSAGE_CLIENTID);
        ConcurrentHashMap<String, MQTTAdapterPublisher> topicSpecificEventPublisherMap =
                clientIdSpecificEventSenderMap.get(clientId);
        if (null == topicSpecificEventPublisherMap) {
            topicSpecificEventPublisherMap = new ConcurrentHashMap<String, MQTTAdapterPublisher>();
            if (null != clientIdSpecificEventSenderMap.putIfAbsent(clientId, topicSpecificEventPublisherMap)) {
                topicSpecificEventPublisherMap = clientIdSpecificEventSenderMap.get(clientId);
            }
        }

        String topic = dynamicProperties.get(MQTTEventAdapterConstants.ADAPTER_MESSAGE_TOPIC);
        MQTTAdapterPublisher mqttAdaptorPublisher = topicSpecificEventPublisherMap.get(topic);
        if (mqttAdaptorPublisher == null) {
            MQTTBrokerConnectionConfiguration mqttBrokerConnectionConfiguration =
                    new MQTTBrokerConnectionConfiguration(eventAdapterConfiguration.getStaticProperties()
                            .get(MQTTEventAdapterConstants.ADAPTER_CONF_URL),
                            eventAdapterConfiguration.getStaticProperties()
                                    .get(MQTTEventAdapterConstants.ADAPTER_CONF_USERNAME),
                            eventAdapterConfiguration.getStaticProperties()
                                    .get(MQTTEventAdapterConstants.ADAPTER_CONF_PASSWORD),
                            eventAdapterConfiguration.getStaticProperties()
                                    .get(MQTTEventAdapterConstants.ADAPTER_CONF_CLEAN_SESSION),
                            eventAdapterConfiguration.getStaticProperties()
                                    .get(MQTTEventAdapterConstants.ADAPTER_CONF_KEEP_ALIVE));

            mqttAdaptorPublisher = new MQTTAdapterPublisher(mqttBrokerConnectionConfiguration,
                    dynamicProperties.get(MQTTEventAdapterConstants.ADAPTER_MESSAGE_TOPIC),
                    dynamicProperties.get(MQTTEventAdapterConstants.ADAPTER_MESSAGE_CLIENTID));
            topicSpecificEventPublisherMap.put(topic,mqttAdaptorPublisher);
        }
        String qos = eventAdapterConfiguration.getStaticProperties().get(MQTTEventAdapterConstants.ADAPTER_MESSAGE_QOS);

        if (qos == null) {
            mqttAdaptorPublisher.publish(message.toString());
        } else {
            mqttAdaptorPublisher.publish(Integer.parseInt(qos), message.toString());
        }

    }

    @Override
    public void disconnect() {
        //not required
    }

    @Override
    public void destroy() {
        //not required
    }
}

@@ -1993,24 +1993,144 @@
 sher%3E%3E%3E();%0A%0A
+    private ConcurrentHashMap%3CString, ConcurrentHashMap%3CString, MQTTAdapterPublisher%3E%3E clientIdSpecificEventSenderMap;%0A%0A
     public M
@@ -2723,209 +2723,8 @@
     
- //not required%0A    %7D%0A%0A    @Override%0A    public void publish(Object message, Map%3CString, String%3E dynamicProperties) %7B%0A%0A        ConcurrentHashMap%3CString, ConcurrentHashMap%3CString, MQTTAdapterPublisher%3E%3E
  cli
@@ -2752,32 +2752,16 @@
 derMap =
-%0A               
  publish
@@ -3250,32 +3250,133 @@
     %7D%0A        %7D%0A
+    %7D%0A%0A    @Override%0A    public void publish(Object message, Map%3CString, String%3E dynamicProperties) %7B
 %0A        String 
@@ -4123,33 +4123,33 @@
 lisher mqttAdapt
-o
+e
 rPublisher = top
@@ -4201,33 +4201,33 @@
    if (mqttAdapt
-o
+e
 rPublisher == nu
@@ -5188,33 +5188,33 @@
        mqttAdapt
-o
+e
 rPublisher = new
@@ -5511,16 +5511,17 @@
 t(topic,
+ 
 mqttAdap
@@ -5513,33 +5513,33 @@
 topic, mqttAdapt
-o
+e
 rPublisher);%0A   
@@ -5707,33 +5707,33 @@
        mqttAdapt
-o
+e
 rPublisher.publi
@@ -5794,17 +5794,17 @@
 qttAdapt
-o
+e
 rPublish
@@ -5925,38 +5925,45 @@
 ) %7B%0A        
-//not required
+publisherMap.clear();
 %0A    %7D%0A%0A    

Reconnect bug fixed