/*
 * Copyright 2014 - 2016 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 org.agrona.concurrent.broadcast;

import org.agrona.MutableDirectBuffer;
import org.agrona.concurrent.MessageHandler;
import org.agrona.concurrent.UnsafeBuffer;

/**
 * Receiver that copies messages that have been broadcast to enable a simpler API for the client.
 */
public class CopyBroadcastReceiver
{
    private static final int SCRATCH_BUFFER_SIZE = 4096;

    private final BroadcastReceiver receiver;
    private final MutableDirectBuffer scratchBuffer;

    /**
     * Wrap a {@link BroadcastReceiver} to simplify the API for receiving messages.
     *
     * @param receiver to be wrapped.
     */
    public CopyBroadcastReceiver(final BroadcastReceiver receiver)
    {
        this.receiver = receiver;
        scratchBuffer = new UnsafeBuffer(new byte[SCRATCH_BUFFER_SIZE]);

        while (receiver.receiveNext())
        {
            // If we're reconnecting to a broadcast buffer then we need to
            // scan ourselves up to date, otherwise we risk "falling behind"
            // the buffer due to the time taken to catchup.
        }
    }

    /**
     * Receive one message from the broadcast buffer.
     *
     * @param handler to be called for each message received.
     * @return the number of messages that have been received.
     */
    public int receive(final MessageHandler handler)
    {
        int messagesReceived = 0;
        final BroadcastReceiver receiver = this.receiver;
        final long lastSeenLappedCount = receiver.lappedCount();

        if (receiver.receiveNext())
        {
            if (lastSeenLappedCount != receiver.lappedCount())
            {
                throw new IllegalStateException("Unable to keep up with broadcast buffer");
            }

            final int length = receiver.length();
            final int capacity = scratchBuffer.capacity();
            if (length > capacity)
            {
                throw new IllegalStateException(String.format("Buffer required size %d but only has %d", length, capacity));
            }

            final int msgTypeId = receiver.typeId();
            scratchBuffer.putBytes(0, receiver.buffer(), receiver.offset(), length);

            if (!receiver.validate())
            {
                throw new IllegalStateException("Unable to keep up with broadcast buffer");
            }

            handler.onMessage(msgTypeId, scratchBuffer, 0, length);

            messagesReceived = 1;
        }

        return messagesReceived;
    }
}

@@ -641,47 +641,8 @@
 t;%0A%0A
-import org.agrona.MutableDirectBuffer;%0A
 impo
@@ -726,16 +726,45 @@
 uffer;%0A%0A
+import java.nio.ByteBuffer;%0A%0A
 /**%0A * R
@@ -902,23 +902,110 @@
 r%0A%7B%0A    
-private
+/**%0A     * Default length for the scratch buffer for copying messages into.%0A     */%0A    public
  static 
@@ -1029,20 +1029,22 @@
 _BUFFER_
-SIZE
+LENGTH
  = 4096;
@@ -1113,21 +1113,14 @@
 nal 
-MutableDirect
+Unsafe
 Buff
@@ -1260,16 +1260,27 @@
 eceiver 
+           
 to be wr
@@ -1282,24 +1282,122 @@
 be wrapped.%0A
+     * @param scratchBufferLength is the maximum length of a message to be copied when receiving.%0A
      */%0A    
@@ -1457,16 +1457,47 @@
 receiver
+, final int scratchBufferLength
 )%0A    %7B%0A
@@ -1575,37 +1575,54 @@
 fer(
-new byte%5BSCRATCH_BUFFER_SIZE%5D
+ByteBuffer.allocateDirect(scratchBufferLength)
 );%0A%0A
@@ -1895,24 +1895,296 @@
    %7D%0A    %7D%0A%0A
+    /**%0A     * Wrap a %7B@link BroadcastReceiver%7D to simplify the API for receiving messages.%0A     *%0A     * @param receiver to be wrapped.%0A     */%0A    public CopyBroadcastReceiver(final BroadcastReceiver receiver)%0A    %7B%0A        this(receiver, SCRATCH_BUFFER_LENGTH);%0A    %7D%0A%0A
     /**%0A    
@@ -3061,12 +3061,17 @@
 red 
-size
+length of
  %25d 

Add the ability to change the scratch buffer length in the CopyBroadcastReceiver and have it use an off heap buffer.