nixevol/NixMsg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 完成 broker 复审 B-03 至 B-12

Browse Source 
每连接异步下发与背压、写出后断开、校验当前连接与订阅、生命周期串行、登录条件更新、闲置按在线计、认证超时并发与 Shutdown 0x8B。
This commit is contained in:
Nixevol
2026-09-30 15:24:15 +08:00
parent 42160720bd
commit 8b845843d6
12 changed files with 1099 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/broker/session.go
+26 -12
View File
@@ -210,6 +210,13 @@ func (s *Session) handleHello(ctx context.Context, conn port.ConnInfo, st *connS
}
s.b.SetMaxReceiveBytes(conn.EndpointID, conn.ConnID, maxRecv)
if conn.SessionToken != "" && s.login != nil {
if keep, chkErr := s.login.TokenMatchesDB(ctx, conn.EndpointID, conn.SessionToken); chkErr != nil {
s.log.Error("re-read session token", "endpoint", conn.EndpointID, "err", chkErr)
} else if !keep {
conn.SessionToken = ""
}
}
data := protocol.HelloData{
ServerTimeMs: s.now().UnixMilli(),
ServerVersion: s.limits.ServerVersion,
@@ -279,14 +286,17 @@ func (s *Session) handleLogout(ctx context.Context, conn port.ConnInfo, req *pro
}
}
resp := protocol.Resp{V: protocol.Version, Type: protocol.TypeResp, RID: req.RID, OK: true}
if err := s.publishJSON(ctx, conn, resp, 1); err != nil {
s.log.Error("logout resp", "endpoint", conn.EndpointID, "err", err)
raw, err := protocol.Marshal(resp)
if err != nil {
s.replyErr(ctx, conn, req.RID, protocol.CodeBusy, "marshal logout resp")
return nil
}
if pubErr := s.b.PublishThenDisconnect(ctx, conn.EndpointID, conn.ConnID, raw, 1, port.DisconnectNormal); pubErr != nil {
s.log.Error("logout resp", "endpoint", conn.EndpointID, "err", pubErr)
go func() {
_ = s.b.Disconnect(context.Background(), conn.EndpointID, conn.ConnID, port.DisconnectNormal)
}()
}
go func() {
// 稍等让 QoS1 resp 写入连接，再断开
time.Sleep(50 * time.Millisecond)
_ = s.b.Disconnect(context.Background(), conn.EndpointID, conn.ConnID, port.DisconnectNormal)
}()
return nil
}
@@ -327,11 +337,15 @@ func (s *Session) fatalKick(ctx context.Context, endpointID, reason string) erro
return nil
}
fatal := protocol.Fatal{V: protocol.Version, Type: protocol.TypeFatal, Reason: reason}
_ = s.publishJSON(ctx, info, fatal, 1)
go func() {
time.Sleep(20 * time.Millisecond)
_ = s.b.Disconnect(context.Background(), endpointID, info.ConnID, port.DisconnectFatal)
}()
raw, err := protocol.Marshal(fatal)
if err != nil {
return err
}
if pubErr := s.b.PublishThenDisconnect(ctx, info.EndpointID, info.ConnID, raw, 1, port.DisconnectFatal); pubErr != nil {
go func() {
_ = s.b.Disconnect(context.Background(), endpointID, info.ConnID, port.DisconnectFatal)
}()
}
return nil
}