/*
 * 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.cassandra.net;

import java.nio.ByteBuffer;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.memory.BufferPool;
import org.apache.cassandra.utils.memory.BufferPools;

public abstract class FrameEncoder extends ChannelOutboundHandlerAdapter
{
    protected static final BufferPool bufferPool = BufferPools.forNetworking();

    /**
     * An abstraction useful for transparently allocating buffers that can be written to upstream
     * of the {@code FrameEncoder} without knowledge of the encoder's frame layout, while ensuring
     * enough space to write the remainder of the frame's contents is reserved.
     */
    public static class Payload
    {
        public static final int MAX_SIZE = 1 << 17;
        // isSelfContained is a flag in the Frame API, indicating if the contents consists of only complete messages
        private boolean isSelfContained;
        // the buffer to write to
        public final ByteBuffer buffer;
        // the number of header bytes to reserve
        final int headerLength;
        // the number of trailer bytes to reserve
        final int trailerLength;
        // an API-misuse detector
        private boolean isFinished = false;

        Payload(boolean isSelfContained, int payloadCapacity)
        {
            this(isSelfContained, payloadCapacity, 0, 0);
        }

        Payload(boolean isSelfContained, int payloadCapacity, int headerLength, int trailerLength)
        {
            this.isSelfContained = isSelfContained;
            this.headerLength = headerLength;
            this.trailerLength = trailerLength;

            buffer = bufferPool.getAtLeast(payloadCapacity + headerLength + trailerLength, BufferType.OFF_HEAP);
            assert buffer.capacity() >= payloadCapacity + headerLength + trailerLength;
            buffer.position(headerLength);
            buffer.limit(buffer.capacity() - trailerLength);
        }

        void setSelfContained(boolean isSelfContained)
        {
            this.isSelfContained = isSelfContained;
        }

        // do not invoke after finish()
        int length()
        {
            assert !isFinished;
            return buffer.position() - headerLength;
        }

        // do not invoke after finish()
        public int remaining()
        {
            assert !isFinished;
            return buffer.remaining();
        }

        // do not invoke after finish()
        void trim(int length)
        {
            assert !isFinished;
            buffer.position(headerLength + length);
        }

        // may not be written to or queried, after this is invoked; must be passed straight to an encoder (or release called)
        public void finish()
        {
            assert !isFinished;
            isFinished = true;
            buffer.limit(buffer.position() + trailerLength);
            buffer.position(0);
            bufferPool.putUnusedPortion(buffer);
        }

        public void release()
        {
            bufferPool.put(buffer);
        }
    }

    public interface PayloadAllocator
    {
        public static final PayloadAllocator simple = Payload::new;
        Payload allocate(boolean isSelfContained, int capacity);
    }

    public PayloadAllocator allocator()
    {
        return PayloadAllocator.simple;
    }

    /**
     * Takes ownership of the lifetime of the provided buffer, which can be assumed to be managed by BufferPool
     */
    abstract ByteBuf encode(boolean isSelfContained, ByteBuffer buffer);

    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception
    {
        if (!(msg instanceof Payload))
            throw new IllegalStateException("Unexpected type: " + msg);

        Payload payload = (Payload) msg;
        ByteBuf write = encode(payload.isSelfContained, payload.buffer);
        ctx.write(write, promise);
    }
}