bourdon/seqlog
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构：重命名核心组件并增强查询功能

Browse Source 
主要更改：

1. 核心重命名
   - Seqlog -> LogHub (更准确地反映其作为日志中枢的角色)
   - NewSeqlog() -> NewLogHub()
   - LogCursor -> ProcessCursor (更准确地反映其用于处理场景)
   - seqlog_manager.go -> loghub.go (文件名与结构体名对应)

2. TopicProcessor.Reset 增强
   - 如果正在运行且没有待处理的日志，会自动停止后重置
   - 如果有待处理的日志，返回详细错误（显示已处理/总记录数）
   - 简化了 LogHub.ResetTopic，移除显式 Stop 调用

3. 新增查询方法
   - TopicProcessor.QueryFromFirst(count) - 从第一条记录向索引递增方向查询
   - TopicProcessor.QueryFromLast(count) - 从最后一条记录向索引递减方向查询
   - LogHub.QueryFromFirst(topic, count)
   - LogHub.QueryFromLast(topic, count)

4. 测试覆盖
   - 添加 query_test.go - QueryFromProcessing 测试
   - 添加 TestQueryFromFirstAndLast - TopicProcessor 查询测试
   - 添加 TestLogHubQueryFromFirstAndLast - LogHub 查询测试
   - 添加 TestTopicResetWithPendingRecords - Reset 增强功能测试

5. 示例代码
   - 添加 example/get_record/ - 演示 QueryFromProcessing 用法
   - 更新所有示例以使用 LogHub 和新 API

所有测试通过 ✅

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
bourdon
2025-10-04 13:26:21 +08:00
parent dfdc27c67f
commit 90cc9e21c9
14 changed files with 795 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
cursor.go
View File

@@ -10,8 +10,8 @@ import (
"github.com/google/uuid"
)
// LogCursor 日志游标（窗口模式）
type LogCursor struct {
// ProcessCursor 日志游标（窗口模式）
type ProcessCursor struct {
fd *os.File
rbuf []byte // 8 MiB 复用
path string // 日志文件路径
@@ -23,7 +23,7 @@ type LogCursor struct {
// NewCursor 创建一个新的日志游标
// index: 外部提供的索引管理器，用于快速定位记录
func NewCursor(path string, index *RecordIndex) (*LogCursor, error) {
func NewCursor(path string, index *RecordIndex) (*ProcessCursor, error) {
if index == nil {
return nil, NewValidationError("index", "index cannot be nil", ErrNilParameter)
}
@@ -32,7 +32,7 @@ func NewCursor(path string, index *RecordIndex) (*LogCursor, error) {
if err != nil {
return nil, err
}
c := &LogCursor{
c := &ProcessCursor{
fd: fd,
rbuf: make([]byte, 8<<20),
path: path,
@@ -47,12 +47,12 @@ func NewCursor(path string, index *RecordIndex) (*LogCursor, error) {
}
// Seek 到任意 offset（支持重启续传）
func (c *LogCursor) Seek(offset int64, whence int) (int64, error) {
func (c *ProcessCursor) Seek(offset int64, whence int) (int64, error) {
return c.fd.Seek(offset, whence)
}
// Next 读取下一条记录（使用索引快速定位）
func (c *LogCursor) Next() (*Record, error) {
func (c *ProcessCursor) Next() (*Record, error) {
// 检查是否超出索引范围
if c.endIdx >= c.index.Count() {
return nil, io.EOF
@@ -110,7 +110,7 @@ func (c *LogCursor) Next() (*Record, error) {
// NextRange 读取指定数量的记录（范围游动）
// count: 要读取的记录数量
// 返回：读取到的记录列表，如果到达文件末尾，返回的记录数可能少于 count
func (c *LogCursor) NextRange(count int) ([]*Record, error) {
func (c *ProcessCursor) NextRange(count int) ([]*Record, error) {
if count <= 0 {
return nil, NewValidationError("count", "count must be greater than 0", ErrInvalidCount)
}
@@ -133,34 +133,34 @@ func (c *LogCursor) NextRange(count int) ([]*Record, error) {
}
// Commit 提交窗口，将 endIdx 移动到 startIdx（表示已处理完这批记录）
func (c *LogCursor) Commit() {
func (c *ProcessCursor) Commit() {
c.startIdx = c.endIdx
}
// Rollback 回滚窗口，将 endIdx 回退到 startIdx（表示放弃这批记录的处理）
func (c *LogCursor) Rollback() error {
func (c *ProcessCursor) Rollback() error {
c.endIdx = c.startIdx
return nil
}
// StartIndex 获取窗口开始索引
func (c *LogCursor) StartIndex() int {
func (c *ProcessCursor) StartIndex() int {
return c.startIdx
}
// EndIndex 获取窗口结束索引
func (c *LogCursor) EndIndex() int {
func (c *ProcessCursor) EndIndex() int {
return c.endIdx
}
// Close 关闭游标并保存位置
func (c *LogCursor) Close() error {
func (c *ProcessCursor) Close() error {
c.savePosition()
return c.fd.Close()
}
// savePosition 保存当前读取位置到文件
func (c *LogCursor) savePosition() error {
func (c *ProcessCursor) savePosition() error {
f, err := os.Create(c.posFile)
if err != nil {
return err
@@ -175,7 +175,7 @@ func (c *LogCursor) savePosition() error {
}
// loadPosition 从文件加载上次的读取位置
func (c *LogCursor) loadPosition() error {
func (c *ProcessCursor) loadPosition() error {
f, err := os.Open(c.posFile)
if err != nil {
if os.IsNotExist(err) {

4
example/README.md
View File

@@ -192,10 +192,10 @@ for i := 0; i < 20; i++ {
**1. "no such file or directory" 错误**
确保在创建 Seqlog 之前先创建目录：
确保在创建 LogHub 之前先创建目录：
```go
os.MkdirAll("log_dir", 0755)
seq := seqlog.NewSeqlog("log_dir", logger, nil)
seq := seqlog.NewLogHub("log_dir", logger, nil)
```
**2. 查询时出现 EOF 错误**

2
example/concurrent/main.go
View File

@@ -40,7 +40,7 @@ func main() {
}
// 创建 Seqlog 实例（默认处理器）
seq := seqlog.NewSeqlog(testDir, logger, nil)
seq := seqlog.NewLogHub(testDir, logger, nil)
topics := []string{"app", "access", "error"}
for _, topic := range topics {

74
example/get_record/main.go Normal file
View File

@@ -0,0 +1,74 @@
package main
import (
"fmt"
"log"
"time"
"code.tczkiot.com/seqlog"
)
func main() {
// 创建 Seqlog 实例
mgr := seqlog.NewLogHub("./logs", nil, nil)
// 注册 topic handler
processedCount := 0
err := mgr.RegisterHandler("app", func(record *seqlog.Record) error {
processedCount++
fmt.Printf("处理记录 #%d: %s\n", processedCount, string(record.Data))
time.Sleep(100 * time.Millisecond) // 模拟处理耗时
return nil
})
if err != nil {
log.Fatal(err)
}
// 写入一些记录
fmt.Println("=== 写入记录 ===")
for i := 0; i < 10; i++ {
data := fmt.Sprintf("日志消息 #%d", i)
offset, err := mgr.Write("app", []byte(data))
if err != nil {
log.Fatal(err)
}
fmt.Printf("写入: offset=%d, data=%s\n", offset, data)
}
// 启动处理
fmt.Println("\n=== 启动日志处理 ===")
err = mgr.Start()
if err != nil {
log.Fatal(err)
}
// 等待一段时间让处理器处理一些记录
time.Sleep(500 * time.Millisecond)
// 查询当前处理窗口的记录
fmt.Println("\n=== 查询当前处理窗口记录 ===")
records, err := mgr.QueryFromProcessing("app", 5)
if err != nil {
log.Fatal(err)
}
fmt.Printf("从处理窗口开始位置查询到 %d 条记录:\n", len(records))
for _, rec := range records {
fmt.Printf(" [索引 %d] %s - 状态: %s\n", rec.Index, string(rec.Record.Data), rec.Status)
}
// 查询更多记录
fmt.Println("\n=== 查询后续记录 ===")
moreRecords, err := mgr.QueryFromProcessing("app", 10)
if err != nil {
log.Fatal(err)
}
fmt.Printf("查询到 %d 条记录:\n", len(moreRecords))
for _, rec := range moreRecords {
fmt.Printf(" [索引 %d] %s - 状态: %s\n", rec.Index, string(rec.Record.Data), rec.Status)
}
// 清理
mgr.Stop()
fmt.Println("\n=== 示例完成 ===")
}

20
example/index/main.go
View File

@@ -70,25 +70,25 @@ func main() {
fmt.Printf("从第 %d 条记录开始查询\n", startIndex)
// 向索引递减方向查询（查询更早的记录）
// QueryNewest(4, 3) 查询索引 2, 3, 4，返回按索引递增排序
backward, err := query.QueryNewest(startIndex-1, 3)
// QueryOldest(5, 3) 查询索引 2, 3, 4（不包含 5），返回按索引递增排序
backward, err := query.QueryOldest(startIndex, 3)
if err != nil {
log.Fatal(err)
}
fmt.Printf("向索引递减方向查询 3 条记录（索引 2-4）:\n")
for i, rec := range backward {
fmt.Printf(" [%d] 数据=%s\n", i, string(rec.Data))
fmt.Printf("向索引递减方向查询 3 条记录:\n")
for _, rec := range backward {
fmt.Printf(" [索引 %d] 数据=%s\n", rec.Index, string(rec.Record.Data))
}
// 向索引递增方向查询（查询更新的记录）
// QueryOldest(5, 3) 查询索引 5, 6, 7，返回按索引递增排序
forward, err := query.QueryOldest(startIndex, 3)
// QueryNewest(5, 3) 查询索引 6, 7, 8（不包含 5），返回按索引递增排序
forward, err := query.QueryNewest(startIndex, 3)
if err != nil {
log.Fatal(err)
}
fmt.Printf("向索引递增方向查询 3 条记录（索引 5-7）:\n")
for i, rec := range forward {
fmt.Printf(" [%d] 数据=%s\n", i, string(rec.Data))
fmt.Printf("向索引递增方向查询 3 条记录:\n")
for _, rec := range forward {
fmt.Printf(" [索引 %d] 数据=%s\n", rec.Index, string(rec.Record.Data))
}
fmt.Println()

44
example/topic_processor/main.go
View File

@@ -40,58 +40,48 @@ func main() {
count := tp.GetRecordCount()
fmt.Printf(" 总共 %d 条记录\n\n", count)
// ===== 3. 获取索引 =====
fmt.Println("3. 使用索引：")
index := tp.Index()
fmt.Printf(" 索引记录数: %d\n", index.Count())
fmt.Printf(" 最后偏移: %d\n\n", index.LastOffset())
// ===== 3. 获取记录数 =====
fmt.Println("3. 查看记录统计：")
totalCount := tp.GetRecordCount()
fmt.Printf(" 记录总数: %d\n\n", totalCount)
// ===== 4. 使用查询器查询 =====
fmt.Println("4. 查询记录：")
// 查询最老的 3 条记录（从索引 0 开始）
oldest, err := tp.QueryOldest(0, 3)
oldest, err := tp.QueryOldest(3, 3)
if err != nil {
log.Fatal(err)
}
fmt.Println(" 查询最老的 3 条:")
for i, rws := range oldest {
fmt.Printf(" [%d] 状态=%s, 数据=%s\n", i, rws.Status, string(rws.Record.Data))
fmt.Printf(" [%d] 索引=%d, 状态=%s, 数据=%s\n", i, rws.Index, rws.Status, string(rws.Record.Data))
}
// 查询最新的 2 条记录（从最后一条开始）
totalCount := tp.GetRecordCount()
newest, err := tp.QueryNewest(totalCount-1, 2)
newest, err := tp.QueryNewest(totalCount-3, 2)
if err != nil {
log.Fatal(err)
}
fmt.Println(" 查询最新的 2 条:")
for i, rws := range newest {
fmt.Printf(" [%d] 状态=%s, 数据=%s\n", i, rws.Status, string(rws.Record.Data))
fmt.Printf(" [%d] 索引=%d, 状态=%s, 数据=%s\n", i, rws.Index, rws.Status, string(rws.Record.Data))
}
fmt.Println()
// ===== 5. 使用游标读取 =====
fmt.Println("5. 使用游标读取：")
cursor, err := tp.Cursor()
// ===== 5. 从处理窗口查询 =====
fmt.Println("5. 从处理窗口查询：")
// 从处理窗口开始位置查询 3 条记录
processing, err := tp.QueryFromProcessing(3)
if err != nil {
log.Fatal(err)
}
defer cursor.Close()
// 读取 3 条记录
records, err := cursor.NextRange(3)
if err != nil {
log.Fatal(err)
fmt.Printf(" 从处理窗口查询到 %d 条记录:\n", len(processing))
for i, rec := range processing {
fmt.Printf(" [%d] 索引=%d, 状态=%s, 数据=%s\n", i, rec.Index, rec.Status, string(rec.Record.Data))
}
fmt.Printf(" 读取了 %d 条记录:\n", len(records))
for i, rec := range records {
fmt.Printf(" [%d] %s\n", i, string(rec.Data))
}
// 提交游标位置
cursor.Commit()
fmt.Printf(" 游标位置: start=%d, end=%d\n\n", cursor.StartIndex(), cursor.EndIndex())
fmt.Println()
// ===== 6. 继续写入 =====
fmt.Println("6. 继续写入：")

128
example/webapp/main.go
View File

@@ -14,7 +14,7 @@ import (
)
var (
seq *seqlog.Seqlog
seq *seqlog.LogHub
logger *slog.Logger
)
@@ -25,7 +25,7 @@ func main() {
}))
// 创建 Seqlog
seq = seqlog.NewSeqlog("logs", logger, func(topic string, rec *seqlog.Record) error {
seq = seqlog.NewLogHub("logs", logger, func(topic string, rec *seqlog.Record) error {
// 简单的日志处理：只打印摘要信息
dataPreview := string(rec.Data)
if len(dataPreview) > 100 {
@@ -155,6 +155,12 @@ func formatBytes(bytes int64) string {
return fmt.Sprintf("%.2f MB", float64(bytes)/1024/1024)
}
type Record struct {
Index int `json:"index"`
Status string `json:"status"`
Data string `json:"data"`
}
// 首页
func handleIndex(w http.ResponseWriter, r *http.Request) {
html := `<!DOCTYPE html>
@@ -366,7 +372,7 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
// 选择 topic
function selectTopic(topic) {
currentTopic = topic;
direction = 'forward'; // 切换 topic 时重置方向
direction = ''; // 切换 topic 时重置方向
startIndex = null;
endIndex = null;
@@ -405,14 +411,14 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
const forward = document.getElementById('forwardCount').value;
// 构建查询 URL
let url = '/api/query?topic=' + currentTopic +
'&backward=' + backward + '&forward=' + forward +
'&direction=' + direction;
let url = '/api/query?topic=' + currentTopic;
if (direction === 'backward' && startIndex != null) {
url += '&index=' + startIndex;
url += '&direction=backward&index=' + startIndex + '&count=' + backward;
} else if (direction === 'forward' && endIndex != null) {
url += '&index=' + endIndex;
url += '&direction=forward&index=' + endIndex + '&count=' + forward;
} else {
url += '&count=10';
}
const response = await fetch(url);
@@ -449,11 +455,16 @@ func handleIndex(w http.ResponseWriter, r *http.Request) {
}).join('');
if (data.records.length > 0) {
startIndex = data.records[0].index;
endIndex = data.records[data.records.length - 1].index;
} else {
startIndex = null;
endIndex = null;
if (startIndex === null) {
startIndex = data.records[0].index;
} else {
startIndex = Math.min(startIndex, data.records[0].index);
}
if (endIndex === null) {
endIndex = data.records[data.records.length - 1].index;
} else {
endIndex = Math.max(endIndex, data.records[data.records.length - 1].index);
}
}
container.innerHTML = html;
@@ -526,14 +537,10 @@ func handleQuery(w http.ResponseWriter, r *http.Request) {
// 获取查询参数
indexParam := r.URL.Query().Get("index")
direction := r.URL.Query().Get("direction")
backward, _ := strconv.Atoi(r.URL.Query().Get("backward"))
forward, _ := strconv.Atoi(r.URL.Query().Get("forward"))
count, _ := strconv.Atoi(r.URL.Query().Get("count"))
if backward == 0 {
backward = 10
}
if forward == 0 {
forward = 10
if count <= 0 {
count = 10
}
// 获取 processor
@@ -546,79 +553,68 @@ func handleQuery(w http.ResponseWriter, r *http.Request) {
// 获取记录总数
totalCount := processor.GetRecordCount()
// 获取当前处理索引和读取索引（用于状态判断）
startIdx := seq.GetProcessingIndex(topic)
endIdx := seq.GetReadIndex(topic)
// 执行查询
var results []*seqlog.RecordWithStatus
var startRecordIndex int
if direction == "backward" {
// 向前翻页：查询更早的记录
var index int
if indexParam == "" {
index = startIdx
} else {
index, _ = strconv.Atoi(indexParam)
if direction == "" {
results, err = processor.QueryFromProcessing(count)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
if index > 0 {
var queryErr error
results, queryErr = processor.QueryNewest(index-1, backward)
if queryErr != nil {
http.Error(w, queryErr.Error(), http.StatusInternalServerError)
if len(results) == 0 {
results, err = processor.QueryFromLast(count)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
if len(results) > 0 {
// QueryNewest 返回的结果按索引递增排列
startRecordIndex = index - len(results)
}
}
} else {
// 向后翻页：查询更新的记录
var index int
var refIndex int
if indexParam == "" {
index = startIdx
http.Error(w, "参数错误", http.StatusNotFound)
return
} else {
index, _ = strconv.Atoi(indexParam)
refIndex, err = strconv.Atoi(indexParam)
if err != nil {
http.Error(w, err.Error(), http.StatusNotFound)
return
}
}
if index < totalCount {
if direction == "backward" {
var queryErr error
results, queryErr = processor.QueryOldest(index, forward)
results, queryErr = processor.QueryNewest(refIndex, count)
if queryErr != nil {
http.Error(w, queryErr.Error(), http.StatusInternalServerError)
return
}
if len(results) > 0 {
// QueryOldest 返回的结果按索引递增排列
startRecordIndex = index
} else if direction == "forward" {
var queryErr error
results, queryErr = processor.QueryOldest(refIndex, count)
if queryErr != nil {
http.Error(w, queryErr.Error(), http.StatusInternalServerError)
return
}
} else {
http.Error(w, "参数错误", http.StatusNotFound)
return
}
}
type Record struct {
Index int `json:"index"`
Status string `json:"status"`
Data string `json:"data"`
}
records := make([]Record, len(results))
for i, r := range results {
for i, result := range results {
records[i] = Record{
Index: startRecordIndex + i,
Status: r.Status.String(),
Data: string(r.Record.Data),
Index: result.Index,
Status: result.Status.String(),
Data: string(result.Record.Data),
}
}
json.NewEncoder(w).Encode(map[string]interface{}{
"records": records,
"total": len(records),
"totalCount": totalCount,
"processingIndex": startIdx,
"readIndex": endIdx,
"records": records,
"total": len(records),
"totalCount": totalCount,
})
}

108
seqlog_manager.go → loghub.go
View File

@@ -8,14 +8,14 @@ import (
"sync"
)
// Seqlog 日志管理器，统一管理多个 topic 的日志分发
// LogHub 日志中枢，统一管理多个 topic 的日志分发
//
// 自动恢复机制：
// - Start() 时自动扫描 baseDir 中所有 .log 文件
// - 为每个发现的日志文件创建 processor
// - 使用 .pos 文件保存的游标位置恢复处理进度
// - 只处理上次中断后新增的日志，避免重复处理
type Seqlog struct {
type LogHub struct {
baseDir string
processors map[string]*TopicProcessor
defaultHandler TopicRecordHandler
@@ -27,13 +27,13 @@ type Seqlog struct {
running bool
}
// NewSeqlog 创建一个新的日志管理器
// NewLogHub 创建一个新的日志中枢
// logger: 内部日志记录器，如果不需要可以传 slog.Default()
func NewSeqlog(baseDir string, logger *slog.Logger, defaultHandler TopicRecordHandler) *Seqlog {
func NewLogHub(baseDir string, logger *slog.Logger, defaultHandler TopicRecordHandler) *LogHub {
if logger == nil {
logger = slog.Default()
}
return &Seqlog{
return &LogHub{
baseDir: baseDir,
processors: make(map[string]*TopicProcessor),
defaultHandler: defaultHandler,
@@ -48,7 +48,7 @@ func NewSeqlog(baseDir string, logger *slog.Logger, defaultHandler TopicRecordHa
}
// SetDefaultTailConfig 设置默认的 tail 配置
func (s *Seqlog) SetDefaultTailConfig(config *TailConfig) {
func (s *LogHub) SetDefaultTailConfig(config *TailConfig) {
s.mu.Lock()
defer s.mu.Unlock()
if config != nil {
@@ -57,13 +57,13 @@ func (s *Seqlog) SetDefaultTailConfig(config *TailConfig) {
}
// RegisterHandler 为指定 topic 注册 handler
func (s *Seqlog) RegisterHandler(topic string, handler RecordHandler) error {
func (s *LogHub) RegisterHandler(topic string, handler RecordHandler) error {
return s.RegisterHandlerWithConfig(topic, &TopicConfig{Handler: handler})
}
// RegisterHandlerWithConfig 为指定 topic 注册 handler 和配置
// 注意：handler 为必填参数，如果 topic 已存在则返回错误
func (s *Seqlog) RegisterHandlerWithConfig(topic string, config *TopicConfig) error {
func (s *LogHub) RegisterHandlerWithConfig(topic string, config *TopicConfig) error {
s.mu.Lock()
defer s.mu.Unlock()
@@ -94,7 +94,7 @@ func (s *Seqlog) RegisterHandlerWithConfig(topic string, config *TopicConfig) er
}
// Write 写入日志到指定 topic
func (s *Seqlog) Write(topic string, data []byte) (int64, error) {
func (s *LogHub) Write(topic string, data []byte) (int64, error) {
processor, err := s.getOrCreateProcessor(topic)
if err != nil {
s.logger.Error("failed to get processor", "topic", topic, "error", err)
@@ -109,8 +109,8 @@ func (s *Seqlog) Write(topic string, data []byte) (int64, error) {
return offset, nil
}
// Start 启动 Seqlog 和所有已注册的 processor
func (s *Seqlog) Start() error {
// Start 启动 LogHub 和所有已注册的 processor
func (s *LogHub) Start() error {
s.mu.Lock()
defer s.mu.Unlock()
@@ -143,7 +143,7 @@ func (s *Seqlog) Start() error {
// discoverExistingTopics 自动发现 baseDir 中已存在的日志文件并创建对应的 processor
// 注意：此方法需要在持有锁的情况下调用
func (s *Seqlog) discoverExistingTopics() error {
func (s *LogHub) discoverExistingTopics() error {
// 确保目录存在
if err := os.MkdirAll(s.baseDir, 0755); err != nil {
return fmt.Errorf("failed to create base directory: %w", err)
@@ -213,7 +213,7 @@ func (s *Seqlog) discoverExistingTopics() error {
}
// Stop 停止所有 processor
func (s *Seqlog) Stop() error {
func (s *LogHub) Stop() error {
s.mu.Lock()
if !s.running {
s.mu.Unlock()
@@ -250,7 +250,7 @@ func (s *Seqlog) Stop() error {
// getOrCreateProcessor 获取或创建指定 topic 的 processor（使用默认配置）
// 如果没有 defaultHandler，使用空 handler（no-op）
func (s *Seqlog) getOrCreateProcessor(topic string) (*TopicProcessor, error) {
func (s *LogHub) getOrCreateProcessor(topic string) (*TopicProcessor, error) {
// 创建默认配置
var config *TopicConfig
if s.defaultHandler != nil {
@@ -284,7 +284,7 @@ func (s *Seqlog) getOrCreateProcessor(topic string) (*TopicProcessor, error) {
}
// getOrCreateProcessorWithConfig 获取或创建指定 topic 的 processor（使用指定配置）
func (s *Seqlog) getOrCreateProcessorWithConfig(topic string, config *TopicConfig) (*TopicProcessor, error) {
func (s *LogHub) getOrCreateProcessorWithConfig(topic string, config *TopicConfig) (*TopicProcessor, error) {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -330,7 +330,7 @@ func (s *Seqlog) getOrCreateProcessorWithConfig(topic string, config *TopicConfi
}
// GetTopics 获取所有已知的 topic
func (s *Seqlog) GetTopics() []string {
func (s *LogHub) GetTopics() []string {
s.mu.RLock()
defer s.mu.RUnlock()
@@ -341,15 +341,15 @@ func (s *Seqlog) GetTopics() []string {
return topics
}
// IsRunning 检查 Seqlog 是否正在运行
func (s *Seqlog) IsRunning() bool {
// IsRunning 检查 LogHub 是否正在运行
func (s *LogHub) IsRunning() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.running
}
// UpdateTopicConfig 动态更新指定 topic 的 tail 配置
func (s *Seqlog) UpdateTopicConfig(topic string, config *TailConfig) error {
func (s *LogHub) UpdateTopicConfig(topic string, config *TailConfig) error {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -362,7 +362,7 @@ func (s *Seqlog) UpdateTopicConfig(topic string, config *TailConfig) error {
}
// GetTopicConfig 获取指定 topic 的 tail 配置
func (s *Seqlog) GetTopicConfig(topic string) (*TailConfig, error) {
func (s *LogHub) GetTopicConfig(topic string) (*TailConfig, error) {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -375,7 +375,7 @@ func (s *Seqlog) GetTopicConfig(topic string) (*TailConfig, error) {
}
// GetTopicStats 获取指定 topic 的统计信息
func (s *Seqlog) GetTopicStats(topic string) (Stats, error) {
func (s *LogHub) GetTopicStats(topic string) (Stats, error) {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -388,7 +388,7 @@ func (s *Seqlog) GetTopicStats(topic string) (Stats, error) {
}
// GetAllStats 获取所有 topic 的统计信息
func (s *Seqlog) GetAllStats() map[string]Stats {
func (s *LogHub) GetAllStats() map[string]Stats {
s.mu.RLock()
defer s.mu.RUnlock()
@@ -400,7 +400,7 @@ func (s *Seqlog) GetAllStats() map[string]Stats {
}
// NewTopicQuery 为指定 topic 获取查询器（返回共享实例）
func (s *Seqlog) NewTopicQuery(topic string) (*RecordQuery, error) {
func (s *LogHub) NewTopicQuery(topic string) (*RecordQuery, error) {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -413,7 +413,7 @@ func (s *Seqlog) NewTopicQuery(topic string) (*RecordQuery, error) {
}
// GetProcessingIndex 获取指定 topic 的当前处理索引
func (s *Seqlog) GetProcessingIndex(topic string) int {
func (s *LogHub) GetProcessingIndex(topic string) int {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -426,7 +426,7 @@ func (s *Seqlog) GetProcessingIndex(topic string) int {
}
// GetReadIndex 获取指定 topic 的当前读取索引
func (s *Seqlog) GetReadIndex(topic string) int {
func (s *LogHub) GetReadIndex(topic string) int {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -439,7 +439,7 @@ func (s *Seqlog) GetReadIndex(topic string) int {
}
// GetProcessor 获取指定 topic 的 processor
func (s *Seqlog) GetProcessor(topic string) (*TopicProcessor, error) {
func (s *LogHub) GetProcessor(topic string) (*TopicProcessor, error) {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -451,8 +451,44 @@ func (s *Seqlog) GetProcessor(topic string) (*TopicProcessor, error) {
return processor, nil
}
// QueryFromProcessing 从当前处理窗口的开始位置向索引递增方向查询记录
// topic: 主题名称
// count: 查询数量
func (s *LogHub) QueryFromProcessing(topic string, count int) ([]*RecordWithStatus, error) {
processor, err := s.GetProcessor(topic)
if err != nil {
return nil, err
}
return processor.QueryFromProcessing(count)
}
// QueryFromFirst 从第一条记录向索引递增方向查询
// topic: 主题名称
// count: 查询数量
func (s *LogHub) QueryFromFirst(topic string, count int) ([]*RecordWithStatus, error) {
processor, err := s.GetProcessor(topic)
if err != nil {
return nil, err
}
return processor.QueryFromFirst(count)
}
// QueryFromLast 从最后一条记录向索引递减方向查询
// topic: 主题名称
// count: 查询数量
func (s *LogHub) QueryFromLast(topic string, count int) ([]*RecordWithStatus, error) {
processor, err := s.GetProcessor(topic)
if err != nil {
return nil, err
}
return processor.QueryFromLast(count)
}
// Subscribe 为指定 topic 订阅事件（如果 topic 不存在，会在创建时应用订阅）
func (s *Seqlog) Subscribe(topic string, eventType EventType, listener EventListener) error {
func (s *LogHub) Subscribe(topic string, eventType EventType, listener EventListener) error {
s.mu.Lock()
defer s.mu.Unlock()
@@ -476,7 +512,7 @@ func (s *Seqlog) Subscribe(topic string, eventType EventType, listener EventList
}
// SubscribeAll 为指定 topic 订阅所有事件
func (s *Seqlog) SubscribeAll(topic string, listener EventListener) error {
func (s *LogHub) SubscribeAll(topic string, listener EventListener) error {
s.mu.Lock()
defer s.mu.Unlock()
@@ -510,7 +546,7 @@ func (s *Seqlog) SubscribeAll(topic string, listener EventListener) error {
}
// SubscribeAllTopics 为所有 topic 订阅指定事件
func (s *Seqlog) SubscribeAllTopics(eventType EventType, listener EventListener) {
func (s *LogHub) SubscribeAllTopics(eventType EventType, listener EventListener) {
s.mu.Lock()
defer s.mu.Unlock()
@@ -525,8 +561,9 @@ func (s *Seqlog) SubscribeAllTopics(eventType EventType, listener EventListener)
}
// ResetTopic 重置指定 topic 的所有数据
// 注意：必须先停止 Seqlog 或至少停止该 topic 的 processor
func (s *Seqlog) ResetTopic(topic string) error {
// 如果 processor 正在运行且没有待处理的日志，会自动停止后重置
// 如果有待处理的日志，则返回错误
func (s *LogHub) ResetTopic(topic string) error {
s.mu.RLock()
processor, exists := s.processors[topic]
s.mu.RUnlock()
@@ -535,17 +572,12 @@ func (s *Seqlog) ResetTopic(topic string) error {
return NewTopicError(topic, "operation", ErrNotFound)
}
// 先停止 processor
if err := processor.Stop(); err != nil {
return fmt.Errorf("failed to stop processor: %w", err)
}
// 执行重置
// 执行重置（如果没有待处理的日志会自动停止）
if err := processor.Reset(); err != nil {
return fmt.Errorf("failed to reset processor: %w", err)
}
// 如果 seqlog 正在运行，重新启动 processor
// 如果 LogHub 正在运行，重新启动 processor
s.mu.RLock()
running := s.running
s.mu.RUnlock()

119
query.go
View File

@@ -36,7 +36,14 @@ func (s RecordStatus) String() string {
// RecordWithStatus 带状态的记录
type RecordWithStatus struct {
Record *Record
Status RecordStatus
Index int // 记录在日志文件中的索引位置
Status RecordStatus // 记录的处理状态
}
// RecordWithIndex 带索引的记录
type RecordWithIndex struct {
Record *Record
Index int // 记录在日志文件中的索引位置
}
// RecordQuery 记录查询器
@@ -115,26 +122,80 @@ func (rq *RecordQuery) readRecordsForward(startIndex, count int) ([]*Record, err
return results, nil
}
// QueryOldest 从指定索引向索引递增方向查询记录
// startIndex: 查询起始索引
// QueryOldest 从参考索引向索引递减方向查询记录（查询更早的记录）
// refIndex: 参考索引位置
// count: 查询数量
// 返回的记录按索引递增方向排序
func (rq *RecordQuery) QueryOldest(startIndex, count int) ([]*Record, error) {
// 返回的记录按索引递增方向排序，包含索引信息
// 例如：QueryOldest(5, 3) 查询索引 2, 3, 4（不包含 5），返回 [2, 3, 4]
func (rq *RecordQuery) QueryOldest(refIndex, count int) ([]*RecordWithIndex, error) {
if count <= 0 {
return nil, NewValidationError("count", "count must be greater than 0", ErrInvalidCount)
}
totalCount := rq.index.Count()
if totalCount == 0 {
return []*Record{}, nil
return []*RecordWithIndex{}, nil
}
// 校验起始索引
// 验证参考索引范围（严格模式）
if refIndex < 0 || refIndex > totalCount {
return nil, NewValidationError("refIndex", fmt.Sprintf("refIndex %d out of range [0, %d]", refIndex, totalCount), ErrInvalidRange)
}
// 计算实际起始索引（向索引递减方向）
startIndex := refIndex - count
if startIndex < 0 {
startIndex = 0
count = refIndex // 调整实际数量
}
if count <= 0 {
return []*RecordWithIndex{}, nil
}
// 读取记录
records, err := rq.readRecordsForward(startIndex, count)
if err != nil {
return nil, err
}
// 转换为带索引的记录
results := make([]*RecordWithIndex, len(records))
for i, rec := range records {
results[i] = &RecordWithIndex{
Record: rec,
Index: startIndex + i,
}
}
return results, nil
}
// QueryNewest 从参考索引向索引递增方向查询记录（查询更新的记录）
// refIndex: 参考索引位置
// count: 查询数量
// 返回的记录按索引递增方向排序，包含索引信息
// 例如：QueryNewest(5, 3) 查询索引 6, 7, 8（不包含 5），返回 [6, 7, 8]
func (rq *RecordQuery) QueryNewest(refIndex, count int) ([]*RecordWithIndex, error) {
if count <= 0 {
return nil, NewValidationError("count", "count must be greater than 0", ErrInvalidCount)
}
totalCount := rq.index.Count()
if totalCount == 0 {
return []*RecordWithIndex{}, nil
}
// 验证参考索引范围（严格模式）
// QueryNewest 允许 refIndex = -1（从头开始查询）
if refIndex < -1 || refIndex >= totalCount {
return nil, NewValidationError("refIndex", fmt.Sprintf("refIndex %d out of range [-1, %d)", refIndex, totalCount), ErrInvalidRange)
}
// 计算实际起始索引（向索引递增方向）
startIndex := refIndex + 1
if startIndex >= totalCount {
return []*Record{}, nil
return []*RecordWithIndex{}, nil
}
// 限制查询数量
@@ -143,40 +204,22 @@ func (rq *RecordQuery) QueryOldest(startIndex, count int) ([]*Record, error) {
count = remainCount
}
return rq.readRecordsForward(startIndex, count)
}
// QueryNewest 从指定索引向索引递减方向查询记录
// endIndex: 查询的最大索引（向前查询更早的记录）
// count: 查询数量
// 返回的记录按索引递增方向排序（与 QueryOldest 一致）
func (rq *RecordQuery) QueryNewest(endIndex, count int) ([]*Record, error) {
if count <= 0 {
return nil, NewValidationError("count", "count must be greater than 0", ErrInvalidCount)
// 读取记录
records, err := rq.readRecordsForward(startIndex, count)
if err != nil {
return nil, err
}
totalCount := rq.index.Count()
if totalCount == 0 {
return []*Record{}, nil
// 转换为带索引的记录
results := make([]*RecordWithIndex, len(records))
for i, rec := range records {
results[i] = &RecordWithIndex{
Record: rec,
Index: startIndex + i,
}
}
// 校验结束索引
if endIndex < 0 {
return []*Record{}, nil
}
if endIndex >= totalCount {
endIndex = totalCount - 1
}
// 计算实际起始索引（向索引递减方向查询 count 条）
queryStartIdx := endIndex - count + 1
if queryStartIdx < 0 {
queryStartIdx = 0
count = endIndex + 1 // 调整实际数量
}
// 向前读取，返回按索引递增排序的结果
return rq.readRecordsForward(queryStartIdx, count)
return results, nil
}
// GetRecordCount 获取记录总数

75
query_test.go Normal file
View File

@@ -0,0 +1,75 @@
package seqlog
import (
"fmt"
"testing"
"time"
)
// TestQueryFromProcessing 测试从处理窗口开始位置查询记录
func TestQueryFromProcessing(t *testing.T) {
// 创建临时目录
tmpDir := t.TempDir()
// 创建配置（带 handler，不处理任何记录以保持窗口稳定）
config := &TopicConfig{
Handler: func(record *Record) error {
// 不处理，只是为了让 tailer 启动
time.Sleep(1 * time.Second) // 延迟处理
return nil
},
}
// 创建 TopicProcessor
processor, err := NewTopicProcessor(tmpDir, "test", nil, config)
if err != nil {
t.Fatal(err)
}
defer processor.Close()
// 写入 10 条记录
for i := 0; i < 10; i++ {
msg := fmt.Sprintf("message %d", i)
_, err := processor.Write([]byte(msg))
if err != nil {
t.Fatal(err)
}
}
// 不启动 tailer，直接测试查询功能
// startIdx 应该是 0（没有处理任何记录）
// 从处理窗口开始位置查询 5 条记录
records, err := processor.QueryFromProcessing(5)
if err != nil {
t.Fatal(err)
}
t.Logf("查询到 %d 条记录", len(records))
if len(records) != 5 {
t.Fatalf("expected 5 records, got %d", len(records))
}
// 验证查询结果从索引 0 开始
for i, rec := range records {
expectedIndex := i
if rec.Index != expectedIndex {
t.Errorf("record[%d]: expected index %d, got %d", i, expectedIndex, rec.Index)
}
expectedMsg := fmt.Sprintf("message %d", expectedIndex)
if string(rec.Record.Data) != expectedMsg {
t.Errorf("record[%d]: expected data '%s', got '%s'", i, expectedMsg, string(rec.Record.Data))
}
// 未启动 tailer，所有记录都应该是 Pending 状态
if rec.Status != StatusPending {
t.Errorf("record[%d]: expected StatusPending, got %s", i, rec.Status)
}
t.Logf(" [索引 %d] %s - 状态: %s", rec.Index, string(rec.Record.Data), rec.Status)
}
t.Log("QueryFromProcessing 测试通过")
}

4
seqlog.go
View File

@@ -42,13 +42,13 @@ import "github.com/google/uuid"
// tailer, _ := seqlog.NewTailer("app.log", handler, nil)
// tailer.Start()
//
// // 使用 Seqlog 管理器（带 slog 支持和自动恢复）
// // 使用 LogHub 管理器（带 slog 支持和自动恢复）
// logger := slog.Default()
// handler := func(topic string, rec *seqlog.Record) error {
// fmt.Printf("[%s] %s\n", topic, string(rec.Data))
// return nil
// }
// seq := seqlog.NewSeqlog("/tmp/logs", logger, handler)
// seq := seqlog.NewLogHub("/tmp/logs", logger, handler)
// seq.Start() // 自动发现并恢复已存在的日志文件
// seq.Write("app", []byte("application log"))
//

344
seqlog_test.go
View File

@@ -437,7 +437,7 @@ func TestSeqlogBasic(t *testing.T) {
os.MkdirAll(tmpDir, 0755)
defer os.RemoveAll(tmpDir)
seqlog := NewSeqlog(tmpDir, slog.Default(), nil)
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
// 注册 handler
appLogs := make([]string, 0)
@@ -493,7 +493,7 @@ func TestSeqlogDefaultHandler(t *testing.T) {
return nil
}
seqlog := NewSeqlog(tmpDir, slog.Default(), defaultHandler)
seqlog := NewLogHub(tmpDir, slog.Default(), defaultHandler)
// 注册特定 handler
seqlog.RegisterHandler("special", func(rec *Record) error {
@@ -534,7 +534,7 @@ func TestSeqlogDynamicRegistration(t *testing.T) {
os.MkdirAll(tmpDir, 0755)
defer os.RemoveAll(tmpDir)
seqlog := NewSeqlog(tmpDir, slog.Default(), nil)
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
// 先注册 handler（handler 现在是必填项）
logs := make([]string, 0)
@@ -566,7 +566,7 @@ func TestDynamicConfigUpdate(t *testing.T) {
os.MkdirAll(tmpDir, 0755)
defer os.RemoveAll(tmpDir)
seqlog := NewSeqlog(tmpDir, slog.Default(), nil)
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
// 注册 handler
logs := make([]string, 0)
@@ -746,7 +746,7 @@ func TestSeqlogAutoRecovery(t *testing.T) {
return nil
}
seqlog1 := NewSeqlog(tmpDir, slog.Default(), defaultHandler)
seqlog1 := NewLogHub(tmpDir, slog.Default(), defaultHandler)
seqlog1.Start()
// 写入一些日志
@@ -777,7 +777,7 @@ func TestSeqlogAutoRecovery(t *testing.T) {
mu.Unlock()
// 第二阶段：重启并自动恢复
seqlog2 := NewSeqlog(tmpDir, slog.Default(), defaultHandler)
seqlog2 := NewLogHub(tmpDir, slog.Default(), defaultHandler)
seqlog2.Start()
// 写入新日志
@@ -880,7 +880,7 @@ func TestSeqlogCleanup(t *testing.T) {
os.MkdirAll(tmpDir, 0755)
defer os.RemoveAll(tmpDir)
seqlog := NewSeqlog(tmpDir, slog.Default(), nil)
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
seqlog.Start()
// 写入多个 topic 的日志
@@ -1118,7 +1118,7 @@ func TestSeqlogStats(t *testing.T) {
return nil
}
seq := NewSeqlog(tmpDir, slog.Default(), handler)
seq := NewLogHub(tmpDir, slog.Default(), handler)
if err := seq.Start(); err != nil {
t.Fatalf("failed to start seqlog: %v", err)
}
@@ -1229,16 +1229,19 @@ func TestRecordQuery(t *testing.T) {
}
defer query.Close()
// 测试查询当前位置
current, err := query.QueryOldest(startIdx, 1)
// 测试查询当前位置（使用 QueryNewest 查询 startIdx）
current, err := query.QueryNewest(startIdx-1, 1)
if err != nil {
t.Fatalf("failed to query current: %v", err)
}
if len(current) != 1 {
t.Fatalf("expected 1 current result, got %d", len(current))
}
if string(current[0].Data) != "message 5" {
t.Errorf("expected current 'message 5', got '%s'", string(current[0].Data))
if string(current[0].Record.Data) != "message 5" {
t.Errorf("expected current 'message 5', got '%s'", string(current[0].Record.Data))
}
if current[0].Index != startIdx {
t.Errorf("expected index %d, got %d", startIdx, current[0].Index)
}
// 手动判断状态
status := GetRecordStatus(startIdx, startIdx, endIdx)
@@ -1246,43 +1249,53 @@ func TestRecordQuery(t *testing.T) {
t.Errorf("expected status Processing, got %s", status)
}
// 测试向后查询（查询更早的记录，返回按索引递增排序）
backResults, err := query.QueryNewest(startIdx-1, 3)
// 测试 QueryOldest：查询更早的记录（向索引递减方向）
// QueryOldest(5, 3) 查询索引 2, 3, 4
backResults, err := query.QueryOldest(startIdx, 3)
if err != nil {
t.Fatalf("failed to query backward: %v", err)
}
if len(backResults) != 3 {
t.Errorf("expected 3 backward results, got %d", len(backResults))
}
// 向后查询返回按索引递增排序的结果
// 返回按索引递增排序的结果：2, 3, 4
expectedBack := []string{"message 2", "message 3", "message 4"}
for i, rec := range backResults {
if string(rec.Data) != expectedBack[i] {
t.Errorf("backward[%d]: expected '%s', got '%s'", i, expectedBack[i], string(rec.Data))
if string(rec.Record.Data) != expectedBack[i] {
t.Errorf("backward[%d]: expected '%s', got '%s'", i, expectedBack[i], string(rec.Record.Data))
}
// 手动判断状态：索引 2, 3, 4
recStatus := GetRecordStatus(startIdx-3+i, startIdx, endIdx)
expectedIndex := startIdx - 3 + i
if rec.Index != expectedIndex {
t.Errorf("backward[%d]: expected index %d, got %d", i, expectedIndex, rec.Index)
}
// 手动判断状态：索引 2, 3, 4 都已处理
recStatus := GetRecordStatus(rec.Index, startIdx, endIdx)
if recStatus != StatusProcessed {
t.Errorf("backward[%d]: expected status Processed, got %s", i, recStatus)
}
}
// 测试向前查询（查询更新的记录）
forwardResults, err := query.QueryOldest(endIdx, 3)
// 测试 QueryNewest：查询更新的记录（向索引递增方向）
// QueryNewest(endIdx, 3) 从 endIdx 向后查询，查询索引 6, 7, 8
forwardResults, err := query.QueryNewest(endIdx-1, 3)
if err != nil {
t.Fatalf("failed to query forward: %v", err)
}
if len(forwardResults) != 3 {
t.Errorf("expected 3 forward results, got %d", len(forwardResults))
}
// 向前查询返回顺序结果
// 返回按索引递增排序的结果：6, 7, 8
expectedForward := []string{"message 6", "message 7", "message 8"}
for i, rec := range forwardResults {
if string(rec.Data) != expectedForward[i] {
t.Errorf("forward[%d]: expected '%s', got '%s'", i, expectedForward[i], string(rec.Data))
if string(rec.Record.Data) != expectedForward[i] {
t.Errorf("forward[%d]: expected '%s', got '%s'", i, expectedForward[i], string(rec.Record.Data))
}
// 手动判断状态：索引 6, 7, 8
recStatus := GetRecordStatus(endIdx+i, startIdx, endIdx)
expectedIndex := endIdx + i
if rec.Index != expectedIndex {
t.Errorf("forward[%d]: expected index %d, got %d", i, expectedIndex, rec.Index)
}
// 手动判断状态：索引 6, 7, 8 待处理
recStatus := GetRecordStatus(rec.Index, startIdx, endIdx)
if recStatus != StatusPending {
t.Errorf("forward[%d]: expected status Pending, got %s", i, recStatus)
}
@@ -1384,7 +1397,7 @@ func TestSeqlogQuery(t *testing.T) {
return nil
}
seq := NewSeqlog(tmpDir, slog.Default(), handler)
seq := NewLogHub(tmpDir, slog.Default(), handler)
if err := seq.Start(); err != nil {
t.Fatalf("failed to start seqlog: %v", err)
}
@@ -1414,7 +1427,7 @@ func TestSeqlogQuery(t *testing.T) {
// 获取 processor 用于查询（带状态）
processor, _ := seq.GetProcessor("app")
index := processor.Index()
totalCount := processor.GetRecordCount()
// 测试查询当前
if startIdx < endIdx {
@@ -1439,7 +1452,7 @@ func TestSeqlogQuery(t *testing.T) {
}
// 测试向前查询
if startIdx < index.Count() {
if startIdx < totalCount {
forward, err := processor.QueryOldest(endIdx, 3)
if err != nil {
t.Fatalf("failed to query forward: %v", err)
@@ -1569,7 +1582,7 @@ func TestSeqlogEventSubscription(t *testing.T) {
return nil
}
seq := NewSeqlog(tmpDir, slog.Default(), handler)
seq := NewLogHub(tmpDir, slog.Default(), handler)
if err := seq.Start(); err != nil {
t.Fatalf("failed to start seqlog: %v", err)
}
@@ -1612,7 +1625,7 @@ func TestMultiTopicEventSubscription(t *testing.T) {
return nil
}
seq := NewSeqlog(tmpDir, slog.Default(), handler)
seq := NewLogHub(tmpDir, slog.Default(), handler)
if err := seq.Start(); err != nil {
t.Fatalf("failed to start seqlog: %v", err)
}
@@ -1658,7 +1671,7 @@ func TestMultiTopicEventSubscription(t *testing.T) {
func TestTopicReset(t *testing.T) {
tmpDir := t.TempDir()
seqlog := NewSeqlog(tmpDir, slog.Default(), nil)
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
// 注册 handler
seqlog.RegisterHandler("test", func(rec *Record) error {
@@ -1750,6 +1763,50 @@ func TestTopicReset(t *testing.T) {
seqlog.Stop()
}
// TestTopicResetWithPendingRecords 测试当有待处理日志时 Reset 返回错误
func TestTopicResetWithPendingRecords(t *testing.T) {
tmpDir := t.TempDir()
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
// 注册一个慢速 handler，让日志堆积
slowHandler := func(rec *Record) error {
time.Sleep(100 * time.Millisecond) // 模拟慢速处理
return nil
}
seqlog.RegisterHandler("test", slowHandler)
seqlog.Start()
// 快速写入多条日志
for i := 0; i < 10; i++ {
data := []byte(fmt.Sprintf("message %d", i))
if _, err := seqlog.Write("test", data); err != nil {
t.Fatalf("写入失败: %v", err)
}
}
// 短暂等待，让一部分日志开始处理但不是全部
time.Sleep(200 * time.Millisecond)
// 尝试重置，应该失败因为有待处理的日志
err := seqlog.ResetTopic("test")
if err == nil {
t.Fatal("期望 Reset 失败因为有待处理的日志，但成功了")
}
t.Logf("预期的错误: %v", err)
// 停止处理
seqlog.Stop()
// 现在 Reset 应该成功（停止后没有待处理的日志）
processor, _ := seqlog.GetProcessor("test")
if err := processor.Reset(); err != nil {
t.Fatalf("停止后 Reset 应该成功: %v", err)
}
}
// TestQueryOldestNewest 测试 QueryOldest 和 QueryNewest
func TestQueryOldestNewest(t *testing.T) {
tmpDir := t.TempDir()
@@ -1773,10 +1830,11 @@ func TestQueryOldestNewest(t *testing.T) {
}
}
// 测试 QueryOldest - 从索引 0 开始查询 3 条
oldest, err := processor.QueryOldest(0, 3)
// 测试 QueryNewest - 查询索引 0, 1, 2（向索引递增方向）
// QueryNewest(-1, 3) 从 -1 向后查询，得到索引 0, 1, 2
oldest, err := processor.QueryNewest(-1, 3)
if err != nil {
t.Fatalf("QueryOldest failed: %v", err)
t.Fatalf("QueryNewest failed: %v", err)
}
if len(oldest) != 3 {
t.Errorf("expected 3 records, got %d", len(oldest))
@@ -1787,14 +1845,18 @@ func TestQueryOldestNewest(t *testing.T) {
if string(oldest[i].Record.Data) != expected {
t.Errorf("oldest[%d]: expected %s, got %s", i, expected, string(oldest[i].Record.Data))
}
t.Logf("Oldest[%d]: %s - %s", i, string(oldest[i].Record.Data), oldest[i].Status)
if oldest[i].Index != i {
t.Errorf("oldest[%d]: expected index %d, got %d", i, i, oldest[i].Index)
}
t.Logf("Oldest[%d]: index=%d, %s - %s", i, oldest[i].Index, string(oldest[i].Record.Data), oldest[i].Status)
}
// 测试 QueryNewest - 从索引 9 结束查询 3 条
// 测试 QueryOldest - 查询索引 7, 8, 9（向索引递减方向）
// QueryOldest(10, 3) 从 10 向前查询，得到索引 7, 8, 9
totalCount := processor.GetRecordCount()
newest, err := processor.QueryNewest(totalCount-1, 3)
newest, err := processor.QueryOldest(totalCount, 3)
if err != nil {
t.Fatalf("QueryNewest failed: %v", err)
t.Fatalf("QueryOldest failed: %v", err)
}
if len(newest) != 3 {
t.Errorf("expected 3 records, got %d", len(newest))
@@ -1805,13 +1867,17 @@ func TestQueryOldestNewest(t *testing.T) {
if string(newest[i].Record.Data) != expected {
t.Errorf("newest[%d]: expected %s, got %s", i, expected, string(newest[i].Record.Data))
}
t.Logf("Newest[%d]: %s - %s", i, string(newest[i].Record.Data), newest[i].Status)
if newest[i].Index != 7+i {
t.Errorf("newest[%d]: expected index %d, got %d", i, 7+i, newest[i].Index)
}
t.Logf("Newest[%d]: index=%d, %s - %s", i, newest[i].Index, string(newest[i].Record.Data), newest[i].Status)
}
// 测试超出范围
all, err := processor.QueryOldest(0, 100)
// 测试超出范围 - 查询所有记录
// QueryNewest(-1, 100) 从 -1 向后查询，会返回所有记录（最多 100 条）
all, err := processor.QueryNewest(-1, 100)
if err != nil {
t.Fatalf("QueryOldest(0, 100) failed: %v", err)
t.Fatalf("QueryNewest(-1, 100) failed: %v", err)
}
if len(all) != 10 {
t.Errorf("expected 10 records, got %d", len(all))
@@ -1828,11 +1894,195 @@ func TestQueryOldestNewest(t *testing.T) {
}
defer processor2.Close()
emptyOldest, err := processor2.QueryOldest(0, 10)
emptyNewest, err := processor2.QueryNewest(-1, 10)
if err != nil {
t.Fatalf("QueryOldest on empty failed: %v", err)
t.Fatalf("QueryNewest on empty failed: %v", err)
}
if len(emptyOldest) != 0 {
t.Errorf("expected 0 records, got %d", len(emptyOldest))
if len(emptyNewest) != 0 {
t.Errorf("expected 0 records, got %d", len(emptyNewest))
}
}
// TestQueryFromFirstAndLast 测试 QueryFromFirst 和 QueryFromLast
func TestQueryFromFirstAndLast(t *testing.T) {
tmpDir := t.TempDir()
// 创建 TopicProcessor
processor, err := NewTopicProcessor(tmpDir, "test", nil, &TopicConfig{
Handler: func(rec *Record) error {
return nil
},
})
if err != nil {
t.Fatal(err)
}
defer processor.Close()
// 写入 10 条测试数据
for i := 0; i < 10; i++ {
data := fmt.Sprintf("message %d", i)
if _, err := processor.Write([]byte(data)); err != nil {
t.Fatal(err)
}
}
// 测试 QueryFromFirst - 从第一条记录向索引递增方向查询
t.Run("QueryFromFirst", func(t *testing.T) {
// 查询前 3 条记录
records, err := processor.QueryFromFirst(3)
if err != nil {
t.Fatalf("QueryFromFirst failed: %v", err)
}
if len(records) != 3 {
t.Fatalf("expected 3 records, got %d", len(records))
}
// 验证结果：应该是索引 0, 1, 2
for i := 0; i < 3; i++ {
expectedData := fmt.Sprintf("message %d", i)
if string(records[i].Record.Data) != expectedData {
t.Errorf("records[%d]: expected %s, got %s", i, expectedData, string(records[i].Record.Data))
}
if records[i].Index != i {
t.Errorf("records[%d]: expected index %d, got %d", i, i, records[i].Index)
}
t.Logf("FromFirst[%d]: index=%d, %s - %s", i, records[i].Index, string(records[i].Record.Data), records[i].Status)
}
// 查询超过总数的记录
allRecords, err := processor.QueryFromFirst(100)
if err != nil {
t.Fatalf("QueryFromFirst(100) failed: %v", err)
}
if len(allRecords) != 10 {
t.Errorf("expected 10 records, got %d", len(allRecords))
}
})
// 测试 QueryFromLast - 从最后一条记录向索引递减方向查询
t.Run("QueryFromLast", func(t *testing.T) {
// 查询最后 3 条记录
records, err := processor.QueryFromLast(3)
if err != nil {
t.Fatalf("QueryFromLast failed: %v", err)
}
if len(records) != 3 {
t.Fatalf("expected 3 records, got %d", len(records))
}
// 验证结果：应该是索引 7, 8, 9（按索引递增顺序排列）
for i := 0; i < 3; i++ {
expectedIndex := 7 + i
expectedData := fmt.Sprintf("message %d", expectedIndex)
if string(records[i].Record.Data) != expectedData {
t.Errorf("records[%d]: expected %s, got %s", i, expectedData, string(records[i].Record.Data))
}
if records[i].Index != expectedIndex {
t.Errorf("records[%d]: expected index %d, got %d", i, expectedIndex, records[i].Index)
}
t.Logf("FromLast[%d]: index=%d, %s - %s", i, records[i].Index, string(records[i].Record.Data), records[i].Status)
}
// 查询超过总数的记录
allRecords, err := processor.QueryFromLast(100)
if err != nil {
t.Fatalf("QueryFromLast(100) failed: %v", err)
}
if len(allRecords) != 10 {
t.Errorf("expected 10 records, got %d", len(allRecords))
}
})
// 测试空数据库
t.Run("EmptyDatabase", func(t *testing.T) {
emptyProcessor, err := NewTopicProcessor(t.TempDir(), "empty", nil, &TopicConfig{
Handler: func(rec *Record) error {
return nil
},
})
if err != nil {
t.Fatal(err)
}
defer emptyProcessor.Close()
// QueryFromFirst 应该返回空数组
firstRecords, err := emptyProcessor.QueryFromFirst(10)
if err != nil {
t.Fatalf("QueryFromFirst on empty failed: %v", err)
}
if len(firstRecords) != 0 {
t.Errorf("expected 0 records, got %d", len(firstRecords))
}
// QueryFromLast 应该返回空数组
lastRecords, err := emptyProcessor.QueryFromLast(10)
if err != nil {
t.Fatalf("QueryFromLast on empty failed: %v", err)
}
if len(lastRecords) != 0 {
t.Errorf("expected 0 records, got %d", len(lastRecords))
}
})
}
// TestLogHubQueryFromFirstAndLast 测试 LogHub 的 QueryFromFirst 和 QueryFromLast
func TestLogHubQueryFromFirstAndLast(t *testing.T) {
tmpDir := t.TempDir()
seqlog := NewLogHub(tmpDir, slog.Default(), nil)
seqlog.RegisterHandler("test", func(rec *Record) error {
return nil
})
seqlog.Start()
defer seqlog.Stop()
// 写入测试数据
for i := 0; i < 10; i++ {
data := fmt.Sprintf("message %d", i)
if _, err := seqlog.Write("test", []byte(data)); err != nil {
t.Fatal(err)
}
}
// 测试 QueryFromFirst
firstRecords, err := seqlog.QueryFromFirst("test", 3)
if err != nil {
t.Fatalf("QueryFromFirst failed: %v", err)
}
if len(firstRecords) != 3 {
t.Fatalf("expected 3 records, got %d", len(firstRecords))
}
for i := 0; i < 3; i++ {
if firstRecords[i].Index != i {
t.Errorf("firstRecords[%d]: expected index %d, got %d", i, i, firstRecords[i].Index)
}
}
// 测试 QueryFromLast
lastRecords, err := seqlog.QueryFromLast("test", 3)
if err != nil {
t.Fatalf("QueryFromLast failed: %v", err)
}
if len(lastRecords) != 3 {
t.Fatalf("expected 3 records, got %d", len(lastRecords))
}
for i := 0; i < 3; i++ {
expectedIndex := 7 + i
if lastRecords[i].Index != expectedIndex {
t.Errorf("lastRecords[%d]: expected index %d, got %d", i, expectedIndex, lastRecords[i].Index)
}
}
// 测试不存在的 topic
_, err = seqlog.QueryFromFirst("nonexistent", 10)
if err == nil {
t.Error("expected error for nonexistent topic")
}
_, err = seqlog.QueryFromLast("nonexistent", 10)
if err == nil {
t.Error("expected error for nonexistent topic")
}
}

4
tailer.go
View File

@@ -22,7 +22,7 @@ type TailConfig struct {
// LogTailer 持续监控处理器
type LogTailer struct {
cursor *LogCursor
cursor *ProcessCursor
handler RecordHandler
config TailConfig
configCh chan TailConfig // 用于动态更新配置
@@ -32,7 +32,7 @@ type LogTailer struct {
// NewTailer 创建一个新的 tail 处理器
// cursor: 外部提供的游标，用于读取和跟踪日志位置
func NewTailer(cursor *LogCursor, handler RecordHandler, config *TailConfig) (*LogTailer, error) {
func NewTailer(cursor *ProcessCursor, handler RecordHandler, config *TailConfig) (*LogTailer, error) {
if cursor == nil {
return nil, fmt.Errorf("cursor cannot be nil")
}

116
topic_processor.go
View File

@@ -20,7 +20,7 @@ type TopicProcessor struct {
writer *LogWriter // 写入器
index *RecordIndex // 索引管理器
query *RecordQuery // 查询器
cursor *LogCursor // 游标
cursor *ProcessCursor // 游标
tailer *LogTailer // 持续处理器
// 配置和状态
@@ -339,13 +339,14 @@ func (tp *TopicProcessor) Query() *RecordQuery {
return tp.query
}
// QueryOldest 从指定索引向索引递增方向查询记录
// startIndex: 查询起始索引
// QueryOldest 从参考索引向索引递减方向查询记录（查询更早的记录）
// refIndex: 参考索引位置
// count: 查询数量
// 返回的记录包含状态信息（基于 tailer 的窗口索引），按索引递增方向排序
func (tp *TopicProcessor) QueryOldest(startIndex, count int) ([]*RecordWithStatus, error) {
// 查询记录
records, err := tp.query.QueryOldest(startIndex, count)
// 返回的记录包含索引和状态信息，按索引递增方向排序
// 例如：QueryOldest(5, 3) 查询索引 2, 3, 4（不包含 5）
func (tp *TopicProcessor) QueryOldest(refIndex, count int) ([]*RecordWithStatus, error) {
// 查询记录（包含索引信息）
records, err := tp.query.QueryOldest(refIndex, count)
if err != nil {
return nil, err
}
@@ -363,21 +364,23 @@ func (tp *TopicProcessor) QueryOldest(startIndex, count int) ([]*RecordWithStatu
results := make([]*RecordWithStatus, len(records))
for i, rec := range records {
results[i] = &RecordWithStatus{
Record: rec,
Status: GetRecordStatus(startIndex+i, startIdx, endIdx),
Record: rec.Record,
Index: rec.Index,
Status: GetRecordStatus(rec.Index, startIdx, endIdx),
}
}
return results, nil
}
// QueryNewest 从指定索引向索引递减方向查询记录
// endIndex: 查询的最大索引（向前查询更早的记录）
// QueryNewest 从参考索引向索引递增方向查询记录（查询更新的记录）
// refIndex: 参考索引位置
// count: 查询数量
// 返回的记录包含状态信息（基于 tailer 的窗口索引），按索引递增方向排序
func (tp *TopicProcessor) QueryNewest(endIndex, count int) ([]*RecordWithStatus, error) {
// 查询记录
records, err := tp.query.QueryNewest(endIndex, count)
// 返回的记录包含索引和状态信息，按索引递增方向排序
// 例如：QueryNewest(5, 3) 查询索引 6, 7, 8（不包含 5）
func (tp *TopicProcessor) QueryNewest(refIndex, count int) ([]*RecordWithStatus, error) {
// 查询记录（包含索引信息）
records, err := tp.query.QueryNewest(refIndex, count)
if err != nil {
return nil, err
}
@@ -391,12 +394,13 @@ func (tp *TopicProcessor) QueryNewest(endIndex, count int) ([]*RecordWithStatus,
}
tp.mu.RUnlock()
// 为每条记录添加状态（倒序：endIndex, endIndex-1, ...）
// 为每条记录添加状态
results := make([]*RecordWithStatus, len(records))
for i, rec := range records {
results[i] = &RecordWithStatus{
Record: rec,
Status: GetRecordStatus(endIndex-i, startIdx, endIdx),
Record: rec.Record,
Index: rec.Index,
Status: GetRecordStatus(rec.Index, startIdx, endIdx),
}
}
@@ -408,16 +412,43 @@ func (tp *TopicProcessor) GetRecordCount() int {
return tp.index.Count()
}
// Cursor 创建一个新的游标实例（使用共享的 index）
// 注意：每次调用都会创建新实例，调用者需要负责关闭
// Tailer 内部有自己的游标，不会与此冲突
func (tp *TopicProcessor) Cursor() (*LogCursor, error) {
return NewCursor(tp.logPath, tp.index)
// QueryFromProcessing 从当前处理窗口的开始位置向索引递增方向查询记录
// count: 查询数量
// 返回从处理窗口开始位置（startIndex）向后的记录，包含状态信息
func (tp *TopicProcessor) QueryFromProcessing(count int) ([]*RecordWithStatus, error) {
// 获取当前处理窗口的开始位置
tp.mu.RLock()
var startIdx int
if tp.tailer != nil {
startIdx = tp.tailer.GetStartIndex()
}
tp.mu.RUnlock()
// 从 startIdx 开始向索引递增方向查询
// QueryNewest(refIndex, count) 查询从 refIndex+1 开始的记录
// 所以要从 startIdx 开始，应该调用 QueryNewest(startIdx - 1, count)
return tp.QueryNewest(startIdx-1, count)
}
// Index 获取索引管理器
func (tp *TopicProcessor) Index() *RecordIndex {
return tp.index
// QueryFromFirst 从第一条记录向索引递增方向查询
// count: 查询数量
// 返回从第一条记录（索引 0）开始的记录，包含状态信息
// 例如：QueryFromFirst(3) 查询索引 0, 1, 2
func (tp *TopicProcessor) QueryFromFirst(count int) ([]*RecordWithStatus, error) {
// QueryNewest(-1, count) 会从索引 0 开始向后查询
return tp.QueryNewest(-1, count)
}
// QueryFromLast 从最后一条记录向索引递减方向查询
// count: 查询数量
// 返回从最后一条记录开始向前的记录，包含状态信息，按索引递增方向排序
// 例如：如果总共有 10 条记录，QueryFromLast(3) 查询索引 7, 8, 9
func (tp *TopicProcessor) QueryFromLast(count int) ([]*RecordWithStatus, error) {
// 获取记录总数
totalCount := tp.index.Count()
// QueryOldest(totalCount, count) 会从最后一条记录开始向前查询
return tp.QueryOldest(totalCount, count)
}
// GetProcessingIndex 获取当前处理索引（窗口开始索引）
@@ -460,13 +491,42 @@ func (tp *TopicProcessor) Unsubscribe(eventType EventType) {
}
// Reset 清空 topic 的所有数据，包括日志文件、位置文件和统计文件
// 注意：必须在 Stop 之后调用
// 如果 processor 正在运行且没有待处理的日志，会自动停止
// 如果有待处理的日志，则返回错误，需要等待处理完成或手动停止
func (tp *TopicProcessor) Reset() error {
tp.mu.Lock()
defer tp.mu.Unlock()
if tp.running {
return fmt.Errorf("cannot reset while processor is running, please stop first")
// 检查是否有待处理的日志
processingIndex := 0
if tp.tailer != nil {
processingIndex = tp.tailer.GetStartIndex()
}
recordCount := tp.index.Count()
hasPendingRecords := processingIndex < recordCount
if hasPendingRecords {
return fmt.Errorf("cannot reset while processor is running with pending records (%d/%d processed), please stop first or wait for processing to complete", processingIndex, recordCount)
}
// 没有待处理的日志，自动停止
tp.logger.Debug("auto-stopping processor before reset (no pending records)")
tp.running = false
tp.cancel()
// 释放锁以等待 tailer 停止
tp.mu.Unlock()
tp.wg.Wait()
tp.mu.Lock()
// 发布停止事件
tp.eventBus.Publish(&Event{
Type: EventProcessorStop,
Topic: tp.topic,
Timestamp: time.Now(),
})
}
tp.logger.Debug("resetting processor")