/**
 * 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.jms.provider.amqp.message;

import static org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport.JMS_BYTES_MESSAGE;
import static org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport.JMS_MSG_TYPE;
import static org.apache.qpid.jms.provider.amqp.message.AmqpMessageSupport.OCTET_STREAM_CONTENT_TYPE;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.Unpooled;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;

import javax.jms.IllegalStateException;
import javax.jms.JMSException;

import org.apache.qpid.jms.message.facade.JmsBytesMessageFacade;
import org.apache.qpid.jms.provider.amqp.AmqpConnection;
import org.apache.qpid.jms.provider.amqp.AmqpConsumer;
import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Section;
import org.apache.qpid.proton.message.Message;

/**
 * A JmsBytesMessageFacade that wraps around Proton AMQP Message instances to provide
 * access to the underlying bytes contained in the message.
 */
public class AmqpJmsBytesMessageFacade extends AmqpJmsMessageFacade implements JmsBytesMessageFacade {

    private static final Binary EMPTY_BODY = new Binary(new byte[0]);
    private static final Data EMPTY_DATA = new Data(EMPTY_BODY);

    private transient ByteBufInputStream bytesIn;
    private transient ByteBufOutputStream bytesOut;

    /**
     * Creates a new facade instance
     *
     * @param connection
     *        the AmqpConnection that under which this facade was created.
     */
    public AmqpJmsBytesMessageFacade(AmqpConnection connection) {
        super(connection);
        setContentType(OCTET_STREAM_CONTENT_TYPE);
        setMessageAnnotation(JMS_MSG_TYPE, JMS_BYTES_MESSAGE);
    }

    /**
     * Creates a new Facade around an incoming AMQP Message for dispatch to the
     * JMS Consumer instance.
     *
     * @param consumer
     *        the consumer that received this message.
     * @param message
     *        the incoming Message instance that is being wrapped.
     */
    public AmqpJmsBytesMessageFacade(AmqpConsumer consumer, Message message) {
        super(consumer, message);
    }

    @Override
    public AmqpJmsBytesMessageFacade copy() {
        reset();
        AmqpJmsBytesMessageFacade copy = new AmqpJmsBytesMessageFacade(connection);
        copyInto(copy);

        Binary payload = getBinaryFromBody();
        if (payload != null && payload.getLength() > 0) {
            byte[] result = new byte[payload.getLength()];
            System.arraycopy(payload.getArray(), payload.getArrayOffset(), result, 0, payload.getLength());
            copy.message.setBody(new Data(new Binary(result)));
        } else {
            copy.message.setBody(EMPTY_DATA);
        }

        return copy;
    }

    @Override
    public byte getJmsMsgType() {
        return JMS_BYTES_MESSAGE;
    }

    @Override
    public boolean isEmpty() {
        Binary payload = getBinaryFromBody();
        return payload == null || payload.getLength() == 0;
    }

    @Override
    public void clearBody() {
        if (bytesIn != null) {
            try {
                bytesIn.close();
            } catch (IOException e) {
            }
            bytesIn = null;
        }
        if (bytesOut != null) {
            try {
                bytesOut.close();
            } catch (IOException e) {
            }

            bytesOut = null;
        }

        message.setBody(EMPTY_DATA);
    }

    @Override
    public InputStream getInputStream() throws JMSException {
        if (bytesOut != null) {
            throw new IllegalStateException("Body is being written to, cannot perform a read.");
        }

        if (bytesIn == null) {
            Binary body = getBinaryFromBody();
            // Duplicate the content buffer to allow for getBodyLength() validity.
            bytesIn = new ByteBufInputStream(
                Unpooled.wrappedBuffer(body.getArray(), body.getArrayOffset(), body.getLength()));
        }

        return bytesIn;
    }

    @Override
    public OutputStream getOutputStream() throws JMSException {
        if (bytesIn != null) {
            throw new IllegalStateException("Body is being read from, cannot perform a write.");
        }

        if (bytesOut == null) {
            bytesOut = new ByteBufOutputStream(Unpooled.buffer());
            message.setBody(EMPTY_DATA);
        }

        return bytesOut;
    }

    @Override
    public void reset() {
        if (bytesOut != null) {
            ByteBuf writeBuf = bytesOut.buffer();
            Binary body = new Binary(writeBuf.array(), writeBuf.arrayOffset(), writeBuf.readableBytes());
            message.setBody(new Data(body));
            try {
                bytesOut.close();
            } catch (IOException e) {
            }
            bytesOut = null;
        } else if (bytesIn != null) {
            try {
                bytesIn.close();
            } catch (IOException e) {
            }
            bytesIn = null;
        }
    }

    @Override
    public int getBodyLength() {
        return getBinaryFromBody().getLength();
    }

    private Binary getBinaryFromBody() {
        Section body = getAmqpMessage().getBody();
        Binary result = EMPTY_BODY;

        if (body == null) {
            return result;
        } else if (body instanceof Data) {
            Binary payload = ((Data) body).getValue();
            if (payload != null && payload.getLength() != 0) {
                result = payload;
            }
        } else if(body instanceof AmqpValue) {
            Object value = ((AmqpValue) body).getValue();
            if (value == null) {
                return result;
            }

            if (value instanceof Binary) {
                Binary payload = (Binary)value;
                if (payload.getLength() != 0) {
                    result = payload;
                }
            } else {
                throw new java.lang.IllegalStateException("Unexpected amqp-value body content type: " + value.getClass().getSimpleName());
            }
        } else {
            throw new java.lang.IllegalStateException("Unexpected body content type: " + body.getClass().getSimpleName());
        }

        return result;
    }
}

@@ -1375,33 +1375,8 @@
 eam;
-%0Aimport java.util.Arrays;
 %0A%0Aim
@@ -2171,18 +2171,20 @@
  EMPTY_B
-OD
+INAR
 Y = new 
@@ -2240,20 +2240,20 @@
 a EMPTY_
-DATA
+BODY
  = new D
@@ -2263,18 +2263,20 @@
 (EMPTY_B
-OD
+INAR
 Y);%0A%0A   
@@ -3738,36 +3738,36 @@
 e.setBody(EMPTY_
-DATA
+BODY
 );%0A        %7D%0A%0A  
@@ -4458,28 +4458,28 @@
 tBody(EMPTY_
-DATA
+BODY
 );%0A    %7D%0A%0A  
@@ -5394,12 +5394,12 @@
 PTY_
-DATA
+BODY
 );%0A 
@@ -6257,18 +6257,20 @@
  EMPTY_B
-OD
+INAR
 Y;%0A%0A    

rename for clarity