aeron-client/src/main/java/io/aeron/Image.java
public int controlledPoll(final ControlledFragmentHandler handler, final int fragmentLimit)
    {
        if (isClosed)
        {
            return 0;
        }

        int fragmentsRead = 0;
        long initialPosition = subscriberPosition.get();
        int initialOffset = (int)initialPosition & termLengthMask;
        int resultingOffset = initialOffset;
        final UnsafeBuffer termBuffer = activeTermBuffer(initialPosition);
        final int capacity = termBuffer.capacity();
        final Header header = this.header;
        header.buffer(termBuffer);

        try
        {
            while (fragmentsRead < fragmentLimit && resultingOffset < capacity)
            {
                final int length = frameLengthVolatile(termBuffer, resultingOffset);
                if (length <= 0)
                {
                    break;
                }

                final int frameOffset = resultingOffset;
                final int alignedLength = BitUtil.align(length, FRAME_ALIGNMENT);
                resultingOffset += alignedLength;

                if (isPaddingFrame(termBuffer, frameOffset))
                {
                    continue;
                }

                header.offset(frameOffset);

                final Action action = handler.onFragment(
                    termBuffer, frameOffset + HEADER_LENGTH, length - HEADER_LENGTH, header);

                if (action == ABORT)
                {
                    resultingOffset -= alignedLength;
                    break;
                }

                ++fragmentsRead;

                if (action == BREAK)
                {
                    break;
                }
                else if (action == COMMIT)
                {
                    initialPosition += (resultingOffset - initialOffset);
                    initialOffset = resultingOffset;
                    subscriberPosition.setOrdered(initialPosition);
                }
            }
        }
        catch (final Throwable t)
        {
            errorHandler.onError(t);
        }
        finally
        {
            final long resultingPosition = initialPosition + (resultingOffset - initialOffset);
            if (resultingPosition > initialPosition)
            {
                subscriberPosition.setOrdered(resultingPosition);
            }
        }

        return fragmentsRead;
    }
public int controlledPoll(final ControlledFragmentHandler handler, final int fragmentLimit)
    {
        if (isClosed)
        {
            return 0;
        }

        int fragmentsRead = 0;
        long initialPosition = subscriberPosition.get();
        int initialOffset = (int)initialPosition & termLengthMask;
        int resultingOffset = initialOffset;
        final UnsafeBuffer termBuffer = activeTermBuffer(initialPosition);
        final int capacity = termBuffer.capacity();
        final Header header = this.header;
        header.buffer(termBuffer);

        try
        {
            while (fragmentsRead < fragmentLimit && resultingOffset < capacity)
            {
                final int length = frameLengthVolatile(termBuffer, resultingOffset);
                if (length <= 0)
                {
                    break;
                }

                final int frameOffset = resultingOffset;
                final int alignedLength = BitUtil.align(length, FRAME_ALIGNMENT);
                resultingOffset += alignedLength;

                if (isPaddingFrame(termBuffer, frameOffset))
                {
                    continue;
                }

                header.offset(frameOffset);

                final Action action = handler.onFragment(
                    termBuffer, frameOffset + HEADER_LENGTH, length - HEADER_LENGTH, header);

                if (action == ABORT)
                {
                    resultingOffset -= alignedLength;
                    break;
                }

                ++fragmentsRead;

                if (action == BREAK)
                {
                    break;
                }
                else if (action == COMMIT)
                {
                    initialPosition += (resultingOffset - initialOffset);
                    initialOffset = resultingOffset;
                    subscriberPosition.setOrdered(initialPosition);
                }
            }
        }
        catch (final Throwable t)
        {
            errorHandler.onError(t);
        }
        finally
        {
            final long resultingPosition = initialPosition + (resultingOffset - initialOffset);
            if (resultingPosition > initialPosition)
            {
                subscriberPosition.setOrdered(resultingPosition);
            }
        }

        return fragmentsRead;
    }
Poll for new messages in a stream. If new messages are found beyond the last consumed position then they
will be delivered to the {@link ControlledFragmentHandler} up to a limited number of fragments as specified.
<p>
Use a {@link ControlledFragmentAssembler} to assemble messages which span multiple fragments.

@param handler       to which message fragments are delivered.
@param fragmentLimit for the number of fragments to be consumed during one polling operation.
@return the number of fragments that have been consumed.
@see ControlledFragmentAssembler
@see ImageControlledFragmentAssembler