package io.joynr.messaging.mqtt;

/*
 * #%L
 * %%
 * Copyright (C) 2011 - 2016 BMW Car IT GmbH
 * %%
 * Licensed 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.
 * #L%
 */

import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Maps;
import com.google.inject.Inject;
import com.google.inject.name.Named;

import io.joynr.exceptions.JoynrMessageNotSentException;
import io.joynr.exceptions.JoynrSendBufferFullException;
import io.joynr.exceptions.JoynrSerializationException;
import io.joynr.messaging.FailureAction;
import io.joynr.messaging.IMessagingMulticastSubscriber;
import io.joynr.messaging.IMessagingSkeleton;
import io.joynr.messaging.JoynrMessageSerializer;
import io.joynr.messaging.routing.MessageRouter;
import joynr.JoynrMessage;
import joynr.system.RoutingTypes.MqttAddress;
import joynr.system.RoutingTypes.RoutingTypesUtil;

/**
 * Connects to the MQTT broker
 */
public class MqttMessagingSkeleton implements IMessagingSkeleton, IMessagingMulticastSubscriber {

    private static final Logger LOG = LoggerFactory.getLogger(MqttMessagingSkeleton.class);
    private MessageRouter messageRouter;
    private JoynrMqttClient mqttClient;
    private JoynrMessageSerializer messageSerializer;
    private MqttClientFactory mqttClientFactory;
    private MqttAddress ownAddress;
    private ConcurrentMap<String, AtomicInteger> multicastSubscriptionCount = Maps.newConcurrentMap();

    @Inject
    public MqttMessagingSkeleton(@Named(MqttModule.PROPERTY_MQTT_ADDRESS) MqttAddress ownAddress,
                                 MessageRouter messageRouter,
                                 MqttClientFactory mqttClientFactory,
                                 MqttMessageSerializerFactory messageSerializerFactory) {
        this.ownAddress = ownAddress;
        this.messageRouter = messageRouter;
        this.mqttClientFactory = mqttClientFactory;
        messageSerializer = messageSerializerFactory.create(ownAddress);
    }

    @Override
    public void init() {
        mqttClient = mqttClientFactory.create();
        mqttClient.setMessageListener(this);
        mqttClient.start();
        subscribe();
    }

    /**
     * Performs standard subscription to the {@link #ownAddress own address'} topic; override this method to perform
     * custom subscriptions. One use-case could be to subscribe to one topic for incoming messages and another topic for
     * replies.
     */
    protected void subscribe() {
        mqttClient.subscribe(ownAddress.getTopic() + "/#");
    }

    @Override
    public void shutdown() {
        mqttClient.shutdown();
    }

    @Override
    public void registerMulticastSubscription(String multicastId) {
        multicastSubscriptionCount.putIfAbsent(multicastId, new AtomicInteger());
        int numberOfSubscriptions = multicastSubscriptionCount.get(multicastId).incrementAndGet();
        if (numberOfSubscriptions == 1) {
            mqttClient.subscribe(translateWildcard(multicastId));
        }
    }

    @Override
    public void unregisterMulticastSubscription(String multicastId) {
        AtomicInteger subscribersCount = multicastSubscriptionCount.get(multicastId);
        if (subscribersCount != null) {
            int remainingCount = subscribersCount.decrementAndGet();
            if (remainingCount == 0) {
                mqttClient.unsubscribe(translateWildcard(multicastId));
            }
        }
    }

    private String translateWildcard(String multicastId) {
        String topic = multicastId;
        if (topic.endsWith("/*")) {
            topic = topic.replaceFirst("/\\*$", "/#");
        }
        return topic;
    }

    @Override
    public void transmit(JoynrMessage message, FailureAction failureAction) {
        LOG.debug("<<< INCOMING <<< {}", message.toLogMessage());
        try {
            if (JoynrMessage.MESSAGE_TYPE_MULTICAST.equals(message.getType())) {
                message.setReceivedFromGlobal(true);
            }
            String replyToMqttAddress = message.getHeaderValue(JoynrMessage.HEADER_NAME_REPLY_CHANNELID);
            if (replyToMqttAddress != null && !replyToMqttAddress.isEmpty()) {
                messageRouter.addNextHop(message.getFrom(), RoutingTypesUtil.fromAddressString(replyToMqttAddress));
            }
            messageRouter.route(message);
        } catch (JoynrSendBufferFullException | JoynrMessageNotSentException e) {
            failureAction.execute(e);
        }
    }

    @Override
    public void transmit(String serializedMessage, FailureAction failureAction) {
        try {
            JoynrMessage message = messageSerializer.deserialize(serializedMessage);
            transmit(message, failureAction);
        } catch (JoynrSerializationException e) {
            LOG.error("Message: \"{}\", could not be serialized, exception: {}", serializedMessage, e.getMessage());
        }
    }

    protected JoynrMqttClient getClient() {
        return mqttClient;
    }

    protected MqttAddress getOwnAddress() {
        return ownAddress;
    }

}

@@ -74,9 +74,9 @@
  201
-6
+7
  BMW
@@ -930,122 +930,8 @@
 d;%0A%0A
-import io.joynr.exceptions.JoynrMessageNotSentException;%0Aimport io.joynr.exceptions.JoynrSendBufferFullException;%0A
 impo
@@ -4823,72 +4823,136 @@
 ch (
-JoynrSendBufferFullException %7C JoynrMessageNotSentException e) %7B
+Exception e) %7B%0A            LOG.error(%22Error processing incoming message. Message will be dropped: %7B%7D %22, message.getHeader(), e);
 %0A   

fix error handling in MqttMessagingSkeleton.transmit