package channelbridge // 本文件承载渠道桥接器的热重载机制(design/04 // §3.1)。面板把某平台的账号配置保存到 channels.json 并调用 // Bridge.Reload(platform):工厂重建该平台的适配器,它们 // 先连接(新消息使用新配置),随后旧适配器 // 被优雅停止。已在途的消息仍在其旧适配器上完成, // 因为消息处理是以桥接器生命周期 context 派生的,而非 // 重载所取消的按适配器读取 context。 import ( "context" "errors" "fmt" "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" ) // PlatformFactory 依据平台当前生效的账号配置 // (channels.json,回退到环境变量)重建其适配器。它以注入方式提供, // 使桥接器能在面板保存新配置时热替换该平台的适配器 // 而无需重启。具体实现位于组合根 // (cmd/cyrene),平台适配器的构造函数由那里持有。 type PlatformFactory interface { // Build 依据生效的账号配置返回该平台的适配器。 // 平台没有已配置账号时,返回的切片可能为空。 Build(platform string) ([]adapter.PlatformAdapter, error) } // ChannelRegistry 是能力注册表的**窄端口**,只暴露热重载需要同步的两件事: // 把某个名字的渠道条目换成新适配器、把已消失的账号条目删掉。桥接器用它让管理面板 // (渠道列表 / 在线状态 / open·close 解析)读到的永远是**当前**适配器实例,而不必 // 依赖 capability/registry 的具体类型 —— 注册表实现与桥接器因此保持解耦。 // // 未接线(Options.Channels 为 nil)时桥接器照常热重载,但注册表条目会停在被保存前 // 的那个实例上:面板显示「未连接」、点「连接」驱动的是旧实例。此时只记一次 WARN。 type ChannelRegistry interface { // ReplaceChannel 用同名的新适配器覆盖(或新增)渠道条目。 ReplaceChannel(ch sdk.Channel) error // RemoveChannel 删除该名字的渠道条目。 RemoveChannel(name string) error } // ErrNilFactory 表示请求重载时未注入平台工厂 // (没有热重载配置来源)。 var ErrNilFactory = errors.New("channelbridge: nil platform factory") // ErrBridgeClosed 表示在已停止的桥接器上请求重载。 var ErrBridgeClosed = errors.New("channelbridge: bridge is not running") // connectAndRun 拨号单个适配器(受 connectTimeout 约束),成功时 // 装配其入站处理器,并在按适配器 context 下启动其读取 // 循环。热重载正是取消这个按适配器 context 来停止旧 // 读取循环;消息处理则以桥接器生命周期 context 派生, // 因此 Reload 之前已在途的消息会在其旧适配器上完成。 func (b *Bridge) connectAndRun(a adapter.PlatformAdapter) { if a == nil { return } if !b.configured(a) { b.log.Warn("channel skipped: no credentials", "channel", a.Name()) return } connCtx, cancel := context.WithTimeout(context.Background(), connectTimeout) err := a.Connect(connCtx) cancel() if err != nil { // 原始 dial 错误里带着整条拨号地址(`?access_token=…`),不能原样落 WARN; // connectFailureAttrs 只改文案,不动这里的判定与分支(见 connectlog.go)。 b.log.Warn("channel connect failed, leaving disconnected", connectFailureAttrs(a.Name(), err)...) return } // 闭包捕获该适配器,使入站回调知道是哪个账号收到了 // 消息——多账号回复路由,以及读取该账号的 // 允许/拒绝名单都依赖它(design/16 §4.6 §5)。 a.SetMessageHandler(func(_ context.Context, msg adapter.UnifiedMessage) { b.onMessage(b.ctx, msg, a) }) b.mu.Lock() if b.closed { b.mu.Unlock() ctx, c := context.WithTimeout(context.Background(), connectTimeout) _ = a.Disconnect(ctx) c() return } readCtx, rcancel := context.WithCancel(b.ctx) b.readCancels[a.Name()] = rcancel b.connected[a.Name()] = true b.mu.Unlock() // 适配器已上线,此时缓存该账号的渠道级角色: // 入站路径读取缓存而不是配置,而重载(由面板保存 // 触发)会重跑这里以拾取新的绑定。 b.rememberAdapterPersona(a.Name()) b.wg.Add(1) go b.readLoop(a, readCtx) b.log.Info("channel connected", "channel", a.Name()) } // Reload 依据平台当前生效的配置(经由注入的工厂)重建其适配器, // 并在不重启的前提下热替换正在运行的适配器 // (design/04 §3.1)。重建后的适配器先连接;到达旧适配器的 // 消息在旧适配器上完成,新消息则由新配置 // 接管。被移除的账号会停掉其旧适配器;名称未变的 // 账号在新适配器上线后停掉旧适配器。 func (b *Bridge) Reload(platform string) error { if b.factory == nil { return ErrNilFactory } b.mu.Lock() started := b.started closed := b.closed b.mu.Unlock() if !started || closed { return ErrBridgeClosed } newAdapters, err := b.factory.Build(platform) if err != nil { return fmt.Errorf("channelbridge: build %s: %w", platform, err) } old, oldCancels := b.adapterSnap(platform) newNames := make(map[string]bool, len(newAdapters)) for _, a := range newAdapters { newNames[a.Name()] = true } for _, a := range newAdapters { b.connectAndRun(a) // 紧跟上线之后同步能力注册表:新实例先连上,面板才改为读它, // 避免「注册表已指向新实例、它却还没连」的窗口。 b.replaceChannelEntry(a) } for _, o := range old { if newNames[o.Name()] { // 同名账号被重建后的适配器替换:只停旧实例的读取循环, // 注册表条目已由 replaceChannelEntry 指向新实例。 b.cancelAndDisconnect(o, oldCancels[o.Name()]) continue } // 账号被彻底移除(或更名):停止它并删除其注册表 // 条目,后续 snapshot 与面板便不再看到它。 b.removeAdapter(o, oldCancels[o.Name()]) } b.log.Info("channel reloaded", "platform", platform, "adapters", len(newAdapters)) return nil } // adapterSnap 返回某平台当前已注册的适配器,以及 // 控制各自读取循环的 context 取消函数。Reload 用它 // 在重建的适配器上线后停止旧适配器。 func (b *Bridge) adapterSnap(platform string) ([]adapter.PlatformAdapter, map[string]context.CancelFunc) { b.mu.RLock() defer b.mu.RUnlock() var list []adapter.PlatformAdapter cancels := make(map[string]context.CancelFunc) for name, a := range b.adapters { if basePlatform(name) != platform { continue } list = append(list, a) if c, ok := b.readCancels[name]; ok { cancels[name] = c } } return list, cancels } // replaceChannelEntry 把刚上线的适配器写回能力注册表。注册表里的渠道条目只在启动 // 时写过一次,而热重载重建出来的是同名的新实例:不同步的话,管理面板(渠道列表、 // 账号在线状态、open/close 的目标解析)读到的仍是保存前的**旧实例** —— 卡片显示 // 「未连接」,点「连接」拨的是旧配置里的地址(design/04 §3.1)。 // // 同步失败只记日志、绝不返回错误:面板保存不该被注册表同步拖垮。 func (b *Bridge) replaceChannelEntry(a adapter.PlatformAdapter) { if a == nil { return } name := a.Name() if !b.IsConnected(name) { // 没上线就不动注册表:换成连不上的实例只会让面板的在线状态更失真。 b.log.Warn("channel registry left unchanged: adapter not connected", "channel", name) return } if b.channels == nil { b.warnChannelRegistryUnwired() return } if err := b.channels.ReplaceChannel(a); err != nil { b.log.Warn("channel registry replace failed", "channel", name, "error", err) } } // dropChannelEntry 在账号被移除(或更名)后删掉它的注册表条目,使管理面板不再列出 // 这条已经不存在的账号。同 replaceChannelEntry:失败只记日志。 func (b *Bridge) dropChannelEntry(name string) { if b.channels == nil { b.warnChannelRegistryUnwired() return } if err := b.channels.RemoveChannel(name); err != nil { b.log.Warn("channel registry remove failed", "channel", name, "error", err) } } // warnChannelRegistryUnwired 只在首次同步时提醒一次:未注入 Channels 端口时热重载 // 照常进行(适配器确实换了),只是面板读到的是保存前的旧实例。 func (b *Bridge) warnChannelRegistryUnwired() { b.channelsWarnOnce.Do(func() { b.log.Warn("channel registry not wired: admin panel keeps the pre-reload adapter instance") }) } // cancelAndDisconnect 停止适配器的读取循环并拆除其会话。 // 它绝不改动桥接器自己的适配器表与能力注册表:替换同名适配器 // 的调用方希望(新)条目保留,而移除账号的调用方使用 // removeAdapter(它会连注册表条目一起清掉)。 func (b *Bridge) cancelAndDisconnect(o adapter.PlatformAdapter, c context.CancelFunc) { if c != nil { c() } ctx, cancel := context.WithTimeout(context.Background(), connectTimeout) _ = o.Disconnect(ctx) cancel() b.log.Info("channel (reload) stopped", "channel", o.Name()) } // removeAdapter 停止账号的适配器并删除其注册表条目。它 // 只对未被 Reload 重建的名称调用,因此删除绝不会 // 误删如今占用该名称的替换适配器。 func (b *Bridge) removeAdapter(o adapter.PlatformAdapter, c context.CancelFunc) { b.cancelAndDisconnect(o, c) name := o.Name() b.mu.Lock() delete(b.adapters, name) delete(b.connected, name) delete(b.readCancels, name) b.mu.Unlock() b.forgetAdapterPersona(name) b.dropChannelEntry(name) b.log.Info("channel (reload) removed", "channel", name) }