/*
 * 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.camel.component.jms;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.Serializable;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageFormatException;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.StreamMessage;
import javax.jms.TextMessage;

import org.w3c.dom.Node;

import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.StreamCache;
import org.apache.camel.WrappedFile;
import org.apache.camel.converter.stream.CachedOutputStream;
import org.apache.camel.spi.HeaderFilterStrategy;
import org.apache.camel.support.CamelContextHelper;
import org.apache.camel.support.DefaultExchangeHolder;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.ObjectHelper;
import org.apache.camel.support.PatternHelper;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.IOHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.camel.component.jms.JmsConstants.JMS_X_GROUP_ID;
import static org.apache.camel.component.jms.JmsMessageHelper.getSafeLongProperty;
import static org.apache.camel.component.jms.JmsMessageHelper.isVendor;
import static org.apache.camel.component.jms.JmsMessageHelper.normalizeDestinationName;
import static org.apache.camel.component.jms.JmsMessageType.Bytes;
import static org.apache.camel.component.jms.JmsMessageType.Map;
import static org.apache.camel.component.jms.JmsMessageType.Object;
import static org.apache.camel.component.jms.JmsMessageType.Stream;
import static org.apache.camel.component.jms.JmsMessageType.Text;

/**
 * A Strategy used to convert between a Camel {@link Exchange} and {@link JmsMessage}
 * to and from a JMS {@link Message}
 */
public class JmsBinding {
    private static final Logger LOG = LoggerFactory.getLogger(JmsBinding.class);
    private final JmsEndpoint endpoint;
    private final HeaderFilterStrategy headerFilterStrategy;
    private final JmsKeyFormatStrategy jmsKeyFormatStrategy;
    private final MessageCreatedStrategy messageCreatedStrategy;

    public JmsBinding() {
        this.endpoint = null;
        this.headerFilterStrategy = new JmsHeaderFilterStrategy(false);
        this.jmsKeyFormatStrategy = new DefaultJmsKeyFormatStrategy();
        this.messageCreatedStrategy = null;
    }

    public JmsBinding(JmsEndpoint endpoint) {
        this.endpoint = endpoint;
        if (endpoint.getHeaderFilterStrategy() != null) {
            this.headerFilterStrategy = endpoint.getHeaderFilterStrategy();
        } else {
            this.headerFilterStrategy = new JmsHeaderFilterStrategy(endpoint.isIncludeAllJMSXProperties());
        }
        if (endpoint.getJmsKeyFormatStrategy() != null) {
            this.jmsKeyFormatStrategy = endpoint.getJmsKeyFormatStrategy();
        } else {
            this.jmsKeyFormatStrategy = new DefaultJmsKeyFormatStrategy();
        }
        if (endpoint.getMessageCreatedStrategy() != null) {
            this.messageCreatedStrategy = endpoint.getMessageCreatedStrategy();
        } else if (endpoint.getComponent() != null) {
            // fallback and use from component
            this.messageCreatedStrategy = endpoint.getComponent().getMessageCreatedStrategy();
        } else {
            this.messageCreatedStrategy = null;
        }
    }

    /**
     * Extracts the body from the JMS message
     *
     * @param exchange the exchange
     * @param message  the message to extract its body
     * @return the body, can be <tt>null</tt>
     */
    public Object extractBodyFromJms(Exchange exchange, Message message) {
        try {
            // is a custom message converter configured on endpoint then use it instead of doing the extraction
            // based on message type
            if (endpoint != null && endpoint.getMessageConverter() != null) {
                if (LOG.isTraceEnabled()) {
                    LOG.trace("Extracting body using a custom MessageConverter: {} from JMS message: {}", endpoint.getMessageConverter(), message);
                }
                return endpoint.getMessageConverter().fromMessage(message);
            }

            // if we are configured to not map the jms message then return it as body
            if (endpoint != null && !endpoint.getConfiguration().isMapJmsMessage()) {
                LOG.trace("Option map JMS message is false so using JMS message as body: {}", message);
                return message;
            }

            if (message instanceof ObjectMessage) {
                LOG.trace("Extracting body as a ObjectMessage from JMS message: {}", message);
                ObjectMessage objectMessage = (ObjectMessage)message;
                Object payload = objectMessage.getObject();
                if (payload instanceof DefaultExchangeHolder) {
                    DefaultExchangeHolder holder = (DefaultExchangeHolder) payload;
                    DefaultExchangeHolder.unmarshal(exchange, holder);
                    // enrich with JMS headers also as otherwise they will get lost when use the transferExchange option.
                    Map<String, Object> jmsHeaders = extractHeadersFromJms(message, exchange);
                    exchange.getIn().getHeaders().putAll(jmsHeaders);
                    return exchange.getIn().getBody();
                } else {
                    return objectMessage.getObject();
                }
            } else if (message instanceof TextMessage) {
                LOG.trace("Extracting body as a TextMessage from JMS message: {}", message);
                TextMessage textMessage = (TextMessage)message;
                return textMessage.getText();
            } else if (message instanceof MapMessage) {
                LOG.trace("Extracting body as a MapMessage from JMS message: {}", message);
                return createMapFromMapMessage((MapMessage)message);
            } else if (message instanceof BytesMessage) {
                LOG.trace("Extracting body as a BytesMessage from JMS message: {}", message);
                return createByteArrayFromBytesMessage(exchange, (BytesMessage)message);
            } else if (message instanceof StreamMessage) {
                LOG.trace("Extracting body as a StreamMessage from JMS message: {}", message);
                StreamMessage streamMessage = (StreamMessage)message;
                return createInputStreamFromStreamMessage(exchange, streamMessage);
            } else {
                return null;
            }
        } catch (JMSException e) {
            throw new RuntimeCamelException("Failed to extract body due to: " + e + ". Message: " + message, e);
        }
    }

    public Map<String, Object> extractHeadersFromJms(Message jmsMessage, Exchange exchange) {
        Map<String, Object> map = new HashMap<>();
        if (jmsMessage != null) {
            // lets populate the standard JMS message headers
            try {
                map.put("JMSCorrelationID", jmsMessage.getJMSCorrelationID());
                map.put("JMSCorrelationIDAsBytes", JmsMessageHelper.getJMSCorrelationIDAsBytes(jmsMessage));
                map.put("JMSDeliveryMode", jmsMessage.getJMSDeliveryMode());
                map.put("JMSDestination", jmsMessage.getJMSDestination());
                map.put("JMSExpiration", jmsMessage.getJMSExpiration());
                map.put("JMSMessageID", jmsMessage.getJMSMessageID());
                map.put("JMSPriority", jmsMessage.getJMSPriority());
                map.put("JMSRedelivered", jmsMessage.getJMSRedelivered());
                map.put("JMSTimestamp", jmsMessage.getJMSTimestamp());

                map.put("JMSReplyTo", JmsMessageHelper.getJMSReplyTo(jmsMessage));
                map.put("JMSType", JmsMessageHelper.getJMSType(jmsMessage));

                // this works around a bug in the ActiveMQ property handling
                map.put(JMS_X_GROUP_ID, JmsMessageHelper.getStringProperty(jmsMessage, JMS_X_GROUP_ID));
                map.put("JMSXUserID", JmsMessageHelper.getStringProperty(jmsMessage, "JMSXUserID"));
            } catch (JMSException e) {
                throw new RuntimeCamelException(e);
            }

            Enumeration<?> names;
            try {
                names = jmsMessage.getPropertyNames();
            } catch (JMSException e) {
                throw new RuntimeCamelException(e);
            }
            while (names.hasMoreElements()) {
                String name = names.nextElement().toString();
                try {
                    Object value = JmsMessageHelper.getProperty(jmsMessage, name);
                    if (headerFilterStrategy != null
                        && headerFilterStrategy.applyFilterToExternalHeaders(name, value, exchange)) {
                        continue;
                    }

                    // must decode back from safe JMS header name to original header name
                    // when storing on this Camel JmsMessage object.
                    String key = jmsKeyFormatStrategy.decodeKey(name);
                    map.put(key, value);
                } catch (JMSException e) {
                    throw new RuntimeCamelException(name, e);
                }
            }
        }

        return map;
    }

    /**
     * @deprecated not in use
     */
    @Deprecated
    public Object getObjectProperty(Message jmsMessage, String name) throws JMSException {
        // try a direct lookup first
        Object answer = jmsMessage.getObjectProperty(name);
        if (answer == null) {
            // then encode the key and do another lookup
            String key = jmsKeyFormatStrategy.encodeKey(name);
            answer = jmsMessage.getObjectProperty(key);
        }
        return answer;
    }

    protected Object createByteArrayFromBytesMessage(Exchange exchange, BytesMessage message) throws JMSException {
        // ActiveMQ has special optimised mode for bytes message, so we should use streaming if possible
        Long size = getSafeLongProperty(message, "_AMQ_LARGE_SIZE");
        if (size != null && size > 0) {
            LOG.trace("Optimised for Artemis: Reading from BytesMessage in streaming mode directly into CachedOutputStream payload");
            CachedOutputStream cos = new CachedOutputStream(exchange, true);
            // this will save the stream and wait until the entire message is written before continuing.
            message.setObjectProperty("JMS_AMQ_SaveStream", cos);
            try {
                // and then lets get the input stream of this so we can read it
                return cos.getInputStream();
            } catch (IOException e) {
                JMSException cause = new MessageFormatException(e.getMessage());
                cause.initCause(e);
                throw cause;
            } finally {
                IOHelper.close(cos);
            }
        }

        if (message.getBodyLength() > Integer.MAX_VALUE) {
            LOG.warn("Length of BytesMessage is too long: {}", message.getBodyLength());
            return null;
        }
        byte[] result = new byte[(int)message.getBodyLength()];
        message.readBytes(result);
        return result;
    }

    protected Object createInputStreamFromStreamMessage(Exchange exchange, StreamMessage message) throws JMSException {
        return new StreamMessageInputStream(message);
    }

    /**
     * Creates a JMS message from the Camel exchange and message
     *
     * @param exchange the current exchange
     * @param session the JMS session used to create the message
     * @return a newly created JMS Message instance containing the
     * @throws JMSException if the message could not be created
     */
    public Message makeJmsMessage(Exchange exchange, Session session) throws JMSException {
        Message answer = makeJmsMessage(exchange, exchange.getIn(), session, null);
        if (answer != null && messageCreatedStrategy != null) {
            messageCreatedStrategy.onMessageCreated(answer, session, exchange, null);
        }
        return answer;
    }

    /**
     * Creates a JMS message from the Camel exchange and message
     *
     * @param exchange the current exchange
     * @param camelMessage the body to make a javax.jms.Message as
     * @param session the JMS session used to create the message
     * @param cause optional exception occurred that should be sent as reply instead of a regular body
     * @return a newly created JMS Message instance containing the
     * @throws JMSException if the message could not be created
     */
    public Message makeJmsMessage(Exchange exchange, org.apache.camel.Message camelMessage, Session session, Exception cause) throws JMSException {
        Message answer = null;

        boolean alwaysCopy = endpoint != null && endpoint.getConfiguration().isAlwaysCopyMessage();
        boolean force = endpoint != null && endpoint.getConfiguration().isForceSendOriginalMessage();
        if (!alwaysCopy && camelMessage instanceof JmsMessage) {
            JmsMessage jmsMessage = (JmsMessage)camelMessage;
            if (!jmsMessage.shouldCreateNewMessage() || force) {
                answer = jmsMessage.getJmsMessage();

                if (!force) {
                    // answer must match endpoint type
                    JmsMessageType type = endpoint != null ? endpoint.getConfiguration().getJmsMessageType() : null;
                    if (type != null && answer != null) {
                        if (type == JmsMessageType.Text) {
                            answer = answer instanceof TextMessage ? answer : null;
                        } else if (type == JmsMessageType.Bytes) {
                            answer = answer instanceof BytesMessage ? answer : null;
                        } else if (type == JmsMessageType.Map) {
                            answer = answer instanceof MapMessage ? answer : null;
                        } else if (type == JmsMessageType.Object) {
                            answer = answer instanceof ObjectMessage ? answer : null;
                        } else if (type == Stream) {
                            answer = answer instanceof StreamMessage ? answer : null;
                        }
                    }
                }
            }
        }

        if (answer == null) {
            if (cause != null) {
                // an exception occurred so send it as response
                LOG.debug("Will create JmsMessage with caused exception: {}", cause);
                // create jms message containing the caused exception
                answer = createJmsMessage(cause, session);
            } else {
                org.apache.camel.util.ObjectHelper.notNull(camelMessage, "message");
                // create regular jms message using the camel message body
                answer = createJmsMessage(exchange, camelMessage, session, exchange.getContext());
                appendJmsProperties(answer, exchange, camelMessage);
            }
        }

        if (answer != null && messageCreatedStrategy != null) {
            messageCreatedStrategy.onMessageCreated(answer, session, exchange, null);
        }
        return answer;
    }

    /**
     * Appends the JMS headers from the Camel {@link JmsMessage}
     */
    public void appendJmsProperties(Message jmsMessage, Exchange exchange) throws JMSException {
        appendJmsProperties(jmsMessage, exchange, exchange.getIn());
    }

    /**
     * Appends the JMS headers from the Camel {@link JmsMessage}
     */
    public void appendJmsProperties(Message jmsMessage, Exchange exchange, org.apache.camel.Message in) throws JMSException {
        Set<Map.Entry<String, Object>> entries = in.getHeaders().entrySet();        
        for (Map.Entry<String, Object> entry : entries) {
            String headerName = entry.getKey();
            Object headerValue = entry.getValue();
            appendJmsProperty(jmsMessage, exchange, in, headerName, headerValue);
        }
    }

    public void appendJmsProperty(Message jmsMessage, Exchange exchange, org.apache.camel.Message in,
                                  String headerName, Object headerValue) throws JMSException {
        if (isStandardJMSHeader(headerName)) {
            if (headerName.equals("JMSCorrelationID")) {
                jmsMessage.setJMSCorrelationID(ExchangeHelper.convertToType(exchange, String.class, headerValue));
            } else if (headerName.equals("JMSReplyTo") && headerValue != null) {
                if (headerValue instanceof String) {
                    // if the value is a String we must normalize it first, and must include the prefix
                    // as ActiveMQ requires that when converting the String to a javax.jms.Destination type
                    headerValue = normalizeDestinationName((String) headerValue, true);
                }
                Destination replyTo = ExchangeHelper.convertToType(exchange, Destination.class, headerValue);
                JmsMessageHelper.setJMSReplyTo(jmsMessage, replyTo);
            } else if (headerName.equals("JMSType")) {
                jmsMessage.setJMSType(ExchangeHelper.convertToType(exchange, String.class, headerValue));
            } else if (headerName.equals("JMSPriority")) {
                jmsMessage.setJMSPriority(ExchangeHelper.convertToType(exchange, Integer.class, headerValue));
            } else if (headerName.equals("JMSDeliveryMode")) {
                JmsMessageHelper.setJMSDeliveryMode(exchange, jmsMessage, headerValue);
            } else if (headerName.equals("JMSExpiration")) {
                jmsMessage.setJMSExpiration(ExchangeHelper.convertToType(exchange, Long.class, headerValue));
            } else {
                // The following properties are set by the MessageProducer:
                // JMSDestination
                // The following are set on the underlying JMS provider:
                // JMSMessageID, JMSTimestamp, JMSRedelivered
                // log at trace level to not spam log
                LOG.trace("Ignoring JMS header: {} with value: {}", headerName, headerValue);
            }
        } else if (shouldOutputHeader(in, headerName, headerValue, exchange)) {
            // only primitive headers and strings is allowed as properties
            // see message properties: http://java.sun.com/j2ee/1.4/docs/api/javax/jms/Message.html
            Object value = getValidJMSHeaderValue(headerName, headerValue);
            // if the value was null, then it may be allowed as an additional header
            if (value == null && endpoint.getConfiguration().getAllowAdditionalHeaders() != null) {
                Iterator it = ObjectHelper.createIterator(endpoint.getConfiguration().getAllowAdditionalHeaders());
                while (it.hasNext()) {
                    String pattern = (String) it.next();
                    if (PatternHelper.matchPattern(headerName, pattern)) {
                        LOG.debug("Header {} allowed as additional header despite not being valid according to the JMS specification", headerName);
                        value = headerValue;
                        break;
                    }
                }
            }
            if (value != null) {
                // must encode to safe JMS header name before setting property on jmsMessage
                String key = jmsKeyFormatStrategy.encodeKey(headerName);
                // set the property
                JmsMessageHelper.setProperty(jmsMessage, key, value);
            } else if (LOG.isDebugEnabled()) {
                // okay the value is not a primitive or string so we cannot sent it over the wire
                LOG.debug("Ignoring non primitive header: {} of class: {} with value: {}",
                        new Object[]{headerName, headerValue.getClass().getName(), headerValue});
            }
        }
    }

    /**
     * Is the given header a standard JMS header
     * @param headerName the header name
     * @return <tt>true</tt> if its a standard JMS header
     */
    protected boolean isStandardJMSHeader(String headerName) {
        if (!headerName.startsWith("JMS")) {
            return false;
        }
        if (headerName.startsWith("JMSX")) {
            return false;
        }
        // vendors will use JMS_XXX as their special headers (where XXX is vendor name, such as JMS_IBM)
        if (headerName.startsWith("JMS_")) {
            return false;
        }

        // the 4th char must be a letter to be a standard JMS header
        if (headerName.length() > 3) {
            Character fourth = headerName.charAt(3);
            if (Character.isLetter(fourth)) {
                return true;
            }
        }

        return false;
    }

    /**
     * Strategy to test if the given header is valid according to the JMS spec to be set as a property
     * on the JMS message.
     * <p/>
     * This default implementation will allow:
     * <ul>
     *   <li>any primitives and their counter Objects (Integer, Double etc.)</li>
     *   <li>String and any other literals, Character, CharSequence</li>
     *   <li>Boolean</li>
     *   <li>Number</li>
     *   <li>java.math.BigInteger</li>
     *   <li>java.math.BigDecimal</li>
     *   <li>java.util.Date</li>
     * </ul>
     *
     * @param headerName   the header name
     * @param headerValue  the header value
     * @return  the value to use, <tt>null</tt> to ignore this header
     */
    protected Object getValidJMSHeaderValue(String headerName, Object headerValue) {
        if (headerValue instanceof String) {
            return headerValue;
        } else if (headerValue instanceof BigInteger) {
            return headerValue.toString();
        } else if (headerValue instanceof BigDecimal) {
            return headerValue.toString();
        } else if (headerValue instanceof Number) {
            return headerValue;
        } else if (headerValue instanceof Character) {
            return headerValue;
        } else if (headerValue instanceof CharSequence) {
            return headerValue.toString();
        } else if (headerValue instanceof Boolean) {
            return headerValue;
        } else if (headerValue instanceof Date) {
            if (this.endpoint.getConfiguration().isFormatDateHeadersToIso8601()) {
                return ZonedDateTime.ofInstant(((Date)headerValue).toInstant(), ZoneOffset.UTC).toString();
            } else {
                return headerValue.toString();
            }
        }
        return null;
    }

    protected Message createJmsMessage(Exception cause, Session session) throws JMSException {
        LOG.trace("Using JmsMessageType: {}", Object);
        Message answer = session.createObjectMessage(cause);
        // ensure default delivery mode is used by default
        answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
        return answer;
    }

    protected Message createJmsMessage(Exchange exchange, org.apache.camel.Message camelMessage, Session session, CamelContext context) throws JMSException {
        Message answer = createJmsMessage(exchange, camelMessage.getBody(), camelMessage.getHeaders(), session, context);

        // special for transferFault
        boolean isFault = camelMessage.isFault();
        if (answer != null && isFault && endpoint != null && endpoint.isTransferFault()) {
            answer.setBooleanProperty(JmsConstants.JMS_TRANSFER_FAULT, true);
        }
        return answer;
    }

    protected Message createJmsMessage(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context) throws JMSException {
        JmsMessageType type;

        // special for transferExchange
        if (endpoint != null && endpoint.isTransferExchange()) {
            LOG.trace("Option transferExchange=true so we use JmsMessageType: Object");
            Serializable holder = DefaultExchangeHolder.marshal(exchange, true, endpoint.isAllowSerializedHeaders());
            Message answer = session.createObjectMessage(holder);
            // ensure default delivery mode is used by default
            answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
            return answer;
        }

        // use a custom message converter
        if (endpoint != null && endpoint.getMessageConverter() != null) {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Creating JmsMessage using a custom MessageConverter: {} with body: {}", endpoint.getMessageConverter(), body);
            }
            return endpoint.getMessageConverter().toMessage(body, session);
        }

        // check if header have a type set, if so we force to use it
        if (headers.containsKey(JmsConstants.JMS_MESSAGE_TYPE)) {
            type = context.getTypeConverter().convertTo(JmsMessageType.class, headers.get(JmsConstants.JMS_MESSAGE_TYPE));
        } else if (endpoint != null && endpoint.getConfiguration().getJmsMessageType() != null) {
            // force a specific type from the endpoint configuration
            type = endpoint.getConfiguration().getJmsMessageType();
        } else {
            type = getJMSMessageTypeForBody(exchange, body, headers, session, context);
        }

        // create the JmsMessage based on the type
        if (type != null) {
            if (body == null && (endpoint != null && !endpoint.getConfiguration().isAllowNullBody())) {
                throw new JMSException("Cannot send message as message body is null, and option allowNullBody is false.");
            }
            LOG.trace("Using JmsMessageType: {}", type);
            Message answer = createJmsMessageForType(exchange, body, headers, session, context, type);
            // ensure default delivery mode is used by default
            answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
            return answer;
        }

        // check for null body
        if (body == null && (endpoint != null && !endpoint.getConfiguration().isAllowNullBody())) {
            throw new JMSException("Cannot send message as message body is null, and option allowNullBody is false.");
        }

        // warn if the body could not be mapped
        if (body != null && LOG.isWarnEnabled()) {
            LOG.warn("Cannot determine specific JmsMessage type to use from body class."
                    + " Will use generic JmsMessage."
                    + " Body class: " + org.apache.camel.util.ObjectHelper.classCanonicalName(body)
                    + ". If you want to send a POJO then your class might need to implement java.io.Serializable"
                    + ", or you can force a specific type by setting the jmsMessageType option on the JMS endpoint.");
        }

        // return a default message
        Message answer = session.createMessage();
        // ensure default delivery mode is used by default
        answer.setJMSDeliveryMode(Message.DEFAULT_DELIVERY_MODE);
        return answer;
    }

    /**
     * Return the {@link JmsMessageType} 
     * 
     * @return type or null if no mapping was possible
     */
    protected JmsMessageType getJMSMessageTypeForBody(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context) {
        boolean streamingEnabled = endpoint.getConfiguration().isStreamMessageTypeEnabled();

        JmsMessageType type = null;
        // let body determine the type
        if (body instanceof Node || body instanceof String) {
            type = Text;
        } else if (body instanceof byte[] || body instanceof ByteBuffer) {
            type = Bytes;
        } else if (body instanceof WrappedFile || body instanceof File || body instanceof Reader
                || body instanceof InputStream || body instanceof StreamCache) {
            type = streamingEnabled ? Stream : Bytes;
        } else if (body instanceof Map) {
            type = Map;
        } else if (body instanceof Serializable) {
            type = Object;            
        } else if (exchange.getContext().getTypeConverter().tryConvertTo(File.class, body) != null
                || exchange.getContext().getTypeConverter().tryConvertTo(InputStream.class, body) != null) {
            type = streamingEnabled ? Stream : Bytes;
        }

        if (type == Stream) {
            boolean artemis = isVendor(session, "Artemis");
            if (artemis) {
                // if running ActiveMQ Artemis then it has optimised streaming mode using byte messages so enforce as bytes
                type = Bytes;
            }
        }

        return type;
    }
    
    /**
     * 
     * Create the {@link Message} 
     * 
     * @return jmsMessage or null if the mapping was not successfully
     */
    protected Message createJmsMessageForType(Exchange exchange, Object body, Map<String, Object> headers, Session session, CamelContext context, JmsMessageType type) throws JMSException {
        switch (type) {
        case Text: {
            TextMessage message = session.createTextMessage();
            if (body != null) {
                String payload = context.getTypeConverter().convertTo(String.class, exchange, body);
                message.setText(payload);
            }
            return message;
        }
        case Bytes: {
            BytesMessage message = session.createBytesMessage();
            if (body != null) {
                try {
                    if (isVendor(session, "Artemis")) {
                        LOG.trace("Optimised for Artemis: Streaming payload in BytesMessage");
                        InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
                        message.setObjectProperty("JMS_AMQ_InputStream", is);
                        LOG.trace("Optimised for Artemis: Finished streaming payload in BytesMessage");
                    } else {
                        byte[] payload = context.getTypeConverter().mandatoryConvertTo(byte[].class, exchange, body);
                        message.writeBytes(payload);
                    }
                } catch (NoTypeConversionAvailableException e) {
                    // cannot convert to inputstream then thrown an exception to avoid sending a null message
                    JMSException cause = new MessageFormatException(e.getMessage());
                    cause.initCause(e);
                    throw cause;
                }
            }
            return message;
        }
        case Map: {
            MapMessage message = session.createMapMessage();
            if (body != null) {
                Map<?, ?> payload = context.getTypeConverter().convertTo(Map.class, exchange, body);
                populateMapMessage(message, payload, context);
            }
            return message;
        }
        case Object: {
            ObjectMessage message = session.createObjectMessage();
            if (body != null) {
                try {
                    Serializable payload = context.getTypeConverter().mandatoryConvertTo(Serializable.class, exchange, body);
                    message.setObject(payload);
                } catch (NoTypeConversionAvailableException e) {
                    // cannot convert to serializable then thrown an exception to avoid sending a null message
                    JMSException cause = new MessageFormatException(e.getMessage());
                    cause.initCause(e);
                    throw cause;
                }
            }
            return message;
        }
        case Stream: {
            StreamMessage message = session.createStreamMessage();
            if (body != null) {
                long size = 0;
                try {
                    InputStream is = context.getTypeConverter().mandatoryConvertTo(InputStream.class, exchange, body);
                    LOG.trace("Writing payload in StreamMessage");
                    // assume streaming is bigger payload so use same buffer size as the file component
                    byte[] buffer = new byte[FileUtil.BUFFER_SIZE];
                    int len = 0;
                    int count = 0;
                    while (len >= 0) {
                        count++;
                        len = is.read(buffer);
                        if (len >= 0) {
                            size += len;
                            LOG.trace("Writing payload chunk {} as bytes in StreamMessage", count);
                            message.writeBytes(buffer, 0, len);
                        }
                    }
                    LOG.trace("Finished writing payload (size {}) as bytes in StreamMessage", size);
                } catch (NoTypeConversionAvailableException | IOException e) {
                    // cannot convert to inputstream then thrown an exception to avoid sending a null message
                    JMSException cause = new MessageFormatException(e.getMessage());
                    cause.initCause(e);
                    throw cause;
                }
            }
            return message;
        }
        default:
            break;
        }
        return null;
    }
    /**
     * Populates a {@link MapMessage} from a {@link Map} instance.
     */
    protected void populateMapMessage(MapMessage message, Map<?, ?> map, CamelContext context)
        throws JMSException {
        for (Entry<?, ?> entry : map.entrySet()) {
            String keyString = CamelContextHelper.convertTo(context, String.class, entry.getKey());
            if (keyString != null) {
                message.setObject(keyString, entry.getValue());
            }
        }
    }

    /**
     * Extracts a {@link Map} from a {@link MapMessage}
     */
    public Map<String, Object> createMapFromMapMessage(MapMessage message) throws JMSException {
        Map<String, Object> answer = new HashMap<>();
        Enumeration<?> names = message.getMapNames();
        while (names.hasMoreElements()) {
            String name = names.nextElement().toString();
            Object value = message.getObject(name);
            answer.put(name, value);
        }
        return answer;
    }

    /**
     * Strategy to allow filtering of headers which are put on the JMS message
     * <p/>
     * <b>Note</b>: Currently only supports sending java identifiers as keys
     */
    protected boolean shouldOutputHeader(org.apache.camel.Message camelMessage, String headerName,
                                         Object headerValue, Exchange exchange) {
        return headerFilterStrategy == null
            || !headerFilterStrategy.applyFilterToCamelHeaders(headerName, headerValue, exchange);
    }

}