feat(imap): 新邮件实时推送(IDLE)+ 后台当前连接监控
- IMAP 推送:imapBackend 实现 backend.BackendUpdater,SMTP 本地投递 与 Web 写信投递成功后 NotifyNewMessage,挂起 IDLE 的客户端即时收到 新邮件 FETCH 通知(按用户名+INBOX 过滤广播,通道满非阻塞丢弃) - 当前连接:新增 internal/connhub 连接注册中心,SMTP/IMAP/POP3 三协议 注册/注销/用户名/TLS/活跃时间追踪;后台新增「当前连接」页 (/admin/connections,统计卡片+连接表格,每 5 秒自动刷新) - 新增测试:connhub 并发安全、推送内容/非阻塞/nil 安全、 后台页面渲染;全量 -race 通过
This commit is contained in:
@@ -6,8 +6,11 @@ import (
|
||||
"log"
|
||||
"net"
|
||||
"strconv"
|
||||
"sync"
|
||||
|
||||
"mail_go/config"
|
||||
"mail_go/internal/connhub"
|
||||
"mail_go/internal/db"
|
||||
"mail_go/internal/store"
|
||||
"mail_go/internal/tlsutil"
|
||||
|
||||
@@ -21,19 +24,56 @@ type IMAPServer struct {
|
||||
cfg config.IMAPConfig
|
||||
banCfg config.BanConfig
|
||||
tlsLoader *tlsutil.Loader
|
||||
hub *connhub.Hub
|
||||
|
||||
beMu sync.Mutex
|
||||
bes []*imapBackend // 各监听器(明文/TLS)的 backend,用于新邮件推送
|
||||
}
|
||||
|
||||
// NewIMAPServer creates a new IMAP server instance. tlsLoader may be nil
|
||||
// when TLS is not configured.
|
||||
func NewIMAPServer(cfg config.IMAPConfig, stores *store.Stores, tlsLoader *tlsutil.Loader, banCfg config.BanConfig) *IMAPServer {
|
||||
func NewIMAPServer(cfg config.IMAPConfig, stores *store.Stores, tlsLoader *tlsutil.Loader, banCfg config.BanConfig, hub *connhub.Hub) *IMAPServer {
|
||||
return &IMAPServer{
|
||||
stores: stores,
|
||||
cfg: cfg,
|
||||
banCfg: banCfg,
|
||||
tlsLoader: tlsLoader,
|
||||
hub: hub,
|
||||
}
|
||||
}
|
||||
|
||||
// NotifyNewMessage 向所有 IMAP 监听器推送新邮件通知(go-imap 广播时按
|
||||
// 用户名+邮箱过滤,只送达已选中 INBOX 的客户端,IDLE 挂起时实时收到
|
||||
// FETCH 响应)。由 SMTP/Web 本地投递成功时调用;channel 满时非阻塞丢弃。
|
||||
func (s *IMAPServer) NotifyNewMessage(userEmail string, msg *db.Message) {
|
||||
if s == nil || userEmail == "" || msg == nil {
|
||||
return
|
||||
}
|
||||
update := buildNewMessageUpdate(s.stores, userEmail, msg)
|
||||
if update == nil {
|
||||
return
|
||||
}
|
||||
|
||||
s.beMu.Lock()
|
||||
bes := append([]*imapBackend(nil), s.bes...)
|
||||
s.beMu.Unlock()
|
||||
|
||||
for _, b := range bes {
|
||||
select {
|
||||
case b.updates <- update:
|
||||
default:
|
||||
log.Printf("IMAP: 新邮件推送通道已满,丢弃 %s 的更新 (msg=%d)", userEmail, msg.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// registerBackend 记录新建的 backend(用于新邮件推送)。
|
||||
func (s *IMAPServer) registerBackend(be *imapBackend) {
|
||||
s.beMu.Lock()
|
||||
s.bes = append(s.bes, be)
|
||||
s.beMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *IMAPServer) tlsConfig() (*tls.Config, error) {
|
||||
if s.tlsLoader == nil {
|
||||
return nil, fmt.Errorf("IMAP TLS certificate or key not configured")
|
||||
@@ -44,7 +84,14 @@ func (s *IMAPServer) tlsConfig() (*tls.Config, error) {
|
||||
|
||||
// newServer creates a configured imapserver.Server with the given address.
|
||||
func (s *IMAPServer) newServer(addr string, tlsConfig *tls.Config) *imapserver.Server {
|
||||
be := &imapBackend{stores: s.stores, banCfg: s.banCfg, port: portOf(addr)}
|
||||
be := &imapBackend{
|
||||
stores: s.stores,
|
||||
banCfg: s.banCfg,
|
||||
port: portOf(addr),
|
||||
hub: s.hub,
|
||||
updates: make(chan backend.Update, 256),
|
||||
}
|
||||
s.registerBackend(be)
|
||||
srv := imapserver.New(be)
|
||||
srv.Addr = addr
|
||||
srv.TLSConfig = tlsConfig
|
||||
|
||||
Reference in New Issue
Block a user