From 97aea66f7c5f29ea97732fcbfa94a49c572cf6a7 Mon Sep 17 00:00:00 2001 From: huang <1724659546@qq.com> Date: Wed, 5 Nov 2025 22:22:46 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9infra=E9=99=A4repository?= =?UTF-8?q?=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../provide-logger-with-mothed/task-infra.md | 172 +++++++++--------- internal/core/component_initializers.go | 2 +- internal/infra/database/postgres.go | 119 ++++++------ internal/infra/database/storage.go | 16 +- internal/infra/models/device_template.go | 1 + internal/infra/notify/lark.go | 15 +- internal/infra/notify/log_notifier.go | 13 +- internal/infra/notify/smtp.go | 7 +- internal/infra/notify/wechat.go | 8 +- internal/infra/transport/lora/chirp_stack.go | 20 +- .../lora_mesh_uart_passthrough_transport.go | 134 ++++++++------ .../transport/lora/placeholder_transport.go | 17 +- internal/infra/transport/transport.go | 11 +- 13 files changed, 293 insertions(+), 242 deletions(-) diff --git a/design/provide-logger-with-mothed/task-infra.md b/design/provide-logger-with-mothed/task-infra.md index c972fa5..595c63e 100644 --- a/design/provide-logger-with-mothed/task-infra.md +++ b/design/provide-logger-with-mothed/task-infra.md @@ -1,157 +1,157 @@ - **`internal/infra/database/storage.go` (`database.Storage`)** - **工厂函数改造 (`NewStorage`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，将接收到的 `ctx` 作为第一个参数传递给 `NewPostgresStorage` 函数。 + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，将接收到的 `ctx` 作为第一个参数传递给 `NewPostgresStorage` 函数。 - **`internal/infra/database/postgres.go`** - **结构体改造**: - - [ ] 移除 `PostgresStorage` 结构体中的 `logger *logs.Logger` 成员。 - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 移除 `PostgresStorage` 结构体中的 `logger *logs.Logger` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewPostgresStorage`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，为 `PostgresStorage` 创建其专属的 `selfCtx`： + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，为 `PostgresStorage` 创建其专属的 `selfCtx`： `selfCtx := logs.AddCompName(ctx, "PostgresStorage")`。 - - [ ] 将这个 `selfCtx` 赋值给 `PostgresStorage` 结构体的 `selfCtx` 成员。 + - [x] 将这个 `selfCtx` 赋值给 `PostgresStorage` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Connect`, `Disconnect`, `Migrate`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, ps.selfCtx, "MethodName")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, ps.selfCtx, "MethodName")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `ps.logger.Info`, `ps.logger.Errorw`, `ps.logger.Debugw` 的调用替换为 `logger.Info`, + - [x] 将所有对 `ps.logger.Info`, `ps.logger.Errorw`, `ps.logger.Debugw` 的调用替换为 `logger.Info`, `logger.Errorw`, `logger.Debugw`。 - - [ ] 在 `Connect` 方法中，调用 `logs.NewGormLogger` 时，将 `newCtx` 传递给它。 - - [ ] 在 `Connect` 方法中，调用 `gorm.Open` 时，使用 + - [x] 在 `Connect` 方法中，调用 `logs.NewGormLogger` 时，将 `newCtx` 传递给它。 + - [x] 在 `Connect` 方法中，调用 `gorm.Open` 时，使用 `ps.db, err = gorm.Open(postgres.Open(ps.connectionString), &gorm.Config{Logger: logs.NewGormLogger(newCtx)})`。 - - [ ] 在 `Migrate` 方法中，确保所有对 `ps.db.AutoMigrate` 和 `ps.db.Exec` 的调用都使用 `newCtx`。 + - [x] 在 `Migrate` 方法中，确保所有对 `ps.db.AutoMigrate` 和 `ps.db.Exec` 的调用都使用 `newCtx`。 - **内部辅助方法改造 (`setupTimescaleDB`, `creatingHyperTable`, `applyCompressionPolicies`, `creatingIndex`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在每个方法入口处，使用 `newCtx, logger := logs.Trace(ctx, ps.selfCtx, "MethodName")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在每个方法入口处，使用 `newCtx, logger := logs.Trace(ctx, ps.selfCtx, "MethodName")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `ps.logger.Info`, `ps.logger.Errorw`, `ps.logger.Debugw`, `ps.logger.Warnw`, `ps.logger.Debug` + - [x] 将所有对 `ps.logger.Info`, `ps.logger.Errorw`, `ps.logger.Debugw`, `ps.logger.Warnw`, `ps.logger.Debug` 的调用替换为 `logger.Info`, `logger.Errorw`, `logger.Debugw`, `logger.Warnw`, `logger.Debug`。 - - [ ] 确保所有对 `ps.db.Exec` 的调用都使用 `newCtx`。 + - [x] 确保所有对 `ps.db.Exec` 的调用都使用 `newCtx`。 - **`internal/infra/notify/log_notifier.go` (`notify.LogNotifier`)** - **结构体改造**: - - [ ] 移除 `logNotifier` 结构体中的 `logger *logs.Logger` 成员。 - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 移除 `logNotifier` 结构体中的 `logger *logs.Logger` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewLogNotifier`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，为 `logNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "LogNotifier")`。 - - [ ] 将这个 `selfCtx` 赋值给 `logNotifier` 结构体的 `selfCtx` 成员。 + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，为 `logNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "LogNotifier")`。 + - [x] 将这个 `selfCtx` 赋值给 `logNotifier` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Send`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "Send")` 获取新的 `context.Context` 和 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "Send")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `l.logger.Infow` 的调用替换为 `logger.Infow`。 + - [x] 将所有对 `l.logger.Infow` 的调用替换为 `logger.Infow`。 - **公共方法 (`Type`)**: - - [ ] 此方法不涉及日志或上下文传递，无需改造。 + - [x] 此方法不涉及日志或上下文传递，无需改造。 - **`internal/infra/notify/lark.go` (`notify.LarkNotifier`)** - **结构体改造**: - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewLarkNotifier`)**: - - [ ] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在函数内部，为 `larkNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "LarkNotifier")`。 - - [ ] 将这个 `selfCtx` 赋值给 `larkNotifier` 结构体的 `selfCtx` 成员。 + - [x] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在函数内部，为 `larkNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "LarkNotifier")`。 + - [x] 将这个 `selfCtx` 赋值给 `larkNotifier` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Send`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "Send")` 获取新的 `context.Context` 和 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "Send")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误，以及 `getAccessToken` 中的错误）通过 `logger.Errorf` + - [x] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误，以及 `getAccessToken` 中的错误）通过 `logger.Errorf` 记录。 - **内部辅助方法改造 (`getAccessToken`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "getAccessToken")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, l.selfCtx, "getAccessToken")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有内部的错误日志通过 `logger.Errorf` 记录。 + - [x] 将所有内部的错误日志通过 `logger.Errorf` 记录。 - **`internal/infra/notify/smtp.go` (`notify.SMTPNotifier`)** - **结构体改造**: - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewSMTPNotifier`)**: - - [ ] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在函数内部，为 `smtpNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "SMTPNotifier")`。 - - [ ] 将这个 `selfCtx` 赋值给 `smtpNotifier` 结构体的 `selfCtx` 成员。 + - [x] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在函数内部，为 `smtpNotifier` 创建其专属的 `selfCtx`：`selfCtx := logs.AddCompName(ctx, "SMTPNotifier")`。 + - [x] 将这个 `selfCtx` 赋值给 `smtpNotifier` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Send`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, s.selfCtx, "Send")` 获取新的 `context.Context` 和 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, s.selfCtx, "Send")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误）通过 `logger.Errorf` 记录。 + - [x] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误）通过 `logger.Errorf` 记录。 - **`internal/infra/notify/wechat.go` (`notify.WechatNotifier`)** - **结构体改造**: - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewWechatNotifier`)**: - - [ ] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在函数内部，为 `wechatNotifier` 创建其专属的 `selfCtx`： + - [x] 修改函数签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在函数内部，为 `wechatNotifier` 创建其专属的 `selfCtx`： `selfCtx := logs.AddCompName(ctx, "WechatNotifier")`。 - - [ ] 将这个 `selfCtx` 赋值给 `wechatNotifier` 结构体的 `selfCtx` 成员。 + - [x] 将这个 `selfCtx` 赋值给 `wechatNotifier` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Send`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, w.selfCtx, "Send")` 获取新的 `context.Context` 和 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, w.selfCtx, "Send")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误，以及 `getAccessToken` 中的错误）通过 `logger.Errorf` + - [x] 将所有内部的错误日志（例如 `fmt.Errorf` 返回的错误，以及 `getAccessToken` 中的错误）通过 `logger.Errorf` 记录。 - **内部辅助方法改造 (`getAccessToken`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, w.selfCtx, "getAccessToken")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, w.selfCtx, "getAccessToken")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有内部的错误日志通过 `logger.Errorf` 记录。 + - [x] 将所有内部的错误日志通过 `logger.Errorf` 记录。 - **`internal/infra/transport/lora/chirp_stack.go` (`lora.ChirpStackTransport`)** - **结构体改造**: - - [ ] 移除 `ChirpStackTransport` 结构体中的 `logger *logs.Logger` 成员。 - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 移除 `ChirpStackTransport` 结构体中的 `logger *logs.Logger` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewChirpStackTransport`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，为 `ChirpStackTransport` 创建其专属的 `selfCtx`： + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，为 `ChirpStackTransport` 创建其专属的 `selfCtx`： `selfCtx := logs.AddCompName(ctx, "ChirpStackTransport")`。 - - [ ] 将这个 `selfCtx` 赋值给 `ChirpStackTransport` 结构体的 `selfCtx` 成员。 + - [x] 将这个 `selfCtx` 赋值给 `ChirpStackTransport` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Send`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, c.selfCtx, "Send")` 获取新的 `context.Context` 和 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, c.selfCtx, "Send")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `c.logger.Errorf` 和 `c.logger.Infof` 的调用替换为 `logger.Errorf` 和 `logger.Infof`。 - - [ ] 在调用 `c.client.DeviceService.DeviceServiceEnqueue` 时，确保将 `newCtx` 传递给 + - [x] 将所有对 `c.logger.Errorf` 和 `c.logger.Infof` 的调用替换为 `logger.Errorf` 和 `logger.Infof`。 + - [x] 在调用 `c.client.DeviceService.DeviceServiceEnqueue` 时，确保将 `newCtx` 传递给 `params.WithContext(newCtx)`，以便 ChirpStack 客户端内部的 HTTP 请求也能携带上下文。 - **`internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go` (`lora.LoRaMeshUartPassthroughTransport`)** - **结构体改造**: - - [ ] 移除 `LoRaMeshUartPassthroughTransport` 结构体中的 `logger *logs.Logger` 成员。 - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 移除 `LoRaMeshUartPassthroughTransport` 结构体中的 `logger *logs.Logger` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewLoRaMeshUartPassthroughTransport`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，为 `LoRaMeshUartPassthroughTransport` 创建其专属的 `selfCtx`： + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，为 `LoRaMeshUartPassthroughTransport` 创建其专属的 `selfCtx`： `selfCtx := logs.AddCompName(ctx, "LoRaMeshUartPassthroughTransport")`。 - - [ ] 将这个 `selfCtx` 赋值给 `LoRaMeshUartPassthroughTransport` 结构体的 `selfCtx` 成员。 + - [x] 将这个 `selfCtx` 赋值给 `LoRaMeshUartPassthroughTransport` 结构体的 `selfCtx` 成员。 - **公共方法改造 (`Listen`, `Send`, `Stop`)**: - - [ ] 修改 `Listen` 方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 修改 `Send` 方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 修改 `Stop` 方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, t.selfCtx, "MethodName")` 获取新的 + - [x] 修改 `Listen` 方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 修改 `Send` 方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 修改 `Stop` 方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, t.selfCtx, "MethodName")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `t.logger.Info`, `t.logger.Errorf` 的调用替换为 `logger.Info`, `logger.Errorf`。 - - [ ] 在 `Send` 方法中，确保 `t.sendChan <- req` 传递的 `req` 包含了 `newCtx`。 + - [x] 将所有对 `t.logger.Info`, `t.logger.Errorf` 的调用替换为 `logger.Info`, `logger.Errorf`。 + - [x] 在 `Send` 方法中，确保 `t.sendChan <- req` 传递的 `req` 包含了 `newCtx`。 - **内部辅助方法改造 (`workerLoop`, `runIdleState`, `runReceivingState`, `executeSend`, `handleFrame`, `handleUpstreamMessage`, `recordSensorData`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在每个方法入口处，使用 `newCtx, logger := logs.Trace(ctx, t.selfCtx, "MethodName")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在每个方法入口处，使用 `newCtx, logger := logs.Trace(ctx, t.selfCtx, "MethodName")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `t.logger.Info`, `t.logger.Errorf`, `t.logger.Warnf`, `t.logger.Infof`, `t.logger.Debugf` + - [x] 将所有对 `t.logger.Info`, `t.logger.Errorf`, `t.logger.Warnf`, `t.logger.Infof`, `t.logger.Debugf` 的调用替换为 `logger.Info`, `logger.Errorf`, `logger.Warnf`, `logger.Infof`, `logger.Debugf`。 - - [ ] 确保所有对 `t.port`, `t.areaControllerRepo`, `t.pendingCollectionRepo`, `t.deviceRepo`, + - [x] 确保所有对 `t.port`, `t.areaControllerRepo`, `t.pendingCollectionRepo`, `t.deviceRepo`, `t.sensorDataRepo` 等依赖的调用都将 `newCtx` 作为第一个参数传递。 - **`internal/infra/transport/lora/placeholder_transport.go` (`lora.PlaceholderTransport`)** - **结构体改造**: - - [ ] 移除 `PlaceholderTransport` 结构体中的 `logger *logs.Logger` 成员。 - - [ ] 新增 `selfCtx context.Context` 成员。 + - [x] 移除 `PlaceholderTransport` 结构体中的 `logger *logs.Logger` 成员。 + - [x] 新增 `selfCtx context.Context` 成员。 - **构造函数改造 (`NewPlaceholderTransport`)**: - - [ ] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 - - [ ] 在函数内部，为 `PlaceholderTransport` 创建其专属的 `selfCtx`： + - [x] 修改函数签名，移除 `logger *logs.Logger` 参数，改为接收 `ctx context.Context`。 + - [x] 在函数内部，为 `PlaceholderTransport` 创建其专属的 `selfCtx`： `selfCtx := logs.AddCompName(ctx, "PlaceholderTransport")`。 - - [ ] 将这个 `selfCtx` 赋值给 `PlaceholderTransport` 结构体的 `selfCtx` 成员。 - - [ ] 使用 `newCtx, logger := logs.Trace(ctx, selfCtx, "NewPlaceholderTransport")` 获取 `logger` 实例，并替换 + - [x] 将这个 `selfCtx` 赋值给 `PlaceholderTransport` 结构体的 `selfCtx` 成员。 + - [x] 使用 `newCtx, logger := logs.Trace(ctx, selfCtx, "NewPlaceholderTransport")` 获取 `logger` 实例，并替换 `logger.Info` 调用。 - **公共方法改造 (`Listen`, `Stop`)**: - - [ ] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 - - [ ] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, p.selfCtx, "MethodName")` 获取新的 + - [x] 修改方法签名，添加 `ctx context.Context` 作为第一个参数。 + - [x] 在方法入口处，使用 `newCtx, logger := logs.Trace(ctx, p.selfCtx, "MethodName")` 获取新的 `context.Context` 和 `logger` 实例。 - - [ ] 将所有对 `p.logger.Warnf` 的调用替换为 `logger.Warnf`。 + - [x] 将所有对 `p.logger.Warnf` 的调用替换为 `logger.Warnf`。 diff --git a/internal/core/component_initializers.go b/internal/core/component_initializers.go index 00704de..9cf119f 100644 --- a/internal/core/component_initializers.go +++ b/internal/core/component_initializers.go @@ -386,7 +386,7 @@ func initNotifyService( // initStorage 封装了数据库的初始化、连接和迁移逻辑。 func initStorage(ctx context.Context, cfg config.DatabaseConfig) (database.Storage, error) { // 创建存储实例 - storage := database.NewStorage(cfg, logs.AddCompName(context.Background(), "Storage")) + storage := database.NewStorage(logs.AddCompName(context.Background(), "Storage"), cfg) if err := storage.Connect(); err != nil { // 错误已在 Connect 内部被记录，这里只需包装并返回 return nil, fmt.Errorf("数据库连接失败: %w", err) diff --git a/internal/infra/database/postgres.go b/internal/infra/database/postgres.go index 1a40804..f11cd63 100644 --- a/internal/infra/database/postgres.go +++ b/internal/infra/database/postgres.go @@ -4,12 +4,13 @@ package database import ( + "context" "fmt" "time" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" - "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "gorm.io/driver/postgres" "gorm.io/gorm" ) @@ -17,35 +18,36 @@ import ( // PostgresStorage 代表基于PostgreSQL的存储实现 // 使用GORM作为ORM库 type PostgresStorage struct { + ctx context.Context db *gorm.DB isTimescaleDB bool connectionString string maxOpenConns int maxIdleConns int connMaxLifetime int - logger *logs.Logger // 依赖注入的 logger } // NewPostgresStorage 创建并返回一个新的PostgreSQL存储实例 // 它接收一个 logger 实例，而不是自己创建 -func NewPostgresStorage(connectionString string, isTimescaleDB bool, maxOpenConns, maxIdleConns, connMaxLifetime int, logger *logs.Logger) *PostgresStorage { +func NewPostgresStorage(ctx context.Context, connectionString string, isTimescaleDB bool, maxOpenConns, maxIdleConns, connMaxLifetime int) *PostgresStorage { return &PostgresStorage{ + ctx: ctx, connectionString: connectionString, isTimescaleDB: isTimescaleDB, maxOpenConns: maxOpenConns, maxIdleConns: maxIdleConns, connMaxLifetime: connMaxLifetime, - logger: logger, // 注入 logger } } // Connect 建立与PostgreSQL数据库的连接 // 使用GORM建立数据库连接，并使用自定义的 logger 接管 GORM 日志 -func (ps *PostgresStorage) Connect() error { - ps.logger.Info("正在连接PostgreSQL数据库") +func (ps *PostgresStorage) Connect(ctx context.Context) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "Connect") + logger.Info("正在连接PostgreSQL数据库") // 创建 GORM 的 logger 适配器 - gormLogger := logs.NewGormLogger(ps.logger) + gormLogger := logs.NewGormLogger(logger) var err error // 在 gorm.Open 时传入我们自定义的 logger @@ -53,19 +55,19 @@ func (ps *PostgresStorage) Connect() error { Logger: gormLogger, }) if err != nil { - ps.logger.Errorw("数据库连接失败", "error", err) + logger.Errorw("数据库连接失败", "error", err) return fmt.Errorf("数据库连接失败: %w", err) // 使用 %w 进行错误包装 } // 测试连接 - sqlDB, err := ps.db.DB() + sqlDB, err := ps.db.WithContext(storageCtx).DB() if err != nil { - ps.logger.Errorw("获取数据库实例失败", "error", err) + logger.Errorw("获取数据库实例失败", "error", err) return fmt.Errorf("获取数据库实例失败: %w", err) } if err = sqlDB.Ping(); err != nil { - ps.logger.Errorw("数据库连接测试失败", "error", err) + logger.Errorw("数据库连接测试失败", "error", err) return fmt.Errorf("数据库连接测试失败: %w", err) } @@ -77,59 +79,62 @@ func (ps *PostgresStorage) Connect() error { // gorm会根据字段名自动创建外键约束, 但触发器Task的PlanID是不存在的, 所以需要关闭, 这个关闭对 ps.db.DisableForeignKeyConstraintWhenMigrating = true - ps.logger.Info("PostgreSQL数据库连接成功") + logger.Info("PostgreSQL数据库连接成功") return nil } // Disconnect 断开与PostgreSQL数据库的连接 // 安全地关闭所有数据库连接 -func (ps *PostgresStorage) Disconnect() error { +func (ps *PostgresStorage) Disconnect(ctx context.Context) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "Disconnect") if ps.db != nil { - ps.logger.Info("正在断开PostgreSQL数据库连接") + logger.Info("正在断开PostgreSQL数据库连接") - sqlDB, err := ps.db.DB() + sqlDB, err := ps.db.WithContext(storageCtx).DB() if err != nil { - ps.logger.Errorw("获取数据库实例失败", "error", err) + logger.Errorw("获取数据库实例失败", "error", err) return fmt.Errorf("获取数据库实例失败: %w", err) } if err := sqlDB.Close(); err != nil { - ps.logger.Errorw("关闭数据库连接失败", "error", err) + logger.Errorw("关闭数据库连接失败", "error", err) return fmt.Errorf("关闭数据库连接失败: %w", err) } - ps.logger.Info("PostgreSQL数据库连接已断开") + logger.Info("PostgreSQL数据库连接已断开") } return nil } // GetDB 获取GORM数据库实例 // 用于执行具体的数据库操作 -func (ps *PostgresStorage) GetDB() *gorm.DB { - return ps.db +func (ps *PostgresStorage) GetDB(ctx context.Context) *gorm.DB { + storageCtx := logs.AddFuncName(ctx, ps.ctx, "GetDB") + return ps.db.WithContext(storageCtx) } // Migrate 执行数据库迁移 -func (ps *PostgresStorage) Migrate(models ...interface{}) error { +func (ps *PostgresStorage) Migrate(ctx context.Context, models ...interface{}) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "Migrate") if len(models) == 0 { - ps.logger.Info("没有需要迁移的数据库模型，跳过迁移步骤") + logger.Info("没有需要迁移的数据库模型，跳过迁移步骤") return nil } - ps.logger.Info("正在自动迁移数据库表结构") - if err := ps.db.AutoMigrate(models...); err != nil { - ps.logger.Errorw("数据库表结构迁移失败", "error", err) + logger.Info("正在自动迁移数据库表结构") + if err := ps.db.WithContext(storageCtx).AutoMigrate(models...); err != nil { + logger.Errorw("数据库表结构迁移失败", "error", err) return fmt.Errorf("数据库表结构迁移失败: %w", err) } - ps.logger.Info("数据库表结构迁移完成") + logger.Info("数据库表结构迁移完成") // -- 处理gorm做不到的初始化逻辑 -- - if err := ps.creatingIndex(); err != nil { + if err := ps.creatingIndex(storageCtx); err != nil { return err } // 如果是 TimescaleDB, 则将部分表转换为 hypertable if ps.isTimescaleDB { - ps.logger.Info("检测到 TimescaleDB, 准备进行超表转换和压缩策略配置") - if err := ps.setupTimescaleDB(); err != nil { + logger.Info("检测到 TimescaleDB, 准备进行超表转换和压缩策略配置") + if err := ps.setupTimescaleDB(storageCtx); err != nil { return err } } @@ -137,18 +142,20 @@ func (ps *PostgresStorage) Migrate(models ...interface{}) error { } // setupTimescaleDB 统一处理所有 TimescaleDB 相关的设置 -func (ps *PostgresStorage) setupTimescaleDB() error { - if err := ps.creatingHyperTable(); err != nil { +func (ps *PostgresStorage) setupTimescaleDB(ctx context.Context) error { + storageCtx := logs.AddFuncName(ctx, ps.ctx, "setupTimescaleDB") + if err := ps.creatingHyperTable(storageCtx); err != nil { return err } - if err := ps.applyCompressionPolicies(); err != nil { + if err := ps.applyCompressionPolicies(storageCtx); err != nil { return err } return nil } // creatingHyperTable 用于在数据库是 TimescaleDB 时创建超表 -func (ps *PostgresStorage) creatingHyperTable() error { +func (ps *PostgresStorage) creatingHyperTable(ctx context.Context) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "creatingHyperTable") // 定义一个辅助结构体来管理超表转换 tablesToConvert := []struct { model interface{ TableName() string } @@ -177,20 +184,21 @@ func (ps *PostgresStorage) creatingHyperTable() error { for _, table := range tablesToConvert { tableName := table.model.TableName() chunkInterval := "1 days" // 统一设置为1天 - ps.logger.Debugw("准备将表转换为超表", "table", tableName, "chunk_interval", chunkInterval) + logger.Debugw("准备将表转换为超表", "table", tableName, "chunk_interval", chunkInterval) sql := fmt.Sprintf("SELECT create_hypertable('%s', '%s', chunk_time_interval => INTERVAL '%s', if_not_exists => TRUE);", tableName, table.timeColumn, chunkInterval) - if err := ps.db.Exec(sql).Error; err != nil { - ps.logger.Errorw("转换为超表失败", "table", tableName, "error", err) + if err := ps.db.WithContext(storageCtx).Exec(sql).Error; err != nil { + logger.Errorw("转换为超表失败", "table", tableName, "error", err) return fmt.Errorf("将 %s 转换为超表失败: %w", tableName, err) } - ps.logger.Debugw("成功将表转换为超表 (或已转换)", "table", tableName) + logger.Debugw("成功将表转换为超表 (或已转换)", "table", tableName) } return nil } // applyCompressionPolicies 为超表配置自动压缩策略 -func (ps *PostgresStorage) applyCompressionPolicies() error { +func (ps *PostgresStorage) applyCompressionPolicies(ctx context.Context) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "applyCompressionPolicies") policies := []struct { model interface{ TableName() string } segmentColumn string @@ -220,50 +228,51 @@ func (ps *PostgresStorage) applyCompressionPolicies() error { compressAfter := "3 days" // 统一设置为2天后（即进入第3天）开始压缩 // 1. 开启表的压缩设置，并指定分段列 - ps.logger.Debugw("为表启用压缩设置", "table", tableName, "segment_by", policy.segmentColumn) + logger.Debugw("为表启用压缩设置", "table", tableName, "segment_by", policy.segmentColumn) // 使用 + 而非Sprintf以规避goland静态检查报错 alterSQL := "ALTER TABLE" + " " + tableName + " SET (timescaledb.compress, timescaledb.compress_segmentby = '" + policy.segmentColumn + "');" - if err := ps.db.Exec(alterSQL).Error; err != nil { + if err := ps.db.WithContext(storageCtx).Exec(alterSQL).Error; err != nil { // 忽略错误，因为这个设置可能是不可变的，重复执行会报错 - ps.logger.Warnw("启用压缩设置时遇到问题 (可能已设置，可忽略)", "table", tableName, "error", err) + logger.Warnw("启用压缩设置时遇到问题 (可能已设置，可忽略)", "table", tableName, "error", err) } - ps.logger.Debugw("成功为表启用压缩设置 (或已启用)", "table", tableName) + logger.Debugw("成功为表启用压缩设置 (或已启用)", "table", tableName) // 2. 添加压缩策略 - ps.logger.Debugw("为表添加压缩策略", "table", tableName, "compress_after", compressAfter) + logger.Debugw("为表添加压缩策略", "table", tableName, "compress_after", compressAfter) policySQL := fmt.Sprintf("SELECT add_compression_policy('%s', INTERVAL '%s', if_not_exists => TRUE);", tableName, compressAfter) - if err := ps.db.Exec(policySQL).Error; err != nil { - ps.logger.Errorw("添加压缩策略失败", "table", tableName, "error", err) + if err := ps.db.WithContext(storageCtx).Exec(policySQL).Error; err != nil { + logger.Errorw("添加压缩策略失败", "table", tableName, "error", err) return fmt.Errorf("为 %s 添加压缩策略失败: %w", tableName, err) } - ps.logger.Debugw("成功为表添加压缩策略 (或已存在)", "table", tableName) + logger.Debugw("成功为表添加压缩策略 (或已存在)", "table", tableName) } return nil } // creatingIndex 用于创建gorm无法处理的索引, 如gin索引 -func (ps *PostgresStorage) creatingIndex() error { +func (ps *PostgresStorage) creatingIndex(ctx context.Context) error { + storageCtx, logger := logs.Trace(ctx, ps.ctx, "creatingIndex") // 使用 IF NOT EXISTS 保证幂等性 // 如果索引已存在，此命令不会报错 // 为 sensor_data 表的 data 字段创建 GIN 索引 - ps.logger.Debug("正在为 sensor_data 表的 data 字段创建 GIN 索引") + logger.Debug("正在为 sensor_data 表的 data 字段创建 GIN 索引") ginSensorDataIndexSQL := "CREATE INDEX IF NOT EXISTS idx_sensor_data_data_gin ON sensor_data USING GIN (data);" - if err := ps.db.Exec(ginSensorDataIndexSQL).Error; err != nil { - ps.logger.Errorw("为 sensor_data 的 data 字段创建 GIN 索引失败", "error", err) + if err := ps.db.WithContext(storageCtx).Exec(ginSensorDataIndexSQL).Error; err != nil { + logger.Errorw("为 sensor_data 的 data 字段创建 GIN 索引失败", "error", err) return fmt.Errorf("为 sensor_data 的 data 字段创建 GIN 索引失败: %w", err) } - ps.logger.Debug("成功为 sensor_data 的 data 字段创建 GIN 索引 (或已存在)") + logger.Debug("成功为 sensor_data 的 data 字段创建 GIN 索引 (或已存在)") // 为 tasks.parameters 创建 GIN 索引 - ps.logger.Debug("正在为 tasks 表的 parameters 字段创建 GIN 索引") + logger.Debug("正在为 tasks 表的 parameters 字段创建 GIN 索引") taskGinIndexSQL := "CREATE INDEX IF NOT EXISTS idx_tasks_parameters_gin ON tasks USING GIN (parameters);" - if err := ps.db.Exec(taskGinIndexSQL).Error; err != nil { - ps.logger.Errorw("为 tasks 的 parameters 字段创建 GIN 索引失败", "error", err) + if err := ps.db.WithContext(storageCtx).Exec(taskGinIndexSQL).Error; err != nil { + logger.Errorw("为 tasks 的 parameters 字段创建 GIN 索引失败", "error", err) return fmt.Errorf("为 tasks 的 parameters 字段创建 GIN 索引失败: %w", err) } - ps.logger.Debug("成功为 tasks 的 parameters 字段创建 GIN 索引 (或已存在)") + logger.Debug("成功为 tasks 的 parameters 字段创建 GIN 索引 (或已存在)") return nil } diff --git a/internal/infra/database/storage.go b/internal/infra/database/storage.go index ceeb84c..b2472e9 100644 --- a/internal/infra/database/storage.go +++ b/internal/infra/database/storage.go @@ -4,10 +4,12 @@ package database import ( + "context" "fmt" "git.huangwc.com/pig/pig-farm-controller/internal/infra/config" "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" + "gorm.io/gorm" ) @@ -15,22 +17,22 @@ import ( // 所有存储实现都需要实现此接口定义的方法 type Storage interface { // Connect 建立与存储后端的连接 - Connect() error + Connect(ctx context.Context) error // Disconnect 断开与存储后端的连接 - Disconnect() error + Disconnect(ctx context.Context) error // GetDB 获取数据库实例 - GetDB() *gorm.DB + GetDB(ctx context.Context) *gorm.DB // Migrate 执行数据库迁移 // 参数为需要迁移的 GORM 模型 - Migrate(models ...interface{}) error + Migrate(ctx context.Context, models ...interface{}) error } // NewStorage 创建并返回一个存储实例 // 根据配置返回相应的存储实现 -func NewStorage(cfg config.DatabaseConfig, logger *logs.Logger) Storage { +func NewStorage(ctx context.Context, cfg config.DatabaseConfig) Storage { // 构建数据库连接字符串 connectionString := fmt.Sprintf( "user=%s password=%s dbname=%s host=%s port=%d sslmode=%s", @@ -45,11 +47,11 @@ func NewStorage(cfg config.DatabaseConfig, logger *logs.Logger) Storage { // 当前默认返回PostgreSQL存储实现，并将 logger 注入 // 当前默认返回PostgreSQL存储实现，并将 logger 注入 return NewPostgresStorage( + logs.AddCompName(context.Background(), "PostgresStorage"), connectionString, - cfg.IsTimescaleDB, // <--- 添加 IsTimescaleDB + cfg.IsTimescaleDB, cfg.MaxOpenConns, cfg.MaxIdleConns, cfg.ConnMaxLifetime, - logger, ) } diff --git a/internal/infra/models/device_template.go b/internal/infra/models/device_template.go index 7b65e77..bbb432b 100644 --- a/internal/infra/models/device_template.go +++ b/internal/infra/models/device_template.go @@ -6,6 +6,7 @@ import ( "fmt" "git.huangwc.com/pig/pig-farm-controller/internal/infra/utils/command_generater" + "gorm.io/datatypes" "gorm.io/gorm" ) diff --git a/internal/infra/notify/lark.go b/internal/infra/notify/lark.go index 586aa91..c1971ca 100644 --- a/internal/infra/notify/lark.go +++ b/internal/infra/notify/lark.go @@ -2,11 +2,14 @@ package notify import ( "bytes" + "context" "encoding/json" "fmt" "net/http" "sync" "time" + + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" ) const ( @@ -18,6 +21,8 @@ const ( // larkNotifier 实现了 Notifier 接口，用于通过飞书自建应用发送私聊消息。 type larkNotifier struct { + ctx context.Context + appID string // 应用 ID appSecret string // 应用密钥 @@ -29,8 +34,9 @@ type larkNotifier struct { // NewLarkNotifier 创建一个新的 larkNotifier 实例。 // 调用者需要注入飞书应用的 AppID 和 AppSecret。 -func NewLarkNotifier(appID, appSecret string) Notifier { +func NewLarkNotifier(ctx context.Context, appID, appSecret string) Notifier { return &larkNotifier{ + ctx: ctx, appID: appID, appSecret: appSecret, } @@ -38,9 +44,10 @@ func NewLarkNotifier(appID, appSecret string) Notifier { // Send 向指定用户发送一条飞书消息卡片。 // toAddr 参数是接收者的邮箱地址。 -func (l *larkNotifier) Send(content AlarmContent, toAddr string) error { +func (l *larkNotifier) Send(ctx context.Context, content AlarmContent, toAddr string) error { + notifierCtx := logs.AddFuncName(ctx, l.ctx, "Send") // 1. 获取有效的 tenant_access_token - token, err := l.getAccessToken() + token, err := l.getAccessToken(notifierCtx) if err != nil { return err } @@ -115,7 +122,7 @@ func (l *larkNotifier) Send(content AlarmContent, toAddr string) error { } // getAccessToken 获取并缓存 tenant_access_token，处理了线程安全和自动刷新。 -func (l *larkNotifier) getAccessToken() (string, error) { +func (l *larkNotifier) getAccessToken(ctx context.Context) (string, error) { l.mu.Lock() defer l.mu.Unlock() diff --git a/internal/infra/notify/log_notifier.go b/internal/infra/notify/log_notifier.go index 887e524..001bdb6 100644 --- a/internal/infra/notify/log_notifier.go +++ b/internal/infra/notify/log_notifier.go @@ -1,26 +1,29 @@ package notify import ( + "context" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" ) // logNotifier 实现了 Notifier 接口，用于将告警信息记录到日志中。 type logNotifier struct { - logger *logs.Logger + ctx context.Context } // NewLogNotifier 创建一个新的 logNotifier 实例。 // 它接收一个日志记录器，用于实际的日志输出。 -func NewLogNotifier(logger *logs.Logger) Notifier { +func NewLogNotifier(ctx context.Context) Notifier { return &logNotifier{ - logger: logger, + ctx: ctx, } } // Send 将告警内容以结构化的方式记录到日志中。 // toAddr 参数在这里表示告警的预期接收者地址，也会被记录。 -func (l *logNotifier) Send(content AlarmContent, toAddr string) error { - l.logger.Infow("告警已记录到日志", +func (l *logNotifier) Send(ctx context.Context, content AlarmContent, toAddr string) error { + logger := logs.TraceLogger(ctx, l.ctx, "Send") + logger.Infow("告警已记录到日志", "notifierType", NotifierTypeLog, "title", content.Title, "message", content.Message, diff --git a/internal/infra/notify/smtp.go b/internal/infra/notify/smtp.go index f889556..131e788 100644 --- a/internal/infra/notify/smtp.go +++ b/internal/infra/notify/smtp.go @@ -1,6 +1,7 @@ package notify import ( + "context" "fmt" "net/smtp" "strings" @@ -8,6 +9,7 @@ import ( // smtpNotifier 实现了 Notifier 接口，用于通过 SMTP 发送邮件通知。 type smtpNotifier struct { + ctx context.Context host string // SMTP 服务器地址 port int // SMTP 服务器端口 username string // 发件人邮箱地址 @@ -17,8 +19,9 @@ type smtpNotifier struct { // NewSMTPNotifier 创建一个新的 smtpNotifier 实例。 // 调用者需要注入 SMTP 相关的配置。 -func NewSMTPNotifier(host string, port int, username, password, sender string) Notifier { +func NewSMTPNotifier(ctx context.Context, host string, port int, username, password, sender string) Notifier { return &smtpNotifier{ + ctx: ctx, host: host, port: port, username: username, @@ -28,7 +31,7 @@ func NewSMTPNotifier(host string, port int, username, password, sender string) N } // Send 使用 net/smtp 包发送一封邮件。 -func (s *smtpNotifier) Send(content AlarmContent, toAddr string) error { +func (s *smtpNotifier) Send(ctx context.Context, content AlarmContent, toAddr string) error { // 1. 设置认证信息 auth := smtp.PlainAuth("", s.username, s.password, s.host) diff --git a/internal/infra/notify/wechat.go b/internal/infra/notify/wechat.go index 9e7e79e..ec5c33d 100644 --- a/internal/infra/notify/wechat.go +++ b/internal/infra/notify/wechat.go @@ -2,6 +2,7 @@ package notify import ( "bytes" + "context" "encoding/json" "fmt" "net/http" @@ -19,6 +20,8 @@ const ( // wechatNotifier 实现了 Notifier 接口，用于通过企业微信自建应用发送私聊消息。 type wechatNotifier struct { + ctx context.Context + corpID string // 企业ID (CorpID) agentID string // 应用ID (AgentID) secret string // 应用密钥 (Secret) @@ -31,8 +34,9 @@ type wechatNotifier struct { // NewWechatNotifier 创建一个新的 wechatNotifier 实例。 // 调用者需要注入企业微信应用的 CorpID, AgentID 和 Secret。 -func NewWechatNotifier(corpID, agentID, secret string) Notifier { +func NewWechatNotifier(ctx context.Context, corpID, agentID, secret string) Notifier { return &wechatNotifier{ + ctx: ctx, corpID: corpID, agentID: agentID, secret: secret, @@ -41,7 +45,7 @@ func NewWechatNotifier(corpID, agentID, secret string) Notifier { // Send 向指定用户发送一条 markdown 格式的私聊消息。 // toAddr 参数是接收者的 UserID 列表，用逗号或竖线分隔。 -func (w *wechatNotifier) Send(content AlarmContent, toAddr string) error { +func (w *wechatNotifier) Send(ctx context.Context, content AlarmContent, toAddr string) error { // 1. 获取有效的 access_token token, err := w.getAccessToken() if err != nil { diff --git a/internal/infra/transport/lora/chirp_stack.go b/internal/infra/transport/lora/chirp_stack.go index 6bc4a2c..92a0925 100644 --- a/internal/infra/transport/lora/chirp_stack.go +++ b/internal/infra/transport/lora/chirp_stack.go @@ -1,32 +1,33 @@ package lora import ( + "context" "errors" "time" "git.huangwc.com/pig/pig-farm-controller/internal/infra/config" "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/lora/chirp_stack_proto/client" "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/lora/chirp_stack_proto/client/device_service" + "github.com/go-openapi/runtime" httptransport "github.com/go-openapi/runtime/client" "github.com/go-openapi/strfmt" - - "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/lora/chirp_stack_proto/client" ) // ChirpStackTransport 是一个客户端，用于封装与 ChirpStack REST API 的交互。 type ChirpStackTransport struct { + ctx context.Context client *client.ChirpStackRESTAPI authInfo runtime.ClientAuthInfoWriter config config.ChirpStackConfig - logger *logs.Logger } // NewChirpStackTransport 创建一个新的通信实例，用于与 ChirpStack 通信。 func NewChirpStackTransport( + ctx context.Context, config config.ChirpStackConfig, - logger *logs.Logger, ) *ChirpStackTransport { // 使用配置中的服务器地址创建一个 HTTP transport。 // 它会使用生成的客户端中定义的默认 base path 和 schemes。 @@ -39,14 +40,15 @@ func NewChirpStackTransport( authInfo := httptransport.APIKeyAuth("grpc-metadata-authorization", "header", config.GenerateAPIKey()) return &ChirpStackTransport{ + ctx: ctx, client: apiClient, authInfo: authInfo, config: config, - logger: logger, } } -func (c *ChirpStackTransport) Send(address string, payload []byte) (*transport.SendResult, error) { +func (c *ChirpStackTransport) Send(ctx context.Context, address string, payload []byte) (*transport.SendResult, error) { + logger := logs.TraceLogger(ctx, c.ctx, "Send") // 1. 构建 API 请求体。 // - Confirmed: true 表示确认消息, 设为false将不保证消息送达(但可以节约下行容量)。 // - Data: 经过 Base64 编码的数据。 @@ -72,18 +74,18 @@ func (c *ChirpStackTransport) Send(address string, payload []byte) (*transport.S // c.authInfo 是您在 NewChirpStackTransport 中创建的认证信息。 resp, err := c.client.DeviceService.DeviceServiceEnqueue(params, c.authInfo) if err != nil { - c.logger.Errorf("设备 %s 调用ChirpStack Enqueue失败: %v", address, err) + logger.Errorf("设备 %s 调用ChirpStack Enqueue失败: %v", address, err) return nil, err } if resp == nil || resp.Payload == nil || resp.Payload.ID == "" { // 这是一个需要明确处理的错误情况，因为调用方依赖 MessageID。 errMsg := "ChirpStack Enqueue 响应未包含 MessageID (ID)" - c.logger.Errorf(errMsg) + logger.Errorf(errMsg) return nil, errors.New(errMsg) } - c.logger.Infof("成功将 payload 发送到设备 %s 的队列 (MessageID: %s)", address, resp.Payload.ID) + logger.Infof("成功将 payload 发送到设备 %s 的队列 (MessageID: %s)", address, resp.Payload.ID) // 将 MessageID 包装在 SendResult 中返回 result := &transport.SendResult{ diff --git a/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go b/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go index 7212f77..bd51957 100644 --- a/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go +++ b/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go @@ -2,6 +2,7 @@ package lora import ( "bytes" + "context" "encoding/binary" "encoding/json" "fmt" @@ -17,6 +18,7 @@ import ( "git.huangwc.com/pig/pig-farm-controller/internal/infra/repository" "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport" "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto" + "github.com/google/uuid" "github.com/tarm/serial" gproto "google.golang.org/protobuf/proto" @@ -41,8 +43,8 @@ type message struct { // LoRaMeshUartPassthroughTransport 实现了 transport.Communicator 和 transport.Listener 接口 type LoRaMeshUartPassthroughTransport struct { + ctx context.Context config config.LoraMeshConfig - logger *logs.Logger port *serial.Port mu sync.Mutex // 用于保护对外的公共方法（如Send）的并发调用 @@ -87,8 +89,8 @@ type reassemblyBuffer struct { // NewLoRaMeshUartPassthroughTransport 创建一个新的 LoRaMeshUartPassthroughTransport 实例 func NewLoRaMeshUartPassthroughTransport( + ctx context.Context, config config.LoraMeshConfig, - logger *logs.Logger, areaControllerRepo repository.AreaControllerRepository, pendingCollectionRepo repository.PendingCollectionRepository, deviceRepo repository.DeviceRepository, @@ -106,8 +108,8 @@ func NewLoRaMeshUartPassthroughTransport( } t := &LoRaMeshUartPassthroughTransport{ + ctx: ctx, config: config, - logger: logger, port: port, state: stateIdle, stopChan: make(chan struct{}), @@ -126,15 +128,16 @@ func NewLoRaMeshUartPassthroughTransport( } // Listen 启动后台监听协程（非阻塞） -func (t *LoRaMeshUartPassthroughTransport) Listen() error { +func (t *LoRaMeshUartPassthroughTransport) Listen(ctx context.Context) error { + loraCtx, logger := logs.Trace(ctx, t.ctx, "Listen") t.wg.Add(1) - go t.workerLoop() - t.logger.Info("LoRa传输层工作协程已启动") + go t.workerLoop(loraCtx) + logger.Info("LoRa传输层工作协程已启动") return nil } // Send 将发送任务提交给worker协程 -func (t *LoRaMeshUartPassthroughTransport) Send(address string, payload []byte) (*transport.SendResult, error) { +func (t *LoRaMeshUartPassthroughTransport) Send(ctx context.Context, address string, payload []byte) (*transport.SendResult, error) { t.mu.Lock() defer t.mu.Unlock() @@ -156,14 +159,16 @@ func (t *LoRaMeshUartPassthroughTransport) Send(address string, payload []byte) } // Stop 停止传输层 -func (t *LoRaMeshUartPassthroughTransport) Stop() error { +func (t *LoRaMeshUartPassthroughTransport) Stop(ctx context.Context) error { close(t.stopChan) t.wg.Wait() return t.port.Close() } // workerLoop 是核心的状态机和调度器 -func (t *LoRaMeshUartPassthroughTransport) workerLoop() { +func (t *LoRaMeshUartPassthroughTransport) workerLoop(ctx context.Context) { + loraCtx, logger := logs.Trace(ctx, t.ctx, "workerLoop") + defer t.wg.Done() readBuffer := make([]byte, 1024) @@ -176,7 +181,7 @@ func (t *LoRaMeshUartPassthroughTransport) workerLoop() { if t.reassemblyTimeout != nil { t.reassemblyTimeout.Stop() } - t.logger.Info("LoRa传输层工作协程已停止") + logger.Info("LoRa传输层工作协程已停止") return default: } @@ -188,7 +193,7 @@ func (t *LoRaMeshUartPassthroughTransport) workerLoop() { } if err != nil && err != io.EOF { // 忽略预期的超时错误(io.EOF)，只记录真正的IO错误 - t.logger.Errorf("从串口读取数据时发生错误: %v", err) + logger.Errorf("从串口读取数据时发生错误: %v", err) } // 3. 循环解析缓冲区中的完整物理帧 @@ -197,27 +202,29 @@ func (t *LoRaMeshUartPassthroughTransport) workerLoop() { if frame == nil { break // 缓冲区中没有更多完整帧了 } - t.handleFrame(frame) + t.handleFrame(loraCtx, frame) } // 4. 根据当前状态执行主要逻辑 switch t.state { case stateIdle: - t.runIdleState() + t.runIdleState(loraCtx) case stateReceiving: - t.runReceivingState() + t.runReceivingState(loraCtx) default: } } } // runIdleState 处理空闲状态下的逻辑，主要是检查并启动发送任务 -func (t *LoRaMeshUartPassthroughTransport) runIdleState() { +func (t *LoRaMeshUartPassthroughTransport) runIdleState(ctx context.Context) { + loraCtx := logs.AddFuncName(ctx, t.ctx, "Listen") + select { case req := <-t.sendChan: t.state = stateSending // 此处为阻塞式发送 - result, err := t.executeSend(req) + result, err := t.executeSend(loraCtx, req) req.result <- &sendResultTuple{result: result, err: err} t.state = stateIdle default: @@ -226,10 +233,11 @@ func (t *LoRaMeshUartPassthroughTransport) runIdleState() { } // runReceivingState 处理接收状态下的逻辑，主要是检查超时 -func (t *LoRaMeshUartPassthroughTransport) runReceivingState() { +func (t *LoRaMeshUartPassthroughTransport) runReceivingState(ctx context.Context) { + logger := logs.TraceLogger(ctx, t.ctx, "runReceivingState") select { case sourceAddr := <-t.reassemblyTimeoutCh: - t.logger.Warnf("接收来自 0x%04X 的消息超时", sourceAddr) + logger.Warnf("接收来自 0x%04X 的消息超时", sourceAddr) delete(t.reassemblyBuffers, sourceAddr) t.state = stateIdle default: @@ -238,7 +246,8 @@ func (t *LoRaMeshUartPassthroughTransport) runReceivingState() { } // executeSend 执行完整的发送流程（分片、构建、写入） -func (t *LoRaMeshUartPassthroughTransport) executeSend(req *sendRequest) (*transport.SendResult, error) { +func (t *LoRaMeshUartPassthroughTransport) executeSend(ctx context.Context, req *sendRequest) (*transport.SendResult, error) { + logger := logs.TraceLogger(ctx, t.ctx, "executeSend") chunks := splitPayload(req.payload, t.config.MaxChunkSize) totalChunks := uint8(len(chunks)) @@ -257,7 +266,7 @@ func (t *LoRaMeshUartPassthroughTransport) executeSend(req *sendRequest) (*trans frame.WriteByte(currentChunk) // 当前包序号 frame.Write(chunk) // 数据块 - t.logger.Infof("构建LoRa数据包: %v", frame.Bytes()) + logger.Infof("构建LoRa数据包: %v", frame.Bytes()) _, err := t.port.Write(frame.Bytes()) if err != nil { return nil, fmt.Errorf("写入串口失败: %w", err) @@ -272,9 +281,10 @@ func (t *LoRaMeshUartPassthroughTransport) executeSend(req *sendRequest) (*trans } // handleFrame 处理一个从串口解析出的完整物理帧 -func (t *LoRaMeshUartPassthroughTransport) handleFrame(frame []byte) { +func (t *LoRaMeshUartPassthroughTransport) handleFrame(ctx context.Context, frame []byte) { + loraCtx, logger := logs.Trace(ctx, t.ctx, "handleFrame") if len(frame) < 8 { - t.logger.Warnf("收到了一个无效长度的帧: %d", len(frame)) + logger.Warnf("收到了一个无效长度的帧: %d", len(frame)) return } @@ -291,7 +301,7 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(frame []byte) { DestAddr: fmt.Sprintf("%04X", destAddr), Payload: chunkData, } - go t.handleUpstreamMessage(msg) + go t.handleUpstreamMessage(loraCtx, msg) return } @@ -316,18 +326,18 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(frame []byte) { t.reassemblyTimeoutCh <- sourceAddr }) } else { - t.logger.Warnf("在空闲状态下收到了一个来自 0x%04X 的非首包分片，已忽略。", sourceAddr) + logger.Warnf("在空闲状态下收到了一个来自 0x%04X 的非首包分片，已忽略。", sourceAddr) } case stateReceiving: if sourceAddr != t.currentRecvSource { - t.logger.Warnf("正在接收来自 0x%04X 的数据时，收到了另一个源 0x%04X 的分片，已忽略。", t.currentRecvSource, sourceAddr) + logger.Warnf("正在接收来自 0x%04X 的数据时，收到了另一个源 0x%04X 的分片，已忽略。", t.currentRecvSource, sourceAddr) return } buffer, ok := t.reassemblyBuffers[sourceAddr] if !ok { - t.logger.Errorf("内部错误: 处于接收状态，但没有为 0x%04X 找到缓冲区", sourceAddr) + logger.Errorf("内部错误: 处于接收状态，但没有为 0x%04X 找到缓冲区", sourceAddr) t.state = stateIdle // 重置状态 return } @@ -352,23 +362,27 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(frame []byte) { DestAddr: fmt.Sprintf("%04X", destAddr), Payload: fullPayload.Bytes(), } - go t.handleUpstreamMessage(msg) + go t.handleUpstreamMessage(loraCtx, msg) // 清理并返回空闲状态 delete(t.reassemblyBuffers, sourceAddr) t.state = stateIdle } + default: + logger.Errorf("内部错误: 状态机处于未知状态 %d", t.state) } } // handleUpstreamMessage 在独立的协程中处理单个上行的、完整的消息。 -func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(msg *message) { - t.logger.Infof("开始处理来自 %s 的上行消息", msg.SourceAddr) +func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(ctx context.Context, msg *message) { + loraCtx, logger := logs.Trace(ctx, t.ctx, "handleUpstreamMessage") + + logger.Infof("开始处理来自 %s 的上行消息", msg.SourceAddr) // 1. 解析外层 "信封" var instruction proto.Instruction if err := gproto.Unmarshal(msg.Payload, &instruction); err != nil { - t.logger.Errorf("解析上行 Instruction Protobuf 失败: %v, 源地址: %s, 原始数据: %x", err, msg.SourceAddr, msg.Payload) + logger.Errorf("解析上行 Instruction Protobuf 失败: %v, 源地址: %s, 原始数据: %x", err, msg.SourceAddr, msg.Payload) return } @@ -379,39 +393,39 @@ func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(msg *message) { collectResp = p.CollectResult default: // 如果上行的数据不是采集结果，记录日志并忽略 - t.logger.Infof("收到一个非采集响应的上行指令 (类型: %T)，无需处理。源地址: %s", p, msg.SourceAddr) + logger.Infof("收到一个非采集响应的上行指令 (类型: %T)，无需处理。源地址: %s", p, msg.SourceAddr) return } if collectResp == nil { - t.logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil。源地址: %s", msg.SourceAddr) + logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil。源地址: %s", msg.SourceAddr) return } correlationID := collectResp.CorrelationId - t.logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values)) + logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values)) // 3. 查找区域主控 (注意：LoRa Mesh 的 SourceAddr 对应于区域主控的 NetworkID) - regionalController, err := t.areaControllerRepo.FindByNetworkID(msg.SourceAddr) + regionalController, err := t.areaControllerRepo.FindByNetworkID(loraCtx, msg.SourceAddr) if err != nil { - t.logger.Errorf("处理上行消息失败：无法通过源地址 '%s' 找到区域主控设备: %v", msg.SourceAddr, err) + logger.Errorf("处理上行消息失败：无法通过源地址 '%s' 找到区域主控设备: %v", msg.SourceAddr, err) return } if err := regionalController.SelfCheck(); err != nil { - t.logger.Errorf("处理上行消息失败：区域主控 %v(ID: %d) 未通过自检: %v", regionalController.Name, regionalController.ID, err) + logger.Errorf("处理上行消息失败：区域主控 %v(ID: %d) 未通过自检: %v", regionalController.Name, regionalController.ID, err) return } // 4. 根据 CorrelationID 查找待处理请求 - pendingReq, err := t.pendingCollectionRepo.FindByCorrelationID(correlationID) + pendingReq, err := t.pendingCollectionRepo.FindByCorrelationID(loraCtx, correlationID) if err != nil { - t.logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err) + logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err) return } // 检查状态，防止重复处理 if pendingReq.Status != models.PendingStatusPending && pendingReq.Status != models.PendingStatusTimedOut { - t.logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status) + logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status) return } @@ -419,10 +433,10 @@ func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(msg *message) { deviceIDs := pendingReq.CommandMetadata values := collectResp.Values if len(deviceIDs) != len(values) { - t.logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID) - err = t.pendingCollectionRepo.UpdateStatusToFulfilled(correlationID, time.Now()) + logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID) + err = t.pendingCollectionRepo.UpdateStatusToFulfilled(loraCtx, correlationID, time.Now()) if err != nil { - t.logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err) + logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err) } return } @@ -431,31 +445,31 @@ func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(msg *message) { rawSensorValue := values[i] if math.IsNaN(float64(rawSensorValue)) { - t.logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID) + logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID) continue } - dev, err := t.deviceRepo.FindByID(deviceID) + dev, err := t.deviceRepo.FindByID(loraCtx, deviceID) if err != nil { - t.logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err) + logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err) continue } if err := dev.SelfCheck(); err != nil { - t.logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err) continue } if err := dev.DeviceTemplate.SelfCheck(); err != nil { - t.logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err) continue } var valueDescriptors []*models.ValueDescriptor if err := dev.DeviceTemplate.ParseValues(&valueDescriptors); err != nil { - t.logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err) + logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err) continue } if len(valueDescriptors) == 0 { - t.logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID) + logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID) continue } valueDescriptor := valueDescriptors[0] @@ -471,27 +485,29 @@ func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(msg *message) { case models.SensorTypeWeight: dataToRecord = models.WeightData{WeightKilograms: parsedValue} default: - t.logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type) + logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type) dataToRecord = map[string]float64{"value": parsedValue} } - t.recordSensorData(regionalController.ID, dev.ID, time.Now(), valueDescriptor.Type, dataToRecord) - t.logger.Infof("成功记录传感器数据: 设备ID=%d, 类型=%s, 原始值=%f, 解析值=%.2f", dev.ID, valueDescriptor.Type, rawSensorValue, parsedValue) + t.recordSensorData(loraCtx, regionalController.ID, dev.ID, time.Now(), valueDescriptor.Type, dataToRecord) + logger.Infof("成功记录传感器数据: 设备ID=%d, 类型=%s, 原始值=%f, 解析值=%.2f", dev.ID, valueDescriptor.Type, rawSensorValue, parsedValue) } // 6. 更新请求状态为“已完成” - if err := t.pendingCollectionRepo.UpdateStatusToFulfilled(correlationID, time.Now()); err != nil { - t.logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err) + if err := t.pendingCollectionRepo.UpdateStatusToFulfilled(loraCtx, correlationID, time.Now()); err != nil { + logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err) } else { - t.logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID) + logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID) } } // recordSensorData 是一个通用方法，用于将传感器数据存入数据库。 -func (t *LoRaMeshUartPassthroughTransport) recordSensorData(regionalControllerID uint, sensorDeviceID uint, eventTime time.Time, sensorType models.SensorType, data interface{}) { +func (t *LoRaMeshUartPassthroughTransport) recordSensorData(ctx context.Context, regionalControllerID uint, sensorDeviceID uint, eventTime time.Time, sensorType models.SensorType, data interface{}) { + loraCtx, logger := logs.Trace(ctx, t.ctx, "recordSensorData") + jsonData, err := json.Marshal(data) if err != nil { - t.logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err) + logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err) return } @@ -503,8 +519,8 @@ func (t *LoRaMeshUartPassthroughTransport) recordSensorData(regionalControllerID Data: datatypes.JSON(jsonData), } - if err := t.sensorDataRepo.Create(sensorData); err != nil { - t.logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err) + if err := t.sensorDataRepo.Create(loraCtx, sensorData); err != nil { + logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err) } } diff --git a/internal/infra/transport/lora/placeholder_transport.go b/internal/infra/transport/lora/placeholder_transport.go index 967b0b9..90091ec 100644 --- a/internal/infra/transport/lora/placeholder_transport.go +++ b/internal/infra/transport/lora/placeholder_transport.go @@ -1,27 +1,28 @@ package lora import ( + "context" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/transport" ) type PlaceholderTransport struct { - logger *logs.Logger + ctx context.Context } -func NewPlaceholderTransport(logger *logs.Logger) transport.Listener { - logger.Info("当前配置非 LoRaMesh, LoRaMesh UART 透传传输器未激活。") +func NewPlaceholderTransport(ctx context.Context) transport.Listener { return &PlaceholderTransport{ - logger: logger, + ctx: ctx, } } -func (p *PlaceholderTransport) Listen() error { - p.logger.Warnf("当前不是LoRa Mesh 模式, 这只是个占位监听器") +func (p *PlaceholderTransport) Listen(ctx context.Context) error { + logs.TraceLogger(ctx, p.ctx, "Listen").Warnf("当前不是LoRa Mesh 模式, 这只是个占位监听器") return nil } -func (p *PlaceholderTransport) Stop() error { - p.logger.Warnf("当前不是LoRa Mesh 模式, 占位监听器停止工作") +func (p *PlaceholderTransport) Stop(ctx context.Context) error { + logs.TraceLogger(ctx, p.ctx, "Stop").Warnf("当前不是LoRa Mesh 模式, 占位监听器停止工作") return nil } diff --git a/internal/infra/transport/transport.go b/internal/infra/transport/transport.go index fc29ab8..ff5a3d4 100644 --- a/internal/infra/transport/transport.go +++ b/internal/infra/transport/transport.go @@ -1,12 +1,15 @@ package transport -import "time" +import ( + "context" + "time" +) // Communicator 用于其他设备通信 type Communicator interface { // Send 用于发送一条单向数据(不等待回信) // 成功时，它返回一个包含 MessageID 的 SendResult，以便调用方追踪。 - Send(address string, payload []byte) (*SendResult, error) + Send(ctx context.Context, address string, payload []byte) (*SendResult, error) } // SendResult 包含了 SendGo 方法成功执行后返回的结果。 @@ -27,8 +30,8 @@ type SendResult struct { // Listener 用于监听其他设备发送过来的数据 type Listener interface { // Listen 用于开始监听其他设备发送过来的数据 - Listen() error + Listen(ctx context.Context) error // Stop 用于停止监听 - Stop() error + Stop(ctx context.Context) error }