/*
 * Copyright 2014 Real Logic Ltd.
 *
 * 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 uk.co.real_logic.aeron.common.concurrent.logbuffer;

import uk.co.real_logic.aeron.common.BitUtil;
import uk.co.real_logic.aeron.common.concurrent.AtomicBuffer;

import java.nio.ByteOrder;

import static java.nio.ByteOrder.LITTLE_ENDIAN;
import static uk.co.real_logic.aeron.common.concurrent.logbuffer.FrameDescriptor.*;
import static uk.co.real_logic.aeron.common.concurrent.logbuffer.LogBufferDescriptor.HIGH_WATER_MARK_OFFSET;
import static uk.co.real_logic.aeron.common.concurrent.logbuffer.LogBufferDescriptor.TAIL_COUNTER_OFFSET;

/**
 * Rebuild a log buffer based on incoming frames that can be out-of-order.
 *
 * <b>Note:</b> Only one rebuilder should rebuild a log at any given time. This is not thread safe
 * by rebuilder instance or across rebuilder instances.
 */
public class LogRebuilder extends LogBuffer
{
    /**
     * Construct a rebuilder over a log and state buffer.
     *
     * @param logBuffer containing the sequence of frames.
     * @param stateBuffer containing the state of the rebuild process.
     */
    public LogRebuilder(final AtomicBuffer logBuffer, final AtomicBuffer stateBuffer)
    {
        super(logBuffer, stateBuffer);
    }

    /**
     * Set the starting tail offset.
     *
     * @param offset to start tail at
     */
    public void tail(final int offset)
    {
        putTailOrdered(stateBuffer(), offset);
    }

    /**
     * Is the rebuild of this log complete?
     *
     * @return true if it is complete otherwise false.
     */
    public boolean isComplete()
    {
        return stateBuffer().getIntVolatile(TAIL_COUNTER_OFFSET) >= capacity();
    }

    /**
     * Insert a packet of frames into the log at the appropriate offset as indicated by the term offset header.
     *
     * The tail and high-water-mark will be updated as appropriate. Data can be consumed up to the the tail. The
     * high-water-mark can be used to detect loss.
     *
     * @param packet containing a sequence of frames.
     * @param srcOffset in the packet at which the frames begin.
     * @param length of the sequence of frames in bytes.
     */
    public void insert(final AtomicBuffer packet, final int srcOffset, final int length)
    {
        final int termOffset = packet.getInt(termOffsetOffset(srcOffset), LITTLE_ENDIAN);
        int tail = tail();

        if (termOffset >= tail)
        {
            final int lengthOffset = lengthOffset(srcOffset);
            final int frameLength = packet.getInt(lengthOffset, LITTLE_ENDIAN);
            packet.putInt(lengthOffset, 0, LITTLE_ENDIAN);

            final AtomicBuffer logBuffer = logBuffer();
            logBuffer.putBytes(termOffset, packet, srcOffset, length);
            FrameDescriptor.frameLengthOrdered(logBuffer, termOffset, frameLength);

            updateCompetitionStatus(logBuffer, termOffset, length, tail);
        }
    }

    private void updateCompetitionStatus(final AtomicBuffer logBuffer, final int termOffset, final int length, int tail)
    {
        final int capacity = capacity();
        int alignedFrameLength;
        while ((tail < capacity) && (alignedFrameLength = alignedFrameLength(logBuffer, tail)) != 0)
        {
            tail += alignedFrameLength;
        }

        final AtomicBuffer stateBuffer = stateBuffer();
        putTailOrdered(stateBuffer, tail);

        final int endOfFrame = termOffset + length;
        if (endOfFrame > highWaterMark())
        {
            putHighWaterMark(stateBuffer, endOfFrame);
        }
    }

    private static int alignedFrameLength(final AtomicBuffer logBuffer, final int tail)
    {
        return BitUtil.align(logBuffer.getInt(lengthOffset(tail), ByteOrder.LITTLE_ENDIAN), FRAME_ALIGNMENT);
    }

    private static void putTailOrdered(final AtomicBuffer stateBuffer, final int tail)
    {
        stateBuffer.putIntOrdered(TAIL_COUNTER_OFFSET, tail);
    }

    private static void putHighWaterMark(final AtomicBuffer stateBuffer, final int highWaterMark)
    {
        stateBuffer.putIntOrdered(HIGH_WATER_MARK_OFFSET, highWaterMark);
    }
}

@@ -3632,32 +3632,25 @@
         int 
-alignedF
+f
 rameLength;%0A
@@ -3686,24 +3686,17 @@
 ty) && (
-alignedF
+f
 rameLeng
@@ -3704,60 +3704,164 @@
 h = 
-alignedFrameLength(logBuffer, tail)) != 0)%0A        %7B
+logBuffer.getInt(lengthOffset(tail), LITTLE_ENDIAN)) != 0)%0A        %7B%0A            final int alignedFrameLength = BitUtil.align(frameLength, FRAME_ALIGNMENT);
 %0A   
@@ -4188,219 +4188,8 @@
  %7D%0A%0A
-    private static int alignedFrameLength(final AtomicBuffer logBuffer, final int tail)%0A    %7B%0A        return BitUtil.align(logBuffer.getInt(lengthOffset(tail), ByteOrder.LITTLE_ENDIAN), FRAME_ALIGNMENT);%0A    %7D%0A%0A
     

Simplify logic for alignment.