Aeron Subscription.controlledPoll

@FunctionalInterface
public interface ControlledFragmentHandler {

    /**
     * 处理器返回后要采取的动作
     */
    enum Action {
        // 中止当前轮询，不提交当前消息的位置
        ABORT,

        // 中断当前轮询，提交到当前消息为止的位置
        BREAK,

        // 立即提交位置，然后继续处理后续消息
        COMMIT,

        // 继续处理，最后统一提交位置
        CONTINUE
    }

    /**
     * 处理从日志中读取的数据片段的回调方法
     *
     * 注意：在此回调中禁止重入调用 Aeron 客户端，否则会导致未定义行为
     *
     * @param buffer 包含数据的缓冲区
     * @param offset 数据开始的偏移量
     * @param length 数据的字节长度
     * @param header 包含数据元信息的头
     * @return 处理完成后关于流位置应采取的动作
     */
    Action onFragment(
        DirectBuffer buffer,
        int offset,
        int length,
        Header header
    );
}

public Action onFragment(
        final DirectBuffer buffer,
        final int offset,
        final int length,
        final Header header) {

    // 获取帧的标志位
    final byte flags = header.flags();
    Action action = Action.CONTINUE;

    // ==================== 情况1: 非分片消息 ====================
    // 当 BEGIN 和 END 标志都设置时，表示这是完整的消息
    if ((flags & UNFRAGMENTED) == UNFRAGMENTED) {
        // 直接委托给业务处理器，不需要组装
        action = delegate.onFragment(buffer, offset, length, header);
    }
    // ==================== 情况2: 分片消息的第一个片段 ====================
    else if ((flags & BEGIN_FRAG_FLAG) == BEGIN_FRAG_FLAG) {
        // 获取或创建该 session 的 BufferBuilder
        final BufferBuilder builder = getBufferBuilder(header.sessionId());

        // 重置并开始收集新的分片消息
        builder.reset()                               // 清空之前的数据
               .captureHeader(header)                 // 保存第一片的 Header
               .append(buffer, offset, length)       // 追加第一片数据
               .nextTermOffset(header.nextTermOffset()); // 记录下一片的预期位置
    }
    // ==================== 情况3: 中间片段或最后片段 ====================
    else {
        // 尝试获取已存在的 BufferBuilder
        final BufferBuilder builder = builderBySessionIdMap.get(header.sessionId());

        if (null != builder) {
            // 验证片段的连续性（防止乱序或丢失）
            if (header.termOffset() == builder.nextTermOffset()) {
                // 保存追加前的位置（用于 ABORT 回滚）
                final int limit = builder.limit();

                // 追加当前片段
                builder.append(buffer, offset, length);

                // 检查是否为最后一个片段
                if ((flags & END_FRAG_FLAG) == END_FRAG_FLAG) {
                    // ===== 消息组装完成，调用业务处理器 =====
                    action = delegate.onFragment(
                        builder.buffer(),         // 完整消息的缓冲区
                        0,                        // 从头开始
                        builder.limit(),          // 完整消息的长度
                        builder.completeHeader(header)  // 组合后的 Header
                    );

                    // 根据处理结果决定是否回滚
                    if (Action.ABORT == action) {
                        // ABORT: 回滚到追加前的状态，下次重新组装
                        builder.limit(limit);
                    } else {
                        // 其他 Action: 清空缓冲区，准备下一条消息
                        builder.reset();
                    }
                } else {
                    // 还不是最后片段，记录下一片的预期位置
                    builder.nextTermOffset(header.nextTermOffset());
                }
            } else {
                // 片段不连续（乱序或丢失），丢弃已收集的数据
                builder.reset();
            }
        }
    }

    return action;
}

/**
 * 根据位置计算 Term 索引（0, 1, 或 2）
 * 使用位运算实现高效的取模操作
 */
public static int indexByPosition(
        final long position,
        final int positionBitsToShift) {

    // position >>> positionBitsToShift 相当于 position / termLength
    // % PARTITION_COUNT 取模得到索引 (0, 1, 2)
    return (int) ((position >>> positionBitsToShift) % PARTITION_COUNT);
}

/**
 * 计算绝对位置
 * position = (termId - initialTermId) * termLength + termOffset
 */
public static long computePosition(
        final int activeTermId,
        final int termOffset,
        final int positionBitsToShift,
        final int initialTermId) {

    // 计算已经过的 Term 数量（支持 termId 溢出后的回绕）
    final long termCount = activeTermId - initialTermId;

    // termCount << positionBitsToShift 相当于 termCount * termLength
    return (termCount << positionBitsToShift) + termOffset;
}

Subscription.controlledPoll(handler, limit)
│
├── Image.controlledPoll(handler, limit)
│     │
│     ├── subscriberPosition.get()
│     │
│     ├── activeTermBuffer(position)
│     │     └── LogBufferDescriptor.indexByPosition()
│     │
│     └── [LOOP] 读取消息帧
│           │
│           ├── FrameDescriptor.frameLengthVolatile()
│           ├── BitUtil.align()
│           ├── FrameDescriptor.isPaddingFrame()
│           ├── Header.offset()
│           │
│           ├── handler.onFragment()
│           │     │
│           │     ├── [UNFRAGMENTED] delegate.onFragment()
│           │     │
│           │     └── [FRAGMENTED]
│           │           ├── BufferBuilder.reset()
│           │           ├── BufferBuilder.append()
│           │           └── [END] delegate.onFragment()
│           │
│           └── [处理 Action]
│                 ├── ABORT: 回滚 offset
│                 ├── BREAK: 退出循环
│                 ├── COMMIT: subscriberPosition.setRelease()
│                 └── CONTINUE: 继续
│
└── [FINALLY] subscriberPosition.setRelease()