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;
    }
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
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 .