|
|
6fb0731935
|
重构：为核心组件实现 Reset 方法优化重置机制
为所有核心组件添加 Reset() 方法：
- LogWriter.Reset(): 删除并重新创建日志文件，保持 index 和 wbuf 引用不变
- RecordIndex.Reset(): 清空索引数据并重新创建索引文件
- RecordQuery.Reset(): 关闭并重新打开日志文件
- ProcessCursor.Reset(): 删除位置文件并重置游标位置
- LogTailer.Reset(): 重置内部 channel 状态
优化 TopicProcessor.Reset() 实现：
- 不再销毁和重建组件对象
- 通过调用各组件的 Reset() 方法重置状态
- 保持组件间引用关系稳定
- 减少代码行数约 20 行
- 避免空指针风险和内存分配开销
代码改进：
- LogWriter 添加 path 字段用于重置
- 移除 topic_processor.go 中未使用的 os import
- 职责分离更清晰，每个组件管理自己的重置逻辑
测试结果：
- TestTopicReset: PASS
- TestTopicResetWithPendingRecords: PASS
- 所有 TopicProcessor 相关测试通过
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
|
2025-10-04 21:58:54 +08:00 |
|
|
|
955a467248
|
优化：减少代码重复和内存分配
主要优化：
1. 提取重复代码（topic_processor.go）
- 新增 addStatusToRecords() 辅助方法
- QueryOldest 和 QueryNewest 中的状态添加逻辑重复，已提取
- 减少 38 行重复代码
2. 优化内存分配（index.go）
- 在 RecordIndex 结构体中添加可重用的 entryBuf
- Append 方法不再每次都分配 8 字节 buffer
- 高频写入场景下可显著减少 GC 压力
性能提升：
- 减少内存分配次数（每次写入索引节省 1 次分配）
- 提高代码可维护性（消除重复代码）
- 所有测试通过 ✅
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
|
2025-10-04 13:32:44 +08:00 |
|
|
|
6862de12ff
|
新增：统一的错误类型系统 (errors.go)
主要功能：
- 定义哨兵错误（Sentinel Errors）：ErrNilParameter, ErrInvalidCount,
ErrInvalidRange, ErrAlreadyRunning, ErrNotFound, ErrCRCMismatch 等
- 实现结构化错误类型：TopicError, FileError, IndexError, ValidationError
- 提供错误检查辅助函数：IsTopicNotFound, IsIndexOutOfRange, IsCRCMismatch
- 支持 errors.Is 和 errors.As 进行错误判断
更新相关文件使用新错误类型：
- cursor.go: 使用 ValidationError 和 ErrCRCMismatch
- index.go: 使用 IndexError 处理索引越界
- query.go: 使用 ValidationError 验证参数
- seqlog_manager.go: 使用 TopicError 和 ErrAlreadyRegistered
- topic_processor.go: 使用 ErrAlreadyRunning 和 ErrInvalidConfig
测试覆盖：
- errors_test.go 提供完整的错误类型测试
- 所有现有测试继续通过
使用示例：
```go
// 检查 topic 是否存在
if IsTopicNotFound(err) {
// 处理 topic 不存在的情况
}
// 检查索引越界
if IsIndexOutOfRange(err) {
var indexErr *IndexError
errors.As(err, &indexErr)
fmt.Printf("index %d out of range\n", indexErr.Index)
}
```
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
|
2025-10-04 01:56:22 +08:00 |
|
|
|
5c028a55b3
|
重构：简化查询接口
- RecordQuery.QueryOldest 和 QueryNewest 不再接收 startIdx/endIdx 参数
- 查询方法返回纯 Record 列表，状态判断移到调用方
- TopicProcessor 的查询方法负责添加状态信息
- 更新所有测试文件以适配新接口
|
2025-10-04 00:23:09 +08:00 |
|
|
|
a421ca1d85
|
重构：简化 RecordIndex 结构体
删除独立的 IndexHeader 结构体，将 magic 和 version 字段直接放入 RecordIndex，减少不必要的嵌套层级。
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
|
2025-10-03 23:55:45 +08:00 |
|
|
|
de39339620
|
重构：统一使用索引（Index）替代位置（Position）进行状态判断
## 主要变更
### 架构改进
- 明确索引（Index）与偏移（Offset）的职责分离
- Index: 记录序号（逻辑概念），用于状态判断
- Offset: 文件字节位置（物理概念），仅用于 I/O 操作
### API 变更
- 删除所有 Position 相关方法：
- `LogCursor.StartPos()/EndPos()`
- `LogTailer.GetStartPos()/GetEndPos()`
- `TopicProcessor.GetProcessingPosition()/GetReadPosition()`
- `Seqlog.GetProcessingPosition()/GetReadPosition()`
- 新增索引方法：
- `LogCursor.StartIndex()/EndIndex()`
- `LogTailer.GetStartIndex()/GetEndIndex()`
- `TopicProcessor.GetProcessingIndex()/GetReadIndex()`
- `Seqlog.GetProcessingIndex()/GetReadIndex()`
- `Seqlog.GetProcessor()` - 获取 processor 实例以访问 Index
### 查询接口变更
- `RecordQuery.QueryOldest(startIndex, count, startIdx, endIdx)` - 使用索引参数
- `RecordQuery.QueryNewest(endIndex, count, startIdx, endIdx)` - 使用索引参数
- `RecordQuery.QueryAt(position, direction, count, startIdx, endIdx)` - startIdx/endIdx 用于状态判断
### 性能优化
- 状态判断改用整数比较，不再需要计算偏移量
- 减少不必要的索引到偏移的转换
- 只在实际文件 I/O 时才获取 offset
### 测试更新
- 更新所有测试用例使用新的 Index API
- 更新示例代码（topic_processor_example.go, webapp/main.go）
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude <noreply@anthropic.com>
|
2025-10-03 23:50:53 +08:00 |
|
