/**
 * 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.qpid.proton.jms;

import org.apache.qpid.proton.amqp.messaging.Section;
import org.apache.qpid.proton.codec.CompositeWritableBuffer;
import org.apache.qpid.proton.codec.WritableBuffer;
import org.apache.qpid.proton.codec.DroppingWritableBuffer;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedByte;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.*;

import javax.jms.*;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.HashMap;
import org.apache.qpid.proton.message.impl.MessageImpl;

/**
* @author <a href="http://hiramchirino.com">Hiram Chirino</a>
*/
public class JMSMappingOutboundTransformer extends OutboundTransformer {

    String prefixDeliveryAnnotations = "DA_";
    String prefixMessageAnnotations= "MA_";
    String prefixFooter = "FT_";

    public JMSMappingOutboundTransformer(JMSVendor vendor) {
        super(vendor);
    }

    @Override
    public EncodedMessage transform(Message msg) throws Exception {
        if( msg == null )
            return null;
        try {
            if( msg.getBooleanProperty(prefixVendor + "NATIVE") ) {
                return null;
            }
        } catch (MessageFormatException e) {
            return null;
        }
        return transform(this, msg);
    }

    static EncodedMessage transform(JMSMappingOutboundTransformer options, Message msg) throws JMSException, UnsupportedEncodingException {
        final JMSVendor vendor = options.vendor;

        final String messageFormatKey = options.prefixVendor + "MESSAGE_FORMAT";
        final String nativeKey = options.prefixVendor + "NATIVE";
        final String firstAcquirerKey = options.prefixVendor + "FirstAcquirer";
        final String prefixDeliveryAnnotationsKey = options.prefixVendor + options.prefixDeliveryAnnotations;
        final String prefixMessageAnnotationsKey = options.prefixVendor + options.prefixMessageAnnotations;
        final String subjectKey =  options.prefixVendor +"Subject";
        final String contentTypeKey = options.prefixVendor +"ContentType";
        final String contentEncodingKey = options.prefixVendor +"ContentEncoding";
        final String replyToGroupIDKey = options.prefixVendor +"ReplyToGroupID";
        final String prefixFooterKey = options.prefixVendor + options.prefixFooter;

        long messageFormat;
        try {
            messageFormat = msg.getLongProperty(messageFormatKey);
        } catch (MessageFormatException e) {
            return null;
        }

        Header header = new Header();
        Properties props=new Properties();
        HashMap daMap = null;
        HashMap maMap = null;
        HashMap apMap = null;
        Section body=null;
        HashMap footerMap = null;
        if( msg instanceof BytesMessage ) {
            BytesMessage m = (BytesMessage)msg;
            byte data[] = new byte[(int) m.getBodyLength()];
            m.readBytes(data);
            body = new Data(new Binary(data));
        } if( msg instanceof TextMessage ) {
            body = new AmqpValue(((TextMessage) msg).getText());
        } if( msg instanceof MapMessage ) {
            final HashMap map = new HashMap();
            final MapMessage m = (MapMessage) msg;
            final Enumeration names = m.getMapNames();
            while (names.hasMoreElements()) {
                String key = (String) names.nextElement();
                map.put(key, m.getObject(key));
            }
            body = new AmqpValue(map);
        } if( msg instanceof StreamMessage ) {
            ArrayList list = new ArrayList();
            final StreamMessage m = (StreamMessage) msg;
            try {
                while(true) {
                    list.add(m.readObject());
                }
            } catch(MessageEOFException e){}
            body = new AmqpSequence(list);
        } if( msg instanceof ObjectMessage ) {
            body = new AmqpValue(((ObjectMessage) msg).getObject());
        }

        header.setDurable(msg.getJMSDeliveryMode() == DeliveryMode.PERSISTENT ? true : false);
        header.setPriority(new UnsignedByte((byte) msg.getJMSPriority()));
        if( msg.getJMSExpiration() != 0 ) {
            header.setTtl(new UnsignedInteger((int) msg.getJMSExpiration()));
        }
        if( msg.getJMSType()!=null ) {
            if( maMap==null ) maMap = new HashMap();
            maMap.put("x-opt-jms-type", msg.getJMSType());
        }
        if( msg.getJMSMessageID()!=null ) {
            props.setMessageId(msg.getJMSMessageID());
        }
        if( msg.getJMSDestination()!=null ) {
            props.setTo(vendor.toAddress(msg.getJMSDestination()));
            if( maMap==null ) maMap = new HashMap();
            maMap.put("x-opt-to-type", destinationAttributes(msg.getJMSDestination()));
        }
        if( msg.getJMSReplyTo()!=null ) {
            props.setReplyTo(vendor.toAddress(msg.getJMSReplyTo()));
            if( maMap==null ) maMap = new HashMap();
            maMap.put("x-opt-reply-type", destinationAttributes(msg.getJMSReplyTo()));
        }
        if( msg.getJMSCorrelationID()!=null ) {
            props.setCorrelationId(msg.getJMSCorrelationID());
        }
        if( msg.getJMSExpiration() != 0 ) {
            props.setAbsoluteExpiryTime(new Date(msg.getJMSExpiration()));
        }
        if( msg.getJMSTimestamp()!= 0 ) {
            props.setCreationTime(new Date(msg.getJMSTimestamp()));
        }

        final Enumeration keys = msg.getPropertyNames();
        while (keys.hasMoreElements()) {
            String key = (String) keys.nextElement();
            if( key.equals(messageFormatKey) || key.equals(nativeKey)) {
                // skip..
            } else if( key.equals(firstAcquirerKey) ) {
                header.setFirstAcquirer(msg.getBooleanProperty(key));
            } else if( key.startsWith("JMSXDeliveryCount") ) {
                header.setDeliveryCount(new UnsignedInteger(msg.getIntProperty(key)));
            } else if( key.startsWith("JMSXUserID") ) {
                String value = msg.getStringProperty(key);
                props.setUserId(new Binary(value.getBytes("UTF-8")));
            } else if( key.startsWith("JMSXGroupID") ) {
                String value = msg.getStringProperty(key);
                props.setGroupId(value);
                if( apMap==null ) apMap = new HashMap();
                apMap.put(key, value);
            } else if( key.startsWith("JMSXGroupSeq") ) {
                UnsignedInteger value = new UnsignedInteger(msg.getIntProperty(key));
                props.setGroupSequence(value);
                if( apMap==null ) apMap = new HashMap();
                apMap.put(key, value);
            } else if( key.startsWith(prefixDeliveryAnnotationsKey) ) {
                if( daMap == null ) daMap = new HashMap();
                String name = key.substring(prefixDeliveryAnnotationsKey.length());
                daMap.put(name, msg.getObjectProperty(key));
            } else if( key.startsWith(prefixMessageAnnotationsKey) ) {
                if( maMap==null ) maMap = new HashMap();
                String name = key.substring(prefixMessageAnnotationsKey.length());
                maMap.put(name, msg.getObjectProperty(key));
            } else if( key.equals(subjectKey) ) {
                props.setSubject(msg.getStringProperty(key));
            } else if( key.equals(contentTypeKey) ) {
                props.setContentType(Symbol.getSymbol(msg.getStringProperty(key)));
            } else if( key.equals(contentEncodingKey) ) {
                props.setContentEncoding(Symbol.getSymbol(msg.getStringProperty(key)));
            } else if( key.equals(replyToGroupIDKey) ) {
                props.setReplyToGroupId(msg.getStringProperty(key));
            } else if( key.startsWith(prefixFooterKey) ) {
                if( footerMap==null ) footerMap = new HashMap();
                String name = key.substring(prefixFooterKey.length());
                footerMap.put(name, msg.getObjectProperty(key));
            } else {
                if( apMap==null ) apMap = new HashMap();
                apMap.put(key, msg.getObjectProperty(key));
            }
        }


        MessageAnnotations ma=null;
        if( maMap!=null ) ma = new MessageAnnotations(maMap);
        DeliveryAnnotations da=null;
        if( daMap!=null ) da = new DeliveryAnnotations(daMap);
        ApplicationProperties ap=null;
        if( apMap!=null ) ap = new ApplicationProperties(apMap);
        Footer footer=null;
        if( footerMap!=null ) footer = new Footer(footerMap);

        MessageImpl amqp = new MessageImpl(header, da, ma, props, ap, body, footer);

        ByteBuffer buffer = ByteBuffer.wrap(new byte[1024*4]);
        final DroppingWritableBuffer overflow = new DroppingWritableBuffer();
        int c = amqp.encode(new CompositeWritableBuffer(new WritableBuffer.ByteBufferWrapper(buffer), overflow));
        if( overflow.position() > 0 ) {
            buffer = ByteBuffer.wrap(new byte[1024*4+overflow.position()]);
            c = amqp.encode(new WritableBuffer.ByteBufferWrapper(buffer));
        }

        return new EncodedMessage(messageFormat, buffer.array(), 0, c);
    }

    private static String destinationAttributes(Destination destination) {
        if( destination instanceof Queue ) {
            if( destination instanceof TemporaryQueue ) {
                return "temporary,queue";
            } else {
                return "queue";
            }
        }
        if( destination instanceof Topic ) {
            if( destination instanceof TemporaryTopic ) {
                return "temporary,topic";
            } else {
                return "topic";
            }
        }
        return "";
    }
}