/*
 * Copyright 2016 Netflix, Inc.
 *
 * Licensed 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 io.reactivesocket.frame;

import io.reactivesocket.FrameType;
import org.agrona.BitUtil;
import org.agrona.DirectBuffer;
import org.agrona.MutableDirectBuffer;

import java.nio.ByteBuffer;
import java.nio.ByteOrder;

public class RequestFrameFlyweight {

    private RequestFrameFlyweight() {}

    // TODO(lexs) Remove flag for initial N present
    public static final int FLAGS_REQUEST_CHANNEL_N = 0b00_0010_0000;

    // relative to start of passed offset
    private static final int INITIAL_REQUEST_N_FIELD_OFFSET = FrameHeaderFlyweight.FRAME_HEADER_LENGTH;

    public static int computeFrameLength(final FrameType type, final int metadataLength, final int dataLength) {
        int length = FrameHeaderFlyweight.computeFrameHeaderLength(type, metadataLength, dataLength);

        if (type.hasInitialRequestN()) {
            length += BitUtil.SIZE_OF_INT;
        }

        return length;
    }

    public static int encode(
        final MutableDirectBuffer mutableDirectBuffer,
        final int offset,
        final int streamId,
        int flags,
        final FrameType type,
        final int initialRequestN,
        final ByteBuffer metadata,
        final ByteBuffer data
    ) {
        final int frameLength = computeFrameLength(type, metadata.remaining(), data.remaining());

        flags |= FLAGS_REQUEST_CHANNEL_N;
        int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, type, streamId);

        mutableDirectBuffer.putInt(offset + INITIAL_REQUEST_N_FIELD_OFFSET, initialRequestN, ByteOrder.BIG_ENDIAN);
        length += BitUtil.SIZE_OF_INT;

        length += FrameHeaderFlyweight.encodeMetadata(mutableDirectBuffer, offset, offset + length, metadata);
        length += FrameHeaderFlyweight.encodeData(mutableDirectBuffer, offset + length, data);

        return length;
    }

    public static int encode(
        final MutableDirectBuffer mutableDirectBuffer,
        final int offset,
        final int streamId,
        final int flags,
        final FrameType type,
        final ByteBuffer metadata,
        final ByteBuffer data
    ) {
        final int frameLength = computeFrameLength(type, metadata.remaining(), data.remaining());

        int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, type, streamId);

        length += FrameHeaderFlyweight.encodeMetadata(mutableDirectBuffer, offset, offset + length, metadata);
        length += FrameHeaderFlyweight.encodeData(mutableDirectBuffer, offset + length, data);

        return length;
    }

    public static int initialRequestN(final DirectBuffer directBuffer, final int offset) {
        return directBuffer.getInt(offset + INITIAL_REQUEST_N_FIELD_OFFSET, ByteOrder.BIG_ENDIAN);
    }

    public static int payloadOffset(final FrameType type, final DirectBuffer directBuffer, final int offset) {
        int result = offset + FrameHeaderFlyweight.FRAME_HEADER_LENGTH;

        if (type.hasInitialRequestN()) {
            result += BitUtil.SIZE_OF_INT;
        }

        return result;
    }
}

@@ -893,131 +893,8 @@
 %7B%7D%0A%0A
-    // TODO(lexs) Remove flag for initial N present%0A    public static final int FLAGS_REQUEST_CHANNEL_N = 0b00_0010_0000;%0A%0A
     
@@ -931,16 +931,16 @@
  offset%0A
+
     priv
@@ -1776,50 +1776,8 @@
 );%0A%0A
-        flags %7C= FLAGS_REQUEST_CHANNEL_N;%0A
     
@@ -2547,32 +2547,32 @@
 Buffer data%0A
-
     ) %7B%0A
         fina
@@ -2551,32 +2551,178 @@
 er data%0A    ) %7B%0A
+        if (type.hasInitialRequestN()) %7B%0A            throw new AssertionError(type + %22 must not be encoded without initial request N%22);%0A        %7D%0A
         final in

Make request N non-optional for REQUEST_CHANNEL (#238)