/*
 * 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.engine.impl;

import static org.junit.Assert.assertEquals;

import java.nio.ByteBuffer;

import org.apache.qpid.proton.amqp.security.SaslFrameBody;
import org.apache.qpid.proton.amqp.transport.FrameBody;
import org.apache.qpid.proton.codec.AMQPDefinedTypes;
import org.apache.qpid.proton.codec.DecoderImpl;
import org.apache.qpid.proton.codec.EncoderImpl;
import org.apache.qpid.proton.codec.WritableBuffer;

/**
 * Generates frames as per section 2.3.1 of the AMQP spec
 */
public class AmqpFramer
{
    // My test data is generated by the decoder and encoder from proton-j
    // although at run-time the Engine internally uses its own implementation.

    private DecoderImpl _decoder = new DecoderImpl();
    private EncoderImpl _encoder = new EncoderImpl(_decoder);

    public AmqpFramer()
    {
        AMQPDefinedTypes.registerAllTypes(_decoder, _encoder);
    }

    public byte[] createEmptyFrame(int channel)
    {
        byte[] emptyFrame = generateFrame(channel, null);
        return emptyFrame;
    }

    public byte[] generateFrame(int channel, FrameBody frameBody)
    {
        byte[] emptyExtendedHeader = new byte[] {};
        return generateFrame(channel, emptyExtendedHeader, frameBody);
    }

    public byte[] generateFrame(int channel, byte[] extendedHeader, FrameBody frameBody)
    {
        return generateFrame(channel, extendedHeader, frameBody, (byte)0);
    }

    public byte[] generateSaslFrame(int channel, byte[] extendedHeader, SaslFrameBody frameBody)
    {
        return generateFrame(channel, extendedHeader, frameBody, (byte)1);
    }

    /**
     * @param amqpFrameType indicates either AMQP or SASL
     * @param frameBody is currently expected to be a {@link FrameBody} or a {@link SaslFrameBody}
     */
    public byte[] generateFrame(int channel, byte[] extendedHeader, Object frameBody, byte amqpFrameType)
    {
        assertEquals("Extended header must be multiple of 4 bytes", 0, extendedHeader.length % 4);
        int numberOfExtendedHeaderFourByteWords = extendedHeader.length / 4;

        ByteBuffer buffer = ByteBuffer.allocate(1024);

        buffer.position(8); // leave hole for frame header
        _encoder.setByteBuffer(new WritableBuffer.ByteBufferWrapper(buffer));

        // write extended header - maybe empty
        buffer.put(extendedHeader);
        // write frame body
        if (frameBody != null)
        {
            _encoder.writeObject(frameBody);
        }

        int frameSize = buffer.position();
        int framePreambleSizeInFourByteWords = 2;
        byte dataOffsetFourByteWords = (byte)(framePreambleSizeInFourByteWords + numberOfExtendedHeaderFourByteWords);
        buffer.rewind();
        buffer.putInt(frameSize);
        buffer.put(dataOffsetFourByteWords);
        buffer.put(amqpFrameType);
        buffer.putShort((short)channel);

        byte[] target = new byte[frameSize];

        buffer.rewind();
        buffer.get(target, 0, frameSize);
        return target;
    }
}