/*******************************************************************************
 * Copyright (c) 2011, 2017 Eurotech and/or its affiliates and others
 *
 * 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:
 *     Eurotech - initial API and implementation
 *     Red Hat Inc
 *******************************************************************************/
package org.eclipse.kapua.broker.core.converter;

import com.codahale.metrics.Counter;
import org.apache.camel.Converter;
import org.apache.camel.Exchange;
import org.apache.camel.component.jms.JmsMessage;
import org.apache.commons.lang3.SerializationUtils;
import org.eclipse.kapua.KapuaException;
import org.eclipse.kapua.broker.core.listener.CamelConstants;
import org.eclipse.kapua.broker.core.message.CamelKapuaMessage;
import org.eclipse.kapua.broker.core.message.CamelUtil;
import org.eclipse.kapua.broker.core.message.JmsUtil;
import org.eclipse.kapua.broker.core.message.MessageConstants;
import org.eclipse.kapua.broker.core.plugin.ConnectorDescriptor;
import org.eclipse.kapua.broker.core.plugin.ConnectorDescriptor.MessageType;
import org.eclipse.kapua.commons.metric.MetricServiceFactory;
import org.eclipse.kapua.commons.metric.MetricsService;
import org.eclipse.kapua.model.id.KapuaId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.Message;
import java.util.Base64;
import java.util.Date;

/**
 * Kapua message converter reference implementation used to convert from Camel incoming messages ({@link JmsMessage}) to a platform specific message type.
 *
 * @since 1.0
 */
public abstract class AbstractKapuaConverter {

    public static final Logger logger = LoggerFactory.getLogger(AbstractKapuaConverter.class);

    // metrics
    protected static final String METRIC_COMPONENT_NAME = "converter";
    protected static final MetricsService METRICS_SERVICE = MetricServiceFactory.getInstance();

    private final Counter metricConverterJmsMessage;
    private final Counter metricConverterJmsErrorMessage;
    private final Counter metricConverterErrorMessage;

    /**
     * Constructor
     */
    protected AbstractKapuaConverter() {
        metricConverterJmsMessage = METRICS_SERVICE.getCounter(METRIC_COMPONENT_NAME, "kapua", "jms", "messages", "count");
        metricConverterJmsErrorMessage = METRICS_SERVICE.getCounter(METRIC_COMPONENT_NAME, "kapua", "jms", "messages", "error", "count");
        metricConverterErrorMessage = METRICS_SERVICE.getCounter(METRIC_COMPONENT_NAME, "kapua", "kapua_message", "messages", "error", "count");
    }

    /**
     * Convert incoming message to a Kapua message (depending on messageType parameter)
     *
     * @param exchange
     * @param value
     * @param messageType expected incoming message type
     * @return Message container that contains message of asked type
     * @throws KapuaException if incoming message does not contain a javax.jms.BytesMessage or an error during conversion occurred
     */
    protected CamelKapuaMessage<?> convertTo(Exchange exchange, Object value, MessageType messageType) throws KapuaException {
        // assume that the message is a Camel Jms message
        JmsMessage message = exchange.getIn(JmsMessage.class);
        if (message.getJmsMessage() instanceof BytesMessage) {
            try {
                // FIX #164
                Date queuedOn = new Date(message.getHeader(CamelConstants.JMS_HEADER_TIMESTAMP, Long.class));
                KapuaId connectionId = (KapuaId) SerializationUtils.deserialize(Base64.getDecoder().decode(message.getHeader(MessageConstants.HEADER_KAPUA_CONNECTION_ID, String.class)));
                String clientId = message.getHeader(MessageConstants.HEADER_KAPUA_CLIENT_ID, String.class);
                ConnectorDescriptor connectorDescriptor = (ConnectorDescriptor) SerializationUtils
                        .deserialize(Base64.getDecoder().decode(message.getHeader(MessageConstants.HEADER_KAPUA_CONNECTOR_DEVICE_PROTOCOL, String.class)));
                return JmsUtil.convertToCamelKapuaMessage(connectorDescriptor, messageType, (byte[]) value, CamelUtil.getTopic(message), queuedOn, connectionId, clientId);
            } catch (JMSException e) {
                metricConverterErrorMessage.inc();
                logger.error("Exception converting message {}", e.getMessage(), e);
                throw KapuaException.internalError(e, "Cannot convert the message type " + exchange.getIn().getClass());
            }
        }
        metricConverterErrorMessage.inc();
        throw KapuaException.internalError("Cannot convert the message - Wrong instance type: " + exchange.getIn().getClass());
    }

    /**
     * Convert incoming message to a javax.jms.Message
     *
     * @param exchange
     * @param value
     * @return jms Message
     * @throws KapuaException if incoming message does not contain a javax.jms.BytesMessage
     */
    @Converter
    public Message convertToJmsMessage(Exchange exchange, Object value) throws KapuaException {
        metricConverterJmsMessage.inc();
        // assume that the message is a Camel Jms message
        JmsMessage message = exchange.getIn(JmsMessage.class);
        if (message.getJmsMessage() instanceof BytesMessage) {
            return message.getJmsMessage();
        }
        metricConverterJmsErrorMessage.inc();
        throw KapuaException.internalError("Cannot convert the message - Wrong instance type: " + exchange.getIn().getClass());
    }

}

@@ -761,32 +761,77 @@
 jms.JmsMessage;%0A
+import org.apache.camel.impl.DefaultMessage;%0A
 import org.apach
@@ -3418,184 +3418,233 @@
     
-// assume that the message is a Camel Jms message%0A        JmsMessage message = exchange.getIn(JmsMessage.class);%0A        if (message.getJmsMessage() instanceof BytesMessage) %7B%0A
+if (value instanceof byte%5B%5D) %7B%0A            byte%5B%5D messageContent = (byte%5B%5D) value;%0A            if (exchange.getIn() instanceof DefaultMessage) %7B%0A                DefaultMessage message = (DefaultMessage) exchange.getIn();%0A    
     
@@ -3673,16 +3673,20 @@
         
+    
 // FIX #
@@ -3689,16 +3689,20 @@
 IX #164%0A
+    
         
@@ -3799,24 +3799,28 @@
 ng.class));%0A
+    
             
@@ -4010,16 +4010,20 @@
         
+ 
+   
 String c
@@ -4122,16 +4122,20 @@
         
+    
 Connecto
@@ -4233,16 +4233,20 @@
         
+  
+  
 .deseria
@@ -4381,24 +4381,28 @@
             
+    
 return JmsUt
@@ -4465,30 +4465,30 @@
 geType, 
-(byte%5B%5D) value
+messageContent
 , CamelU
@@ -4557,16 +4557,20 @@
         
+    
 %7D catch 
@@ -4584,24 +4584,28 @@
 eption e) %7B%0A
+    
             
@@ -4659,16 +4659,20 @@
         
+    
 logger.e
@@ -4731,16 +4731,20 @@
 (), e);%0A
+    
         
@@ -4852,24 +4852,42 @@
 etClass());%0A
+                %7D%0A
             

fix kapua message converter