[Java] Add RingBuffer.controlledRead methods. Issue #227.

agrona/src/main/java/org/agrona/concurrent/ControlledMessageHandler.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2014-2020 Real Logic Limited.
+ *
+ * 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
+ *
+ * https://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;
+
+import org.agrona.MutableDirectBuffer;
+
+/**
+ * Callback interface for processing of messages that are read from a buffer.
+ */
+@FunctionalInterface
+public interface ControlledMessageHandler
+{
+    /**
+     * Action to be taken on return from {@link #onMessage(int, MutableDirectBuffer, int, int)}.
+     */
+    enum Action
+    {
+        /**
+         * Abort the current read operation and do not advance the position for this message.
+         */
+        ABORT,
+
+        /**
+         * Break from the current read operation and commit the position as of the end of the current message
+         * being handled.
+         */
+        BREAK,
+
+        /**
+         * Continue processing but commit the position as of the end of the current message so that
+         * flow control is applied to this point.
+         */
+        COMMIT,
+
+        /**
+         * Continue processing until limit or no messages with position commit at end of read as the in
+         * {@link #onMessage(int, MutableDirectBuffer, int, int)}.
+         */
+        CONTINUE,
+    }
+
+    /**
+     * Called for the processing of each message read from a buffer in turn.
+     *
+     * @param msgTypeId type of the encoded message.
+     * @param buffer    containing the encoded message.
+     * @param index     at which the encoded message begins.
+     * @param length    in bytes of the encoded message.
+     * @return {@link Action} to be taken to control how the read progresses.
+     */
+    Action onMessage(int msgTypeId, MutableDirectBuffer buffer, int index, int length);
+}

agrona/src/main/java/org/agrona/concurrent/ringbuffer/ManyToOneRingBuffer.java

@@ -17,10 +17,10 @@
 
 import org.agrona.DirectBuffer;
 import org.agrona.UnsafeAccess;
-import org.agrona.concurrent.AtomicBuffer;
-import org.agrona.concurrent.MessageHandler;
+import org.agrona.concurrent.*;
 
 import static org.agrona.BitUtil.align;
+import static org.agrona.concurrent.ControlledMessageHandler.Action.*;
 import static org.agrona.concurrent.ringbuffer.RecordDescriptor.*;
 import static org.agrona.concurrent.ringbuffer.RingBufferDescriptor.*;
 
@@ -98,6 +98,55 @@ public boolean write(final int msgTypeId, final DirectBuffer srcBuffer, final in
         return true;
     }
 
+    /**
+     * {@inheritDoc}
+     */
+    public int tryClaim(final int msgTypeId, final int length)
+    {
+        checkTypeId(msgTypeId);
+        checkMsgLength(length);
+
+        final AtomicBuffer buffer = this.buffer;
+        final int recordLength = length + HEADER_LENGTH;
+        final int recordIndex = claimCapacity(buffer, recordLength);
+
+        if (INSUFFICIENT_CAPACITY == recordIndex)
+        {
+            return recordIndex;
+        }
+
+        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
+        UnsafeAccess.UNSAFE.storeFence();
+        buffer.putInt(typeOffset(recordIndex), msgTypeId);
+
+        return encodedMsgOffset(recordIndex);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void commit(final int index)
+    {
+        final int recordIndex = computeRecordIndex(index);
+        final AtomicBuffer buffer = this.buffer;
+        final int recordLength = verifyClaimedSpaceNotReleased(buffer, recordIndex);
+
+        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public void abort(final int index)
+    {
+        final int recordIndex = computeRecordIndex(index);
+        final AtomicBuffer buffer = this.buffer;
+        final int recordLength = verifyClaimedSpaceNotReleased(buffer, recordIndex);
+
+        buffer.putInt(typeOffset(recordIndex), PADDING_MSG_TYPE_ID);
+        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
+    }
+
     /**
      * {@inheritDoc}
      */
@@ -141,13 +190,94 @@ public int read(final MessageHandler handler, final int messageCountLimit)
                     continue;
                 }
 
-                ++messagesRead;
                 handler.onMessage(messageTypeId, buffer, recordIndex + HEADER_LENGTH, recordLength - HEADER_LENGTH);
+                ++messagesRead;
+            }
+        }
+        finally
+        {
+            if (bytesRead > 0)
+            {
+                buffer.setMemory(headIndex, bytesRead, (byte)0);
+                buffer.putLongOrdered(headPositionIndex, head + bytesRead);
+            }
+        }
+
+        return messagesRead;
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public int controlledRead(final ControlledMessageHandler handler)
+    {
+        return controlledRead(handler, Integer.MAX_VALUE);
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public int controlledRead(final ControlledMessageHandler handler, final int messageCountLimit)
+    {
+        int messagesRead = 0;
+
+        final AtomicBuffer buffer = this.buffer;
+        final int headPositionIndex = this.headPositionIndex;
+        long head = buffer.getLong(headPositionIndex);
+
+        final int capacity = this.capacity;
+        int headIndex = (int)head & (capacity - 1);
+        final int maxBlockLength = capacity - headIndex;
+        int bytesRead = 0;
+
+        try
+        {
+            while ((bytesRead < maxBlockLength) && (messagesRead < messageCountLimit))
+            {
+                final int recordIndex = headIndex + bytesRead;
+                final int recordLength = buffer.getIntVolatile(lengthOffset(recordIndex));
+                if (recordLength <= 0)
+                {
+                    break;
+                }
+
+                final int alignedLength = align(recordLength, ALIGNMENT);
+                bytesRead += alignedLength;
+
+                final int messageTypeId = buffer.getInt(typeOffset(recordIndex));
+                if (PADDING_MSG_TYPE_ID == messageTypeId)
+                {
+                    continue;
+                }
+
+                final ControlledMessageHandler.Action action = handler.onMessage(
+                    messageTypeId, buffer, recordIndex + HEADER_LENGTH, recordLength - HEADER_LENGTH);
+
+                if (ABORT == action)
+                {
+                    bytesRead -= alignedLength;
+                    break;
+                }
+
+                ++messagesRead;
+
+                if (BREAK == action)
+                {
+                    break;
+                }
+                if (COMMIT == action)
+                {
+                    buffer.setMemory(headIndex, bytesRead, (byte)0);
+                    buffer.putLongOrdered(headPositionIndex, head + bytesRead);
+                    headIndex += bytesRead;
+                    head += bytesRead;
+                    bytesRead = 0;
+                }
             }
         }
         finally
         {
-            if (bytesRead != 0)
+            if (bytesRead > 0)
             {
                 buffer.setMemory(headIndex, bytesRead, (byte)0);
                 buffer.putLongOrdered(headPositionIndex, head + bytesRead);
@@ -302,55 +432,6 @@ else if (0 == length)
         return unblocked;
     }
 
-    /**
-     * {@inheritDoc}
-     */
-    public int tryClaim(final int msgTypeId, final int length)
-    {
-        checkTypeId(msgTypeId);
-        checkMsgLength(length);
-
-        final AtomicBuffer buffer = this.buffer;
-        final int recordLength = length + HEADER_LENGTH;
-        final int recordIndex = claimCapacity(buffer, recordLength);
-
-        if (INSUFFICIENT_CAPACITY == recordIndex)
-        {
-            return recordIndex;
-        }
-
-        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
-        UnsafeAccess.UNSAFE.storeFence();
-        buffer.putInt(typeOffset(recordIndex), msgTypeId);
-
-        return encodedMsgOffset(recordIndex);
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public void commit(final int index)
-    {
-        final int recordIndex = computeRecordIndex(index);
-        final AtomicBuffer buffer = this.buffer;
-        final int recordLength = verifyClaimedSpaceNotReleased(buffer, recordIndex);
-
-        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
-    }
-
-    /**
-     * {@inheritDoc}
-     */
-    public void abort(final int index)
-    {
-        final int recordIndex = computeRecordIndex(index);
-        final AtomicBuffer buffer = this.buffer;
-        final int recordLength = verifyClaimedSpaceNotReleased(buffer, recordIndex);
-
-        buffer.putInt(typeOffset(recordIndex), PADDING_MSG_TYPE_ID);
-        buffer.putIntOrdered(lengthOffset(recordIndex), -recordLength);
-    }
-
     private static boolean scanBackToConfirmStillZeroed(final AtomicBuffer buffer, final int from, final int limit)
     {
         int i = from - ALIGNMENT;
@@ -378,7 +459,7 @@ private void checkMsgLength(final int length)
         else if (length > maxMsgLength)
         {
             throw new IllegalArgumentException(
-                "encoded message exceeds maxMsgLength of " + maxMsgLength + ", length=" + length);
+                "encoded message exceeds maxMsgLength=" + maxMsgLength + ", length=" + length);
         }
     }