2 Commits
Author SHA1 Message Date
Nixevol 1f715e4e51 feat: Python/Java SDK 接入清单与打包文档
对真实 nixmsg 跑 DEVELOPMENT 第 9 节接入清单(跳过仅 JS 跨域),补 README/示例,并修 Paho/HiveMQ 真机联调死锁与鉴权分类。
2026-09-30 08:57:06 +08:00
Nixevol 2f750bcb0f test: Go/JS SDK 接入清单集成测与 README 示例 2026-09-30 08:35:09 +08:00
26 changed files with 3833 additions and 154 deletions
+78 -4
View File
@@ -764,11 +764,48 @@
- 原因:`onSession` 作方法名在部分风格指南中易被误认为事件订阅属性。
- 备选:完全同名方法;可在任务 5 文档化时再加别名。
### S1.6 未做范围
### S1.6 任务 1–3 当时未做范围(已被 S1.7 取代)
- 任务 4 真实服务器接入清单(15 条)未做,按总控安排留给后续波次。
- 任务 5 打包试跑/README/示例未做(本期只完成任务 1–3 与许可证文件副本;JS 已能 `npm run build` 出 ESM+CJS,但未发布 npm)。
- Go 裸 TCP(`AllowTCP`)与 JS `allowTcp` 已接线,但无集成级验证(属接入清单)。
- 当时:任务 4 真实服务器接入清单、任务 5 README/示例/打包试跑未做。
- 现况:见 S1.7。
### S1.7 2026-09-30 任务 4–5 接入清单与打包文档
1. **集成测试自建启动器(不 import 根模块 harness)**
- 原条款:TASKS 用 T0.5 harness 起真实服务端。
- 实际做法:`sdk/go` / `sdk/js` 各自在测试里向上查找含 `cmd/nixmsg` 的仓库根,编译二进制,`admin init` + `listen: 127.0.0.1:0` + 临时 `data_dir` 后 `serve`;注册开关与安全码走管理 `PUT /api/admin/registration`。
- 原因:`sdk/go` 是独立 Go 模块,从该目录跑测时 `harness.Binary` 会先命中 `sdk/go/go.mod` 找不到 `./cmd/nixmsg`;JS 也无法直接 import Go 包。
- 备选:给 harness 加 `NIXMSG_ROOT` 环境变量;未改共享目录以免越界。
- 影响:行为与 harness 一致,仅实现重复约百行。
2. **清单第 4 条「ack 丢失后自动再确认」**
- 原条款:模拟 ack 丢失后服务器重推,SDK 自动再确认且不重复回调。
- 实际做法:真实服务用 `ManualAck` + `keep` 消息:收一次不 ack → 管理踢线重连 → 断言回调仍为 1(去重),再手动 `Ack`;「已确认后再推则自动再 ack」仍由 FakeTransport 单测覆盖。
- 原因:自动模式下难以在不改服务器的前提下可靠丢掉已发出的 ack;确认超时默认约 5 分钟,不适合常规集成测。
- 备选:缩短测试用 ack_timeout(需改服务配置/代码,越界)。
- 影响:真实环境覆盖「不重复回调 + 终态确认」;自动再 ack 路径依赖既有单测。
3. **断线期间发送**
- 实际做法:管理 `POST .../kick`(AdministrativeAction,非 `0x8E`)断开连接,SDK 按网络故障重连;在重连窗口调用 `send` 入队,恢复后送达且回调一次。
- 原因:文档写明管理员踢下线只断线、令牌仍可用、SDK 应重连。
- 备选:本地 TCP 代理掐线;未采用以减少测试基础设施。
4. **JS 第 14 条跨源**
- 实际做法:另起本地 HTTP 端口作为「页面」Origin,对注册接口发带 `Origin` 的 OPTIONS/POST,断言 `Access-Control-Allow-Origin: *`;再用 SDK 从「页面」视角连服务器另一端口的 `/mqtt`。未起真实浏览器。
- 原因:Vitest/Node 无完整浏览器;服务端 CORS 与 WS Origin 策略已由身份/连接线保证。
- 备选:Playwright 实浏览器;本期为控制依赖未引入。
5. **任务 5 打包与文档**
- Go:`sdk/go/README.md` + `example/minimal`;不打 `sdk/go/v*` git tag(发布在 Z3)。
- JS:`README.md` + `example/minimal.mjs`;`license` 已为 `SEE LICENSE IN LICENSE`;`npm pack --dry-run` 试跑,**不** `npm publish`。
- 影响:无。
6. **真实 MQTT 收包路径与 request 死锁**
- 原条款:DEVELOPMENT 第 9 节收发/确认;回调串行。
- 实际做法:`resp` 在 MQTT `OnPublishReceived` 路径同步解挂起;`msg`/`receipt`/事件进入 `downCh` 由 `downLoop` 串行处理(可在其中 `request` 发 ack)。
- 原因:若在收包回调里同步 `request` 等 `resp`,而 `resp` 也走同一回调,真实 autopaho 会卡死;假传输因同栈注入 `resp` 掩盖了问题。
- 备选:ack 发后不等 `resp`;未采用,以免丢「ack 结果当 revoked」语义。
- 影响:行为更接近文档;单测仍绿。
## SDK 二 S2
@@ -816,6 +853,43 @@
- 备选方案:依赖库自带重连再改 Clean Start(易漏)。
- 影响:无。
### S2-PY/JAVA 4–5 2026-09-30
1. **接入清单对真实 nixmsg,跳过仅 JS 跨域**
- 原条款:DEVELOPMENT 第 9 节 15 条;任务 4 用 T0.5 启动器起真实服务端。
- 实际做法:Python `tests/harness.py` + `test_checklist.py`、Java `TestHarness` + `ChecklistTest` 自行 `go build`/`admin init`/`serve`(临时目录、`127.0.0.1:0`),管理登录后 `PUT /api/admin/registration` 开注册。覆盖清单 1–13、15;第 14 条(仅 JS 跨域)不做。
- 原因:总控指示跳过 JS 专属跨域;不改服务器业务代码。
- 备选方案:复用 Go `test/harness` 包(SDK 测试不便依赖)。
- 影响:无。
2. **清单第 4 条「ack 丢失后服务器重推」未在真机选择性复现**
- 原条款:模拟 ack 丢失后服务器重推,SDK 自动再确认且不重复回调。
- 实际做法:集成测验证同消息号防重与断线入队重交送达一次;「选择性丢弃 SDK 发出的 ack 帧」在真实 broker 上做不到,去重再 ack 仍由假传输单测覆盖。
- 原因:不改服务器、无中间代理注入丢包。
- 备选方案:toxiproxy 按包过滤(超出本任务、且难按 MQTT 应用帧过滤)。
- 影响:清单 4 真机为部分通过;假传输路径完整。
3. **下行 `resp` 与业务帧分流,避免 auto_ack 自死锁**
- 原条款:自动模式回调后发 ack;回调串行。
- 实际做法:MQTT/`publishes` 回调里对 `resp` 立即完成 pending;`msg` 等进单线程队列再处理(可在队列线程里同步 `ack`/`request`)。Paho 使用 `MQTTv5` 常量与 `transport=websockets`,并等待 SUBACK。
- 原因:若 `resp` 与 `msg` 同队列,auto_ack 等待 `resp` 会永久卡住。
- 备选方案:ack 只发布不等待(弱化协议确认)。
- 影响:与 DEVELOPMENT 行为一致,修复真机联调阻塞。
4. **HiveMQ 鉴权失败与顶号原因码解析**
- 原条款:CONNACK 鉴权失败停重连;`0x8E` 顶号停重连。
- 实际做法:`connect().get()` 抛出的 `Mqtt5ConnAckException` / 文案含 `BAD_USER_*` 时归为 `bad_credentials`(令牌场景 Client 层改为 `session_invalid`);断开原因从 `Mqtt5DisconnectException` 读 `SESSION_TAKEN_OVER`。`connectSync` 等到终态再返回,避免与 attemptConnect 竞态报 `busy/RECONNECTING`。
- 原因:HiveMQ 失败路径多为异常而非成功返回的 CONNACK 对象。
- 备选方案:无。
- 影响:无。
5. **任务 5:README/示例与打包试跑,不发布**
- 原条款:包名与许可证;工具试跑确认能打包;README 与最小示例;真正发布在 Z3。
- 实际做法:更新两端 README;Python `examples/minimal.py`;Java `asia.asio.nixmsg.examples.MinimalExample`;`python -m build` 产出 wheel/sdist;`mvn package -DskipTests` 产出 jar;`javap` major version 52(Java 8)。未上传 PyPI/Maven。
- 原因:本波范围。
- 备选方案:无。
- 影响:无。
## 测试交付 Q
### Q1 / Q4 骨架 2026-09-30
+44
View File
@@ -0,0 +1,44 @@
# NixMsg Go SDK
模块路径:`git.asio.asia/nixevol/NixMsg/sdk/go`,导入包名 `nixmsg`。
## 安装
```bash
# 公共代理访问不到 git.asio.asia 时:
# go env -w GOPRIVATE=git.asio.asia
go get git.asio.asia/nixevol/NixMsg/sdk/go@v0.1.0
```
版本标签带目录前缀:`sdk/go/v0.1.0`(由发布阶段打标,开发期可直接 `replace` 到本地仓库)。
## 最小示例
见 [`example/minimal/main.go`](./example/minimal/main.go):
```go
c := nixmsg.New()
c.OnSession(func(tok string) { /* 应用自行保存 */ })
c.OnMessage(func(msg nixmsg.Message) error {
fmt.Println(msg.From, msg.Body.Data)
return nil
})
ctx := context.Background()
_ = c.Connect(ctx, "ws://127.0.0.1:7443/mqtt", "device-1",
nixmsg.Credential{Password: "secret"}, nixmsg.Options{})
delay := time.Duration(0)
_, _ = c.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "device-2"},
nixmsg.Body{Enc: "utf8", Data: "hello"}, nixmsg.SendOptions{Delay: &delay})
_ = c.Close()
```
静态注册:
```go
res, err := nixmsg.Register(ctx, "ws://127.0.0.1:7443/mqtt", "reg-code",
nixmsg.RegisterOptions{ID: "device-1", LoginPassword: "secret", Name: "门口"})
```
## 许可证
见 `LICENSE`(专有)。
+4
View File
@@ -84,6 +84,9 @@ type Client struct {
helloSentAt time.Time
ctx context.Context
cancel context.CancelFunc
// downCh 串行处理非 resp 下行,避免在 MQTT 收包回调里同步 request 死锁。
downCh chan []byte
}
// New 创建客户端(尚未连接)。
@@ -94,6 +97,7 @@ func New() *Client {
receiptSeen: make(map[string]struct{}),
state: StateOffline,
backoff: newReconnectBackoff(),
downCh: make(chan []byte, 256),
}
}
+14
View File
@@ -44,6 +44,8 @@ func (c *Client) Connect(ctx context.Context, rawURL, endpointID string, credent
c.setStateLocked(StateConnecting, "")
c.mu.Unlock()
go c.downLoop(inner)
tr.SetCredential(pass)
cfg := transportConfig{
URL: rawURL,
@@ -98,11 +100,17 @@ func (c *Client) failAuth(reason AuthReason) {
c.handshook = false
c.setStateLocked(StateAuthFailed, string(reason))
c.failQueuedLocked(apiErr(string(reason), "认证失败,停止重连"))
tr := c.transport
cancel := c.cancel
c.mu.Unlock()
if cancel != nil {
cancel()
}
if tr != nil {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
_ = tr.Stop(ctx)
}
}
func (c *Client) failKicked() {
@@ -111,11 +119,17 @@ func (c *Client) failKicked() {
c.handshook = false
c.setStateLocked(StateKicked, "0x8E")
c.failQueuedLocked(apiErr(CodeKicked, "被顶号,停止重连"))
tr := c.transport
cancel := c.cancel
c.mu.Unlock()
if cancel != nil {
cancel()
}
if tr != nil {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
_ = tr.Stop(ctx)
}
}
func (c *Client) setStateLocked(st ConnectionState, reason string) {
+53
View File
@@ -0,0 +1,53 @@
package main
import (
"context"
"fmt"
"os"
"time"
nixmsg "git.asio.asia/nixevol/NixMsg/sdk/go"
)
func main() {
url := env("NIXMSG_URL", "ws://127.0.0.1:7443/mqtt")
id := env("NIXMSG_ID", "device-1")
pass := env("NIXMSG_PASSWORD", "secret")
peer := env("NIXMSG_PEER", "device-2")
c := nixmsg.New()
c.OnSession(func(tok string) {
fmt.Println("session", tok)
})
c.OnMessage(func(msg nixmsg.Message) error {
fmt.Println("msg", msg.From, msg.Body.Data)
return nil
})
c.OnConnection(func(ev nixmsg.ConnectionEvent) {
fmt.Println("conn", ev.State, ev.Reason)
})
ctx := context.Background()
if err := c.Connect(ctx, url, id, nixmsg.Credential{Password: pass}, nixmsg.Options{}); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
defer c.Close()
delay := time.Duration(0)
res, err := c.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: peer},
nixmsg.Body{Enc: "utf8", Data: "hello from go sdk"}, nixmsg.SendOptions{Delay: &delay})
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
fmt.Println("sent", res.ID, res.State)
time.Sleep(2 * time.Second)
}
func env(k, def string) string {
if v := os.Getenv(k); v != "" {
return v
}
return def
}
+645
View File
@@ -0,0 +1,645 @@
package nixmsg_test
import (
"context"
"errors"
"strings"
"sync/atomic"
"testing"
"time"
nixmsg "git.asio.asia/nixevol/NixMsg/sdk/go"
)
const (
itestPass = "password12"
itestCode = "sdk-go-code1"
)
func TestChecklistAgainstRealServer(t *testing.T) {
srv := startITestServer(t)
admin := newAdminHTTP(t, srv.AdminHTTPBase, srv.AdminPassword)
admin.putRegistration(t, true, itestCode)
ctx := context.Background()
ws := srv.MQTTWS
t.Run("01_handshake_limits", func(t *testing.T) {
mustRegister(t, ctx, ws, "go01a", "A")
c := mustConnect(t, ctx, ws, "go01a", itestPass)
defer c.Close()
lim := c.Limits()
if lim.MaxBodyBytes <= 0 || lim.MaxFrameBytes <= 0 || lim.ServerTimeMs <= 0 {
t.Fatalf("handshake limits incomplete: %+v", lim)
}
if lim.MaxBodyBytes != 262144 {
t.Fatalf("max_body_bytes=%d want 262144", lim.MaxBodyBytes)
}
var tok string
c2 := nixmsg.New()
c2.OnSession(func(token string) { tok = token })
if err := c2.Connect(ctx, ws, "go01a", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatal(err)
}
defer c2.Close()
if tok == "" || !strings.HasPrefix(tok, "nst_") {
t.Fatalf("session token=%q", tok)
}
})
t.Run("02_dm_callback_once", func(t *testing.T) {
mustRegister(t, ctx, ws, "go02a", "A")
mustRegister(t, ctx, ws, "go02b", "B")
a := mustConnect(t, ctx, ws, "go02a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go02b", itestPass)
defer b.Close()
var n atomic.Int32
got := make(chan nixmsg.Message, 4)
b.OnMessage(func(msg nixmsg.Message) error {
n.Add(1)
got <- msg
return nil
})
res, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go02b"}, nixmsg.Body{Enc: "utf8", Data: "hi-once"}, nixmsg.SendOptions{Delay: delay0()})
if err != nil {
t.Fatal(err)
}
msg := waitMsg(t, got, 8*time.Second)
if msg.ID != res.ID || msg.From != "go02a" || msg.Body.Data != "hi-once" {
t.Fatalf("msg=%+v res=%+v", msg, res)
}
time.Sleep(500 * time.Millisecond)
if n.Load() != 1 {
t.Fatalf("callbacks=%d want 1", n.Load())
}
})
t.Run("03_send_while_disconnected", func(t *testing.T) {
mustRegister(t, ctx, ws, "go03a", "A")
mustRegister(t, ctx, ws, "go03b", "B")
a := mustConnect(t, ctx, ws, "go03a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go03b", itestPass)
defer b.Close()
got := make(chan nixmsg.Message, 4)
var n atomic.Int32
b.OnMessage(func(msg nixmsg.Message) error {
n.Add(1)
got <- msg
return nil
})
states := make(chan nixmsg.ConnectionState, 16)
a.OnConnection(func(ev nixmsg.ConnectionEvent) { states <- ev.State })
admin.kick(t, "go03a")
saw := waitConnState(t, states, 15*time.Second, nixmsg.StateReconnecting, nixmsg.StateOnline)
resCh := make(chan nixmsg.SendResult, 1)
errCh := make(chan error, 1)
go func() {
res, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go03b"}, nixmsg.Body{Enc: "utf8", Data: "queued"}, nixmsg.SendOptions{Delay: delay0(), Keep: true, ID: "go03-msg-1"})
if err != nil {
errCh <- err
return
}
resCh <- res
}()
if saw != nixmsg.StateOnline {
_ = waitConnState(t, states, 30*time.Second, nixmsg.StateOnline)
}
select {
case err := <-errCh:
t.Fatal(err)
case res := <-resCh:
if res.ID != "go03-msg-1" {
t.Fatalf("id=%s", res.ID)
}
case <-time.After(30 * time.Second):
t.Fatal("send timeout")
}
msg := waitMsg(t, got, 15*time.Second)
if msg.ID != "go03-msg-1" {
t.Fatalf("msg=%+v", msg)
}
time.Sleep(800 * time.Millisecond)
if n.Load() != 1 {
t.Fatalf("callbacks=%d want 1", n.Load())
}
})
t.Run("04_same_id_retry_and_dedup", func(t *testing.T) {
mustRegister(t, ctx, ws, "go04a", "A")
mustRegister(t, ctx, ws, "go04b", "B")
a := mustConnect(t, ctx, ws, "go04a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go04b", itestPass)
defer b.Close()
// 同号:断线期间入队发送,重连后以同一消息号送达
var n atomic.Int32
got := make(chan nixmsg.Message, 4)
b.OnMessage(func(msg nixmsg.Message) error {
n.Add(1)
got <- msg
return nil
})
states := make(chan nixmsg.ConnectionState, 16)
a.OnConnection(func(ev nixmsg.ConnectionEvent) { states <- ev.State })
admin.kick(t, "go04a")
saw := waitConnState(t, states, 15*time.Second, nixmsg.StateReconnecting, nixmsg.StateOnline)
go func() {
_, _ = a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go04b"}, nixmsg.Body{Enc: "utf8", Data: "same-id"}, nixmsg.SendOptions{Delay: delay0(), Keep: true, ID: "go04-fixed-id"})
}()
if saw != nixmsg.StateOnline {
_ = waitConnState(t, states, 30*time.Second, nixmsg.StateOnline)
}
msg := waitMsg(t, got, 15*time.Second)
if msg.ID != "go04-fixed-id" {
t.Fatalf("want go04-fixed-id got %s", msg.ID)
}
time.Sleep(500 * time.Millisecond)
if n.Load() != 1 {
t.Fatalf("same-id callbacks=%d", n.Load())
}
// 模拟未确认:手动确认模式收一次、踢线重连后服务器重推,SDK 不重复回调;再手动 ack
b2 := nixmsg.New()
var n2 atomic.Int32
got2 := make(chan nixmsg.Message, 4)
b2.OnMessage(func(msg nixmsg.Message) error {
n2.Add(1)
got2 <- msg
return nil
})
if err := b2.Connect(ctx, ws, "go04b", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 20 * time.Second, ManualAck: true}); err != nil {
t.Fatal(err)
}
defer b2.Close()
_, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go04b"}, nixmsg.Body{Enc: "utf8", Data: "repush"}, nixmsg.SendOptions{Delay: delay0(), Keep: true, TTL: int64Ptr(3600), ID: "go04-repush"})
if err != nil {
t.Fatal(err)
}
m1 := waitMsg(t, got2, 10*time.Second)
if m1.ID != "go04-repush" {
t.Fatalf("m1=%+v", m1)
}
st2 := make(chan nixmsg.ConnectionState, 16)
b2.OnConnection(func(ev nixmsg.ConnectionEvent) { st2 <- ev.State })
admin.kick(t, "go04b")
_ = waitConnState(t, st2, 30*time.Second, nixmsg.StateOnline)
time.Sleep(2 * time.Second)
if n2.Load() != 1 {
t.Fatalf("repush callbacks=%d want 1 (no duplicate)", n2.Load())
}
if err := b2.Ack(m1); err != nil {
t.Fatal(err)
}
})
t.Run("05_recall_within_delay", func(t *testing.T) {
mustRegister(t, ctx, ws, "go05a", "A")
mustRegister(t, ctx, ws, "go05b", "B")
a := mustConnect(t, ctx, ws, "go05a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go05b", itestPass)
defer b.Close()
var msgN, revN atomic.Int32
b.OnMessage(func(msg nixmsg.Message) error {
msgN.Add(1)
return nil
})
b.OnRevoked(func(e nixmsg.RevokedEvent) { revN.Add(1) })
d := 10 * time.Second
res, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go05b"}, nixmsg.Body{Enc: "utf8", Data: "will-recall"}, nixmsg.SendOptions{Delay: &d, ID: "go05-rec"})
if err != nil {
t.Fatal(err)
}
if res.State != "scheduled" && res.State != "" {
// state may be scheduled
}
if _, err := a.Recall(ctx, "go05-rec"); err != nil {
t.Fatal(err)
}
time.Sleep(1500 * time.Millisecond)
if msgN.Load() != 0 {
t.Fatalf("bob got msg callbacks=%d", msgN.Load())
}
if revN.Load() != 0 {
t.Fatalf("bob got revoked=%d (undelivered should be silent)", revN.Load())
}
})
t.Run("06_scheduled_about_2s", func(t *testing.T) {
mustRegister(t, ctx, ws, "go06a", "A")
mustRegister(t, ctx, ws, "go06b", "B")
a := mustConnect(t, ctx, ws, "go06a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go06b", itestPass)
defer b.Close()
got := make(chan time.Time, 1)
b.OnMessage(func(msg nixmsg.Message) error {
got <- time.Now()
return nil
})
d := 2 * time.Second
start := time.Now()
if _, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go06b"}, nixmsg.Body{Enc: "utf8", Data: "later"}, nixmsg.SendOptions{Delay: &d}); err != nil {
t.Fatal(err)
}
select {
case at := <-got:
elapsed := at.Sub(start)
if elapsed < 1500*time.Millisecond || elapsed > 6*time.Second {
t.Fatalf("elapsed=%v want ~2s", elapsed)
}
case <-time.After(10 * time.Second):
t.Fatal("timeout waiting scheduled msg")
}
})
t.Run("07_offline_keep", func(t *testing.T) {
mustRegister(t, ctx, ws, "go07a", "A")
mustRegister(t, ctx, ws, "go07b", "B")
mustRegister(t, ctx, ws, "go07c", "C")
a := mustConnect(t, ctx, ws, "go07a", itestPass)
defer a.Close()
// 晚 1 秒上线能收到
ttl := int64(60)
if _, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go07b"}, nixmsg.Body{Enc: "utf8", Data: "keep-ok"}, nixmsg.SendOptions{Delay: delay0(), Keep: true, TTL: &ttl, ID: "go07-keep-ok"}); err != nil {
t.Fatal(err)
}
time.Sleep(1 * time.Second)
got := make(chan nixmsg.Message, 2)
b := nixmsg.New()
b.OnMessage(func(msg nixmsg.Message) error {
got <- msg
return nil
})
if err := b.Connect(ctx, ws, "go07b", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatal(err)
}
defer b.Close()
msg := waitMsg(t, got, 10*time.Second)
if msg.ID != "go07-keep-ok" {
t.Fatalf("keep-ok msg=%+v", msg)
}
// 保留 1 秒,3 秒后上线收不到;发送方收到过期回执
receiptCh := make(chan nixmsg.Receipt, 4)
a.OnReceipt(func(r nixmsg.Receipt) { receiptCh <- r })
ttl1 := int64(1)
if _, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go07c"}, nixmsg.Body{Enc: "utf8", Data: "expire"}, nixmsg.SendOptions{Delay: delay0(), Keep: true, TTL: &ttl1, ID: "go07-exp", Receipt: boolPtr(true)}); err != nil {
t.Fatal(err)
}
time.Sleep(3 * time.Second)
c := mustConnect(t, ctx, ws, "go07c", itestPass)
defer c.Close()
var cN atomic.Int32
c.OnMessage(func(msg nixmsg.Message) error {
cN.Add(1)
return nil
})
time.Sleep(2 * time.Second)
if cN.Load() != 0 {
t.Fatalf("expired offline should not deliver, got %d", cN.Load())
}
deadline := time.Now().Add(15 * time.Second)
for time.Now().Before(deadline) {
select {
case r := <-receiptCh:
if r.ID == "go07-exp" && (r.State == "expired" || r.Reason == "expired" || strings.Contains(r.State, "expir") || strings.Contains(r.Reason, "expir")) {
return
}
case <-time.After(200 * time.Millisecond):
}
}
// 回执可能稍慢;再扫一下
select {
case r := <-receiptCh:
if r.ID != "go07-exp" {
t.Fatalf("unexpected receipt %+v", r)
}
default:
t.Fatal("未收到过期回执")
}
})
t.Run("08_group_no_echo_to_sender", func(t *testing.T) {
mustRegister(t, ctx, ws, "go08a", "A")
mustRegister(t, ctx, ws, "go08b", "B")
mustRegister(t, ctx, ws, "go08c", "C")
a := mustConnect(t, ctx, ws, "go08a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go08b", itestPass)
defer b.Close()
c := mustConnect(t, ctx, ws, "go08c", itestPass)
defer c.Close()
if _, err := a.CreateGroup(ctx, "g_go08", "G8", []nixmsg.GroupMemberIn{{ID: "go08b"}, {ID: "go08c"}}); err != nil {
t.Fatal(err)
}
time.Sleep(300 * time.Millisecond)
bGot := make(chan nixmsg.Message, 2)
cGot := make(chan nixmsg.Message, 2)
var aN atomic.Int32
a.OnMessage(func(msg nixmsg.Message) error {
aN.Add(1)
return nil
})
b.OnMessage(func(msg nixmsg.Message) error {
bGot <- msg
return nil
})
c.OnMessage(func(msg nixmsg.Message) error {
cGot <- msg
return nil
})
if _, err := a.Send(ctx, nixmsg.Target{Kind: "group", ID: "g_go08"}, nixmsg.Body{Enc: "utf8", Data: "ghi"}, nixmsg.SendOptions{Delay: delay0(), ID: "go08-g1"}); err != nil {
t.Fatal(err)
}
mb := waitMsg(t, bGot, 10*time.Second)
mc := waitMsg(t, cGot, 10*time.Second)
if mb.Body.Data != "ghi" || mc.Body.Data != "ghi" {
t.Fatalf("b=%+v c=%+v", mb, mc)
}
time.Sleep(500 * time.Millisecond)
if aN.Load() != 0 {
t.Fatalf("sender got %d group msgs", aN.Load())
}
})
t.Run("09_talk_password", func(t *testing.T) {
mustRegister(t, ctx, ws, "go09a", "A")
mustRegister(t, ctx, ws, "go09b", "B")
a := mustConnect(t, ctx, ws, "go09a", itestPass)
defer a.Close()
b := mustConnect(t, ctx, ws, "go09b", itestPass)
defer b.Close()
if err := b.SetTalkPassword(ctx, "talk9"); err != nil {
t.Fatal(err)
}
_, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go09b"}, nixmsg.Body{Enc: "utf8", Data: "x"}, nixmsg.SendOptions{Delay: delay0()})
if err == nil {
t.Fatal("want talk_password_required")
}
var ae *nixmsg.APIError
if !errors.As(err, &ae) || ae.Code != "talk_password_required" {
t.Fatalf("err=%v", err)
}
if err := a.Unlock(ctx, "go09b", "talk9"); err != nil {
t.Fatal(err)
}
if _, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go09b"}, nixmsg.Body{Enc: "utf8", Data: "ok"}, nixmsg.SendOptions{Delay: delay0()}); err != nil {
t.Fatal(err)
}
if err := b.SetTalkPassword(ctx, "talk9b"); err != nil {
t.Fatal(err)
}
_, err = a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go09b"}, nixmsg.Body{Enc: "utf8", Data: "stale"}, nixmsg.SendOptions{Delay: delay0()})
if err == nil {
t.Fatal("want auth invalid after password change")
}
if !errors.As(err, &ae) || (ae.Code != "talk_password_required" && ae.Code != "talk_password_invalid") {
t.Fatalf("after change err=%v", err)
}
// 对方先发则可回复:b 给 a 发(a 无对话密码)后,a 可回 b(即使用旧授权已失效,回复授权由 b→a 的发送产生)
got := make(chan nixmsg.Message, 2)
a.OnMessage(func(msg nixmsg.Message) error {
got <- msg
return nil
})
if _, err := b.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go09a"}, nixmsg.Body{Enc: "utf8", Data: "first"}, nixmsg.SendOptions{Delay: delay0()}); err != nil {
t.Fatal(err)
}
_ = waitMsg(t, got, 8*time.Second)
if _, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go09b"}, nixmsg.Body{Enc: "utf8", Data: "reply"}, nixmsg.SendOptions{Delay: delay0()}); err != nil {
t.Fatalf("reply after peer first send: %v", err)
}
})
t.Run("10_second_login_kicks_first", func(t *testing.T) {
mustRegister(t, ctx, ws, "go10a", "A")
c1 := mustConnect(t, ctx, ws, "go10a", itestPass)
defer c1.Close()
var kicked atomic.Bool
c1.OnConnection(func(ev nixmsg.ConnectionEvent) {
if ev.State == nixmsg.StateKicked {
kicked.Store(true)
}
})
c2 := mustConnect(t, ctx, ws, "go10a", itestPass)
defer c2.Close()
deadline := time.Now().Add(15 * time.Second)
for time.Now().Before(deadline) {
if kicked.Load() {
break
}
time.Sleep(50 * time.Millisecond)
}
if !kicked.Load() {
t.Fatal("first client not kicked")
}
time.Sleep(2 * time.Second)
// 被顶号后不应恢复 online
if c1.Limits().ServerTimeMs != 0 {
// Limits 仍可能保留旧值;用连接状态回调或再次 Send 探测
}
_, err := c1.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go10a"}, nixmsg.Body{Enc: "utf8", Data: "x"}, nixmsg.SendOptions{Delay: delay0()})
if err == nil {
t.Fatal("kicked client should not send successfully after stop-reconnect")
}
})
t.Run("11_body_too_large_local", func(t *testing.T) {
mustRegister(t, ctx, ws, "go11a", "A")
mustRegister(t, ctx, ws, "go11b", "B")
a := mustConnect(t, ctx, ws, "go11a", itestPass)
defer a.Close()
big := strings.Repeat("x", 262144+1)
_, err := a.Send(ctx, nixmsg.Target{Kind: "endpoint", ID: "go11b"}, nixmsg.Body{Enc: "utf8", Data: big}, nixmsg.SendOptions{Delay: delay0()})
if err == nil {
t.Fatal("want body_too_large")
}
var ae *nixmsg.APIError
if !errors.As(err, &ae) || ae.Code != nixmsg.CodeBodyTooLarge {
t.Fatalf("err=%v", err)
}
})
t.Run("12_registration_switch_and_code", func(t *testing.T) {
admin.putRegistration(t, false, itestCode)
_, err := nixmsg.Register(ctx, ws, itestCode, nixmsg.RegisterOptions{ID: "go12x", LoginPassword: itestPass, Name: "X"})
if err == nil {
t.Fatal("want registration closed")
}
admin.putRegistration(t, true, itestCode)
_, err = nixmsg.Register(ctx, ws, "wrong-code-xx", nixmsg.RegisterOptions{ID: "go12y", LoginPassword: itestPass, Name: "Y"})
if err == nil {
t.Fatal("want bad code")
}
codeNew := "sdk-go-code2"
admin.putRegistration(t, true, itestCode)
res, err := nixmsg.Register(ctx, ws, itestCode, nixmsg.RegisterOptions{ID: "go12ok", LoginPassword: itestPass, Name: "OK"})
if err != nil {
t.Fatal(err)
}
if res.ID != "go12ok" {
t.Fatalf("id=%s", res.ID)
}
c := mustConnect(t, ctx, ws, "go12ok", itestPass)
c.Close()
admin.putRegistration(t, true, codeNew)
_, err = nixmsg.Register(ctx, ws, itestCode, nixmsg.RegisterOptions{ID: "go12old", LoginPassword: itestPass, Name: "Old"})
if err == nil {
t.Fatal("old code should fail")
}
// 已注册端仍可用旧密码登录
c2 := mustConnect(t, ctx, ws, "go12ok", itestPass)
c2.Close()
admin.putRegistration(t, true, itestCode) // 恢复供后续子测试
})
t.Run("13_change_login_password", func(t *testing.T) {
mustRegister(t, ctx, ws, "go13a", "A")
c := mustConnect(t, ctx, ws, "go13a", itestPass)
if err := c.ChangeLoginPassword(ctx, itestPass, "password99"); err != nil {
t.Fatal(err)
}
_ = c.Close()
cNew := mustConnect(t, ctx, ws, "go13a", "password99")
cNew.Close()
cBad := nixmsg.New()
var authFail atomic.Bool
cBad.OnConnection(func(ev nixmsg.ConnectionEvent) {
if ev.State == nixmsg.StateAuthFailed {
authFail.Store(true)
}
})
err := cBad.Connect(ctx, ws, "go13a", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 10 * time.Second})
if err == nil {
_ = cBad.Close()
t.Fatal("old password should fail")
}
deadline := time.Now().Add(5 * time.Second)
for time.Now().Before(deadline) && !authFail.Load() {
time.Sleep(50 * time.Millisecond)
}
time.Sleep(1500 * time.Millisecond)
// 不应恢复为 online
if !authFail.Load() {
t.Log("auth_failed event not observed; connect error present which is enough")
}
_ = cBad.Close()
})
t.Run("15_session_token", func(t *testing.T) {
mustRegister(t, ctx, ws, "go15a", "A")
var tok1 string
c1 := nixmsg.New()
c1.OnSession(func(token string) { tok1 = token })
if err := c1.Connect(ctx, ws, "go15a", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatal(err)
}
if tok1 == "" {
t.Fatal("no token")
}
_ = c1.Close()
cTok := nixmsg.New()
if err := cTok.Connect(ctx, ws, "go15a", nixmsg.Credential{SessionToken: tok1}, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatalf("token reconnect: %v", err)
}
_ = cTok.Close()
// 另一处密码登录使旧令牌失效
cPass := mustConnect(t, ctx, ws, "go15a", itestPass)
defer cPass.Close()
cOld := nixmsg.New()
var inv atomic.Bool
cOld.OnConnection(func(ev nixmsg.ConnectionEvent) {
if ev.State == nixmsg.StateAuthFailed && (ev.Reason == string(nixmsg.AuthSessionInvalid) || strings.Contains(ev.Reason, "session")) {
inv.Store(true)
}
if ev.State == nixmsg.StateAuthFailed {
inv.Store(true)
}
})
err := cOld.Connect(ctx, ws, "go15a", nixmsg.Credential{SessionToken: tok1}, nixmsg.Options{ConnectTimeout: 10 * time.Second})
if err == nil {
_ = cOld.Close()
t.Fatal("old token should fail after other password login")
}
_ = cOld.Close()
// logout 后令牌失效
var tok2 string
c3 := nixmsg.New()
c3.OnSession(func(token string) { tok2 = token })
if err := c3.Connect(ctx, ws, "go15a", nixmsg.Credential{Password: itestPass}, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatal(err)
}
if err := c3.Logout(ctx); err != nil {
t.Fatal(err)
}
_ = c3.Close()
c4 := nixmsg.New()
err = c4.Connect(ctx, ws, "go15a", nixmsg.Credential{SessionToken: tok2}, nixmsg.Options{ConnectTimeout: 10 * time.Second})
if err == nil {
_ = c4.Close()
t.Fatal("token after logout should fail")
}
_ = c4.Close()
})
}
func mustRegister(t *testing.T, ctx context.Context, ws, id, name string) {
t.Helper()
_, err := nixmsg.Register(ctx, ws, itestCode, nixmsg.RegisterOptions{ID: id, LoginPassword: itestPass, Name: name})
if err != nil {
// 可能已存在(重跑);尝试直接登录
t.Logf("register %s: %v (continue if already exists)", id, err)
}
}
func mustConnect(t *testing.T, ctx context.Context, ws, id, password string) *nixmsg.Client {
t.Helper()
c := nixmsg.New()
cred := nixmsg.Credential{Password: password}
if err := c.Connect(ctx, ws, id, cred, nixmsg.Options{ConnectTimeout: 20 * time.Second}); err != nil {
t.Fatalf("connect %s: %v", id, err)
}
return c
}
func waitMsg(t *testing.T, ch <-chan nixmsg.Message, d time.Duration) nixmsg.Message {
t.Helper()
select {
case m := <-ch:
return m
case <-time.After(d):
t.Fatal("timeout waiting message")
return nixmsg.Message{}
}
}
func waitConnState(t *testing.T, states <-chan nixmsg.ConnectionState, d time.Duration, want ...nixmsg.ConnectionState) nixmsg.ConnectionState {
t.Helper()
wants := map[nixmsg.ConnectionState]bool{}
for _, w := range want {
wants[w] = true
}
deadline := time.Now().Add(d)
for time.Now().Before(deadline) {
select {
case st := <-states:
if wants[st] {
return st
}
case <-time.After(50 * time.Millisecond):
}
}
t.Fatalf("timeout waiting state among %v", want)
return ""
}
func int64Ptr(v int64) *int64 { return &v }
func boolPtr(v bool) *bool { return &v }
+297
View File
@@ -0,0 +1,297 @@
package nixmsg_test
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/cookiejar"
"os"
"os/exec"
"path/filepath"
"runtime"
"strings"
"sync"
"testing"
"time"
)
// 集成测试启动器:在仓库根目录编译 nixmsg,临时目录 + 127.0.0.1:0 + admin init。
// 不依赖根模块 harness 包(sdk/go 是独立模块,从本目录起测时 go.mod 会挡住 Binary 找根)。
type itestServer struct {
BinPath string
ConfigPath string
DataDir string
Addr string
HTTPBase string
AdminHTTPBase string
AdminPassword string
MQTTWS string
cmd *exec.Cmd
}
var (
itestBinOnce sync.Once
itestBinPath string
itestBinErr error
)
func repoRoot(t *testing.T) string {
t.Helper()
dir, err := os.Getwd()
if err != nil {
t.Fatal(err)
}
for {
nix := filepath.Join(dir, "cmd", "nixmsg")
mod := filepath.Join(dir, "go.mod")
if st, e := os.Stat(nix); e == nil && st.IsDir() {
if b, e2 := os.ReadFile(mod); e2 == nil && bytes.Contains(b, []byte("module git.asio.asia/nixevol/NixMsg\n")) {
return dir
}
}
parent := filepath.Dir(dir)
if parent == dir {
t.Fatal("找不到仓库根(含 cmd/nixmsg)")
}
dir = parent
}
}
func itestBinary(t *testing.T) string {
t.Helper()
itestBinOnce.Do(func() {
root := ""
dir, err := os.Getwd()
if err != nil {
itestBinErr = err
return
}
for {
nix := filepath.Join(dir, "cmd", "nixmsg")
mod := filepath.Join(dir, "go.mod")
if st, e := os.Stat(nix); e == nil && st.IsDir() {
if b, e2 := os.ReadFile(mod); e2 == nil && bytes.Contains(b, []byte("module git.asio.asia/nixevol/NixMsg\n")) {
root = dir
break
}
}
parent := filepath.Dir(dir)
if parent == dir {
itestBinErr = fmt.Errorf("repo root not found")
return
}
dir = parent
}
tmp, err := os.MkdirTemp("", "nixmsg-sdk-go-bin-*")
if err != nil {
itestBinErr = err
return
}
name := "nixmsg"
if runtime.GOOS == "windows" {
name += ".exe"
}
out := filepath.Join(tmp, name)
cmd := exec.Command("go", "build", "-o", out, "./cmd/nixmsg")
cmd.Dir = root
cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
if b, e := cmd.CombinedOutput(); e != nil {
itestBinErr = fmt.Errorf("build nixmsg: %w\n%s", e, b)
return
}
itestBinPath = out
})
if itestBinErr != nil {
t.Fatal(itestBinErr)
}
return itestBinPath
}
func startITestServer(t *testing.T) *itestServer {
t.Helper()
bin := itestBinary(t)
dataDir, err := os.MkdirTemp("", "nixmsg-sdk-go-itest-*")
if err != nil {
t.Fatal(err)
}
cfgPath := filepath.Join(dataDir, "config.yaml")
cfg := fmt.Sprintf("listen: %q\ndata_dir: %q\n", "127.0.0.1:0", filepath.ToSlash(dataDir))
if err := os.WriteFile(cfgPath, []byte(cfg), 0o644); err != nil {
_ = os.RemoveAll(dataDir)
t.Fatal(err)
}
initCmd := exec.Command(bin, "admin", "init")
initCmd.Env = append(os.Environ(), "NIXMSG_CONFIG="+cfgPath)
initOut, initErr := initCmd.CombinedOutput()
if initErr != nil {
_ = os.RemoveAll(dataDir)
t.Fatalf("admin init: %v\n%s", initErr, initOut)
}
pass := parseAdminPassword(string(initOut))
if pass == "" {
_ = os.RemoveAll(dataDir)
t.Fatalf("admin init 未打印密码:\n%s", initOut)
}
cmd := exec.Command(bin, "serve")
cmd.Env = append(os.Environ(), "NIXMSG_CONFIG="+cfgPath)
cmd.Stdout = os.Stderr
cmd.Stderr = os.Stderr
if err := cmd.Start(); err != nil {
_ = os.RemoveAll(dataDir)
t.Fatal(err)
}
s := &itestServer{
BinPath: bin,
ConfigPath: cfgPath,
DataDir: dataDir,
AdminPassword: pass,
cmd: cmd,
}
addr, err := waitAddrFile(filepath.Join(dataDir, "listen.addr"), 15*time.Second)
if err != nil {
_ = s.Stop()
t.Fatalf("wait listen.addr: %v", err)
}
s.Addr = addr
s.HTTPBase = "http://" + addr
s.AdminHTTPBase = s.HTTPBase
s.MQTTWS = "ws://" + addr + "/mqtt"
t.Cleanup(func() { _ = s.Stop() })
return s
}
func (s *itestServer) Stop() error {
if s == nil || s.cmd == nil {
return nil
}
if s.cmd.Process != nil {
_ = s.cmd.Process.Kill()
_, _ = s.cmd.Process.Wait()
}
s.cmd = nil
if s.DataDir != "" {
return os.RemoveAll(s.DataDir)
}
return nil
}
func parseAdminPassword(out string) string {
for _, line := range strings.Split(out, "\n") {
line = strings.TrimSpace(line)
lower := strings.ToLower(line)
if strings.HasPrefix(lower, "admin password:") {
return strings.TrimSpace(line[len("admin password:"):])
}
if strings.HasPrefix(lower, "password:") {
return strings.TrimSpace(line[len("password:"):])
}
}
return ""
}
func waitAddrFile(path string, timeout time.Duration) (string, error) {
deadline := time.Now().Add(timeout)
var last error
for time.Now().Before(deadline) {
b, err := os.ReadFile(path)
if err == nil {
addr := strings.TrimSpace(string(b))
if addr != "" {
return addr, nil
}
last = fmt.Errorf("empty addr")
} else {
last = err
}
time.Sleep(20 * time.Millisecond)
}
return "", last
}
type adminHTTP struct {
base string
hc *http.Client
}
func newAdminHTTP(t *testing.T, base, password string) *adminHTTP {
t.Helper()
jar, err := cookiejar.New(nil)
if err != nil {
t.Fatal(err)
}
a := &adminHTTP{
base: strings.TrimRight(base, "/"),
hc: &http.Client{Timeout: 30 * time.Second, Jar: jar},
}
body, _ := json.Marshal(map[string]string{"username": "admin", "password": password})
res, err := a.do(http.MethodPost, "/api/admin/login", body, "application/json")
if err != nil {
t.Fatal(err)
}
defer res.Body.Close()
if res.StatusCode != http.StatusOK {
raw, _ := io.ReadAll(res.Body)
t.Fatalf("admin login %d %s", res.StatusCode, raw)
}
return a
}
func (a *adminHTTP) do(method, path string, body []byte, ct string) (*http.Response, error) {
var rdr io.Reader
if body != nil {
rdr = bytes.NewReader(body)
}
req, err := http.NewRequest(method, a.base+path, rdr)
if err != nil {
return nil, err
}
if ct != "" {
req.Header.Set("Content-Type", ct)
}
switch strings.ToUpper(method) {
case http.MethodPost, http.MethodPut, http.MethodPatch, http.MethodDelete:
req.Header.Set("X-Nixmsg-Request", "1")
}
return a.hc.Do(req)
}
func (a *adminHTTP) putRegistration(t *testing.T, enabled bool, code string) {
t.Helper()
payload := map[string]any{"enabled": enabled}
if code != "" {
payload["code"] = code
}
raw, _ := json.Marshal(payload)
res, err := a.do(http.MethodPut, "/api/admin/registration", raw, "application/json")
if err != nil {
t.Fatal(err)
}
defer res.Body.Close()
b, _ := io.ReadAll(res.Body)
if res.StatusCode != http.StatusOK {
t.Fatalf("put registration %d %s", res.StatusCode, b)
}
}
func (a *adminHTTP) kick(t *testing.T, endpointID string) {
t.Helper()
res, err := a.do(http.MethodPost, "/api/admin/endpoints/"+endpointID+"/kick", []byte("{}"), "application/json")
if err != nil {
t.Fatal(err)
}
defer res.Body.Close()
if res.StatusCode != http.StatusOK {
b, _ := io.ReadAll(res.Body)
t.Fatalf("kick %s: %d %s", endpointID, res.StatusCode, b)
}
}
func delay0() *time.Duration {
d := time.Duration(0)
return &d
}
+30 -2
View File
@@ -15,8 +15,8 @@ func (c *Client) handleDown(payload []byte) {
if err := unmarshalJSON(payload, &head); err != nil {
return
}
switch head.Type {
case "resp":
// resp 必须在收包路径同步处理,否则 request/ack 在 downLoop 里等待时会死锁。
if head.Type == "resp" {
var rf respFrame
if err := unmarshalJSON(payload, &rf); err != nil {
return
@@ -33,6 +33,34 @@ func (c *Client) handleDown(payload []byte) {
default:
}
}
return
}
cp := append([]byte(nil), payload...)
select {
case c.downCh <- cp:
case <-c.ctx.Done():
}
}
func (c *Client) downLoop(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case payload := <-c.downCh:
c.handleDownApp(payload)
}
}
}
func (c *Client) handleDownApp(payload []byte) {
var head struct {
Type string `json:"type"`
}
if err := unmarshalJSON(payload, &head); err != nil {
return
}
switch head.Type {
case "msg":
c.handleMsg(payload)
case "receipt":
+37 -4
View File
@@ -2,7 +2,7 @@
坐标:`asia.asio.nixmsg:nixmsg-sdk`
包名:`asia.asio.nixmsg`
字节码目标:Java 8
字节码目标:Java 8(`maven.compiler.release=8`)
接口:`CompletableFuture`
## Android
@@ -17,7 +17,15 @@ Maven / Gradle 仓库:
https://git.asio.asia/api/packages/nixevol/maven
```
HiveMQ MQTT Client(含 WebSocket:`webSocketConfig` + `netty-codec-http`)。
```xml
<dependency>
<groupId>asia.asio.nixmsg</groupId>
<artifactId>nixmsg-sdk</artifactId>
<version>0.1.0</version>
</dependency>
```
HiveMQ MQTT Client(WebSocket:`webSocketConfig` + `netty-codec-http`)。
## 最小示例
@@ -26,9 +34,34 @@ Client c = new Client();
c.onSession(token -> { /* 应用保存 */ });
c.onMessage(msg -> System.out.println(msg.id + " " + msg.body.data));
c.connect("ws://127.0.0.1:7443/mqtt", "device-1", "secret", null)
.thenCompose(v -> c.send(new Types.Target("endpoint", "device-2"), new Types.Body("hello"), new Types.SendOptions()))
.thenCompose(v -> {
Types.SendOptions opt = new Types.SendOptions();
opt.delayMs = 0L;
return c.send(new Types.Target("endpoint", "device-2"), new Types.Body("hello"), opt);
})
.join();
c.close();
```
许可证见 `LICENSE`(专有)。
命令行示例类:`asia.asio.nixmsg.examples.MinimalExample`。
## 打包(不发布)
```bash
mvn package -DskipTests
# 产物 target/nixmsg-sdk-0.1.0.jar;勿部署到 Maven 仓库;正式发布由总控在阶段 3 执行
```
确认字节码为 8:`javap -v target/classes/asia/asio/nixmsg/Client.class | findstr major`(应为 52)。
## 测试
```bash
mvn test
```
含假传输单元测试与 DEVELOPMENT 第 9 节接入清单(会编译并启动真实 `nixmsg`)。可用环境变量 `NIXMSG_BIN` 指定已编译二进制。跳过仅 JS 的跨域项。
## 许可证
见 `LICENSE`(Proprietary)。
@@ -27,7 +27,9 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -82,6 +84,8 @@ public final class Client {
private final Object connWait = new Object();
private Thread worker;
private final Object wake = new Object();
private final BlockingQueue<byte[]> downQueue = new LinkedBlockingQueue<byte[]>();
private final Thread downWorker;
private Consumer<String> sessionHandler;
private Consumer<IncomingMessage> messageHandler;
@@ -106,6 +110,9 @@ public final class Client {
this.clientName = clientName;
this.connectTimeoutMs = connectTimeoutMs;
this.transport.setHandlers(this::onTransportConnected, this::onTransportDisconnected, this::onDown);
this.downWorker = new Thread(this::downLoop, "nixmsg-down");
this.downWorker.setDaemon(true);
this.downWorker.start();
}
public void onSession(Consumer<String> handler) { this.sessionHandler = handler; }
@@ -121,6 +128,11 @@ public final class Client {
public String getSessionToken() { return sessionToken; }
public long getClockSkewMs() { return clockSkewMs; }
/** 同包测试用:断开底层传输以触发重连与发送队列重交。 */
void dropTransportForTest() {
transport.disconnect();
}
public CompletableFuture<Void> connect(String url, String endpointId, String password, String sessionToken) {
return connect(url, endpointId, password, sessionToken, false);
}
@@ -157,9 +169,17 @@ public final class Client {
}
long deadline = System.currentTimeMillis() + connectTimeoutMs + 5000;
synchronized (connWait) {
while (!connReady.get() && System.currentTimeMillis() < deadline) {
while (System.currentTimeMillis() < deadline) {
ConnectionState s = state;
if (s == ConnectionState.ONLINE
|| s == ConnectionState.AUTH_FAILED
|| s == ConnectionState.KICKED
|| s == ConnectionState.OFFLINE
|| handshakeError != null) {
break;
}
try {
connWait.wait(200);
connWait.wait(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
@@ -200,6 +220,7 @@ public final class Client {
transport.disconnect();
} catch (Exception ignored) {
}
downQueue.offer(new byte[0]); // 空载荷哨兵:downLoop 见 closed 退出
wakeUp();
}
@@ -716,57 +737,108 @@ public final class Client {
}
private void onDown(byte[] payload) {
// resp 立即完成 pending,避免 down 工作线程在 autoAck 等待时自死锁。
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
downQueue.offer(payload);
return;
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
return;
}
downQueue.offer(payload);
}
private void downLoop() {
while (true) {
try {
byte[] payload = downQueue.take();
if (payload.length == 0 && closed) {
return;
}
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
continue;
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
} else {
dispatchDownBody(frame);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
} catch (Exception e) {
LOG.log(Level.WARNING, "处理下行帧失败", e);
}
}
}
private void dispatchResp(Map<String, Object> frame) {
String rid = str(frame.get("rid"), "");
Pending p;
synchronized (lock) {
p = pending.remove(rid);
}
if (p == null) {
return;
}
if (p.isSend) {
synchronized (lock) {
inflightSends = Math.max(0, inflightSends - 1);
}
if (!Boolean.TRUE.equals(frame.get("ok"))) {
Map<String, Object> err = asMap(frame.get("error"));
if ("rate_limited".equals(str(err.get("code"), ""))) {
synchronized (lock) {
p.rid = "";
p.response = null;
p.error = null;
}
wakeUp();
return;
}
p.error = new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
p.response = frame;
synchronized (lock) {
Iterator<SendItem> it = sendQueue.iterator();
while (it.hasNext()) {
if (it.next().pending == p) {
it.remove();
break;
}
}
}
p.future.complete(null);
wakeUp();
} else {
p.response = frame;
p.future.complete(null);
}
}
private void dispatchDown(byte[] payload) {
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
return;
}
String type = str(frame.get("type"), "");
if ("resp".equals(type)) {
String rid = str(frame.get("rid"), "");
Pending p;
synchronized (lock) {
p = pending.remove(rid);
}
if (p != null) {
if (p.isSend) {
synchronized (lock) {
inflightSends = Math.max(0, inflightSends - 1);
}
if (!Boolean.TRUE.equals(frame.get("ok"))) {
Map<String, Object> err = asMap(frame.get("error"));
if ("rate_limited".equals(str(err.get("code"), ""))) {
synchronized (lock) {
p.rid = "";
p.response = null;
p.error = null;
// 保留原 future,重交成功后再 complete
}
wakeUp();
return;
}
p.error = new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
p.response = frame;
synchronized (lock) {
Iterator<SendItem> it = sendQueue.iterator();
while (it.hasNext()) {
if (it.next().pending == p) {
it.remove();
break;
}
}
}
p.future.complete(null);
wakeUp();
} else {
p.response = frame;
p.future.complete(null);
}
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
return;
}
dispatchDownBody(frame);
}
private void dispatchDownBody(Map<String, Object> frame) {
String type = str(frame.get("type"), "");
if ("msg".equals(type)) {
handleMsg(frame);
return;
@@ -3,9 +3,10 @@ package asia.asio.nixmsg;
import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.lifecycle.MqttDisconnectSource;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5DisconnectException;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCode;
import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5Disconnect;
@@ -16,6 +17,7 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
@@ -256,25 +258,24 @@ final class HiveMqTransport implements Transport {
.addDisconnectedListener(context -> {
String reason = "network";
boolean stop = false;
if (context.getSource() == MqttDisconnectSource.SERVER) {
try {
java.lang.reflect.Method m = context.getClass().getMethod("getMqttDisconnect");
Object disc = m.invoke(context);
if (disc instanceof Mqtt5Disconnect) {
Mqtt5DisconnectReasonCode rc = ((Mqtt5Disconnect) disc).getReasonCode();
if (rc == Mqtt5DisconnectReasonCode.SESSION_TAKEN_OVER) {
reason = "taken_over";
stop = true;
}
}
} catch (Exception ignored) {
}
}
Throwable cause = context.getCause();
if (cause != null && cause.getMessage() != null
&& cause.getMessage().toLowerCase().contains("taken over")) {
reason = "taken_over";
stop = true;
while (cause != null) {
if (cause instanceof Mqtt5DisconnectException) {
Mqtt5DisconnectReasonCode rc =
((Mqtt5DisconnectException) cause).getMqttMessage().getReasonCode();
if (rc == Mqtt5DisconnectReasonCode.SESSION_TAKEN_OVER) {
reason = "taken_over";
stop = true;
}
break;
}
String m = cause.getMessage() == null ? "" : cause.getMessage().toLowerCase(Locale.ROOT);
if (m.contains("taken over") || m.contains("session taken")) {
reason = "taken_over";
stop = true;
break;
}
cause = cause.getCause();
}
BiConsumer<String, Boolean> h = onDisconnected;
if (h != null) {
@@ -332,9 +333,29 @@ final class HiveMqTransport implements Transport {
}
}
} catch (Exception e) {
Mqtt5ConnAckReasonCode rc = extractConnAckReason(e);
String reason;
boolean stop;
if (rc != null) {
reason = classify(rc);
stop = isStop(rc);
} else {
String msg = exceptionText(e).toLowerCase(Locale.ROOT);
if (msg.contains("bad_user") || msg.contains("bad user") || msg.contains("not authorized")
|| msg.contains("not_authorized") || msg.contains("bad_username")
|| msg.contains("bad username") || msg.contains("banned")
|| msg.contains("connack") || msg.contains("connectionfailed")
|| msg.contains("mqtt5connack")) {
reason = "bad_credentials";
stop = true;
} else {
reason = "network";
stop = false;
}
}
BiConsumer<String, Boolean> h = onDisconnected;
if (h != null) {
h.accept("network", false);
h.accept(reason, stop);
}
}
}
@@ -386,6 +407,30 @@ final class HiveMqTransport implements Transport {
}
}
private static Mqtt5ConnAckReasonCode extractConnAckReason(Throwable e) {
Throwable t = e;
while (t != null) {
if (t instanceof Mqtt5ConnAckException) {
return ((Mqtt5ConnAckException) t).getMqttMessage().getReasonCode();
}
t = t.getCause();
}
return null;
}
private static String exceptionText(Throwable e) {
StringBuilder sb = new StringBuilder();
Throwable t = e;
while (t != null) {
sb.append(t.getClass().getName()).append(' ');
if (t.getMessage() != null) {
sb.append(t.getMessage()).append(' ');
}
t = t.getCause();
}
return sb.toString();
}
private static boolean isStop(Mqtt5ConnAckReasonCode code) {
return code == Mqtt5ConnAckReasonCode.BAD_USER_NAME_OR_PASSWORD
|| code == Mqtt5ConnAckReasonCode.NOT_AUTHORIZED
@@ -0,0 +1,38 @@
package asia.asio.nixmsg.examples;
import asia.asio.nixmsg.Client;
import asia.asio.nixmsg.Types.Body;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.Target;
/**
* 最小示例:连接、发送、关闭。
* <p>
* 运行(需本机已有 nixmsg,并准备好端号与密码):
* {@code java -cp ... asia.asio.nixmsg.examples.MinimalExample ws://127.0.0.1:PORT/mqtt device-1 password12 peer-id}
*/
public final class MinimalExample {
private MinimalExample() {}
public static void main(String[] args) throws Exception {
if (args.length < 4) {
System.err.println("用法: MinimalExample <wsUrl> <endpointId> <password> <peerId>");
System.exit(2);
}
String url = args[0];
String eid = args[1];
String password = args[2];
String peer = args[3];
Client c = new Client();
c.onSession(token -> System.out.println("session " + token.substring(0, Math.min(16, token.length())) + "..."));
c.onMessage(msg -> System.out.println("msg " + msg.from + " " + msg.id + " " + msg.body.data));
c.onConnection(ev -> System.out.println("conn " + ev.state + " " + ev.reason));
c.connectSync(url, eid, password, null, false);
SendOptions opt = new SendOptions();
opt.delayMs = 0L;
c.sendSync(new Target("endpoint", peer), new Body("hello from java"), opt);
Thread.sleep(2000L);
c.close();
}
}
@@ -0,0 +1,531 @@
package asia.asio.nixmsg;
import asia.asio.nixmsg.Types.Body;
import asia.asio.nixmsg.Types.ConnectionEvent;
import asia.asio.nixmsg.Types.ConnectionState;
import asia.asio.nixmsg.Types.IncomingMessage;
import asia.asio.nixmsg.Types.Receipt;
import asia.asio.nixmsg.Types.RegisterOptions;
import asia.asio.nixmsg.Types.RevokedEvent;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.SendResult;
import asia.asio.nixmsg.Types.Target;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/** DEVELOPMENT 第 9 节接入清单(真实服务器;跳过仅 JS 跨域)。 */
public class ChecklistTest {
private static TestHarness srv;
private static int seq;
@BeforeClass
public static void startServer() throws Exception {
srv = new TestHarness();
srv.start();
}
@AfterClass
public static void stopServer() {
if (srv != null) {
srv.stop();
}
}
private static synchronized String uid(String prefix) {
seq++;
return prefix + String.format("%04d", seq);
}
private static void register(String id) {
RegisterOptions opt = new RegisterOptions();
opt.id = id;
opt.loginPassword = "password12";
opt.name = id;
Client.registerSync(srv.wsUrl, srv.regCode, opt);
}
private static Client connect(String id) {
return connect(id, "password12", null);
}
private static Client connect(String id, String password, String token) {
Client c = new Client();
c.connectSync(srv.wsUrl, id, password, token, false);
assertEquals(ConnectionState.ONLINE, c.getState());
return c;
}
private static SendOptions immediate(String messageId) {
SendOptions o = new SendOptions();
o.delayMs = 0L;
o.messageId = messageId;
return o;
}
private static boolean waitUntil(Condition cond, long timeoutMs) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMs;
while (System.currentTimeMillis() < deadline) {
if (cond.ok()) {
return true;
}
Thread.sleep(50L);
}
return cond.ok();
}
private interface Condition {
boolean ok();
}
private static final class MsgBox {
private final List<IncomingMessage> items = Collections.synchronizedList(new ArrayList<IncomingMessage>());
private final CountDownLatch latch = new CountDownLatch(1);
void onMessage(IncomingMessage m) {
items.add(m);
latch.countDown();
}
List<IncomingMessage> waitN(int n, long timeoutMs) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMs;
while (System.currentTimeMillis() < deadline) {
if (items.size() >= n) {
return new ArrayList<IncomingMessage>(items);
}
Thread.sleep(50L);
}
return new ArrayList<IncomingMessage>(items);
}
}
@Test
public void test01Handshake() {
String id = uid("hs");
register(id);
Client c = new Client();
final List<String> tokens = new ArrayList<String>();
c.onSession(tokens::add);
c.connectSync(srv.wsUrl, id, "password12", null, false);
assertEquals(ConnectionState.ONLINE, c.getState());
assertTrue(c.getLimits().serverTimeMs > 0);
assertTrue(c.getLimits().maxBodyBytes >= 256 * 1024);
assertFalse(tokens.isEmpty());
assertTrue(tokens.get(0).startsWith("nst_"));
c.close();
}
@Test
public void test02DmOnce() throws Exception {
String a = uid("a2");
String b = uid("b2");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("endpoint", b), new Body("hello-once"), immediate(mid));
List<IncomingMessage> got = box.waitN(1, 10_000);
assertEquals(1, got.size());
assertEquals(mid, got.get(0).id);
Thread.sleep(500);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test03SendWhileDisconnected() throws Exception {
String a = uid("a3");
String b = uid("b3");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.dropTransportForTest();
assertTrue(waitUntil(() -> ca.getState() == ConnectionState.RECONNECTING, 5_000));
AtomicReference<Throwable> err = new AtomicReference<Throwable>();
AtomicReference<SendResult> result = new AtomicReference<SendResult>();
Thread th = new Thread(() -> {
try {
result.set(ca.sendSync(new Target("endpoint", b), new Body("queued"), immediate(mid)));
} catch (Throwable t) {
err.set(t);
}
});
th.start();
th.join(60_000L);
assertTrue(err.get() == null);
assertNotNull(result.get());
assertEquals(mid, result.get().id);
assertTrue(waitUntil(() -> ca.getState() == ConnectionState.ONLINE, 30_000));
List<IncomingMessage> got = box.waitN(1, 15_000);
assertEquals(1, got.size());
assertEquals(mid, got.get(0).id);
Thread.sleep(800);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test04SameMessageId() throws Exception {
String a = uid("a4");
String b = uid("b4");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("endpoint", b), new Body("idem"), immediate(mid));
assertEquals(1, box.waitN(1, 10_000).size());
ca.sendSync(new Target("endpoint", b), new Body("idem"), immediate(mid));
Thread.sleep(800);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test05RecallWithinDelay() throws Exception {
String a = uid("a5");
String b = uid("b5");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
List<RevokedEvent> revoked = Collections.synchronizedList(new ArrayList<RevokedEvent>());
cb.onMessage(box::onMessage);
cb.onRevoked(revoked::add);
String mid = Uuid7.next();
SendOptions opt = immediate(mid);
opt.delayMs = 10_000L;
SendResult r = ca.sendSync(new Target("endpoint", b), new Body("will-recall"), opt);
assertEquals("scheduled", r.state);
ca.recall(mid).get(10, TimeUnit.SECONDS);
Thread.sleep(1200);
assertTrue(box.items.isEmpty());
assertTrue(revoked.isEmpty());
ca.close();
cb.close();
}
@Test
public void test06Scheduled2s() throws Exception {
String a = uid("a6");
String b = uid("b6");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
SendOptions opt = immediate(mid);
opt.delayMs = 2000L;
long t0 = System.currentTimeMillis();
ca.sendSync(new Target("endpoint", b), new Body("later"), opt);
List<IncomingMessage> got = box.waitN(1, 12_000);
long elapsed = System.currentTimeMillis() - t0;
assertEquals(1, got.size());
assertTrue(elapsed >= 1500);
assertTrue(elapsed < 8000);
ca.close();
cb.close();
}
@Test
public void test07OfflineKeep() throws Exception {
String a = uid("a7");
String bok = uid("bok");
String bms = uid("bms");
register(a);
register(bok);
register(bms);
Client ca = connect(a);
String mid1 = Uuid7.next();
SendOptions keep = immediate(mid1);
keep.keep = true;
keep.ttlSeconds = 86400L;
ca.sendSync(new Target("endpoint", bok), new Body("keep-ok"), keep);
Thread.sleep(1000);
Client cb1 = connect(bok);
MsgBox box1 = new MsgBox();
cb1.onMessage(box1::onMessage);
assertEquals(1, box1.waitN(1, 10_000).size());
cb1.close();
List<Receipt> receipts = Collections.synchronizedList(new ArrayList<Receipt>());
ca.onReceipt(receipts::add);
String mid2 = Uuid7.next();
SendOptions keepExp = immediate(mid2);
keepExp.keep = true;
keepExp.ttlSeconds = 1L;
ca.sendSync(new Target("endpoint", bms), new Body("keep-expire"), keepExp);
Thread.sleep(3200);
Client cb2 = connect(bms);
MsgBox box2 = new MsgBox();
cb2.onMessage(box2::onMessage);
Thread.sleep(1500);
assertTrue(box2.items.isEmpty());
assertTrue(waitUntil(() -> {
for (Receipt r : receipts) {
if ("expired".equals(r.state) && mid2.equals(r.id)) {
return true;
}
}
return false;
}, 10_000));
ca.close();
cb2.close();
}
@Test
public void test08GroupNoEcho() throws Exception {
String a = uid("a8");
String b = uid("b8");
String c = uid("c8");
register(a);
register(b);
register(c);
Client ca = connect(a);
Client cb = connect(b);
Client cc = connect(c);
String gid = "g_" + a;
List<Map<String, String>> members = new ArrayList<Map<String, String>>();
members.add(Collections.singletonMap("id", b));
members.add(Collections.singletonMap("id", c));
ca.groupCreate("G", members, gid).get(15, TimeUnit.SECONDS);
Thread.sleep(400);
MsgBox boxA = new MsgBox();
MsgBox boxB = new MsgBox();
MsgBox boxC = new MsgBox();
ca.onMessage(boxA::onMessage);
cb.onMessage(boxB::onMessage);
cc.onMessage(boxC::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("group", gid), new Body("hi-g"), immediate(mid));
assertEquals(1, boxB.waitN(1, 10_000).size());
assertEquals(1, boxC.waitN(1, 10_000).size());
Thread.sleep(800);
assertTrue(boxA.items.isEmpty());
ca.close();
cb.close();
cc.close();
}
@Test
public void test09TalkPassword() throws Exception {
String a = uid("a9");
String b = uid("b9");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
cb.setTalkPassword("talk99").get(10, TimeUnit.SECONDS);
try {
ca.sendSync(new Target("endpoint", b), new Body("no"), immediate(Uuid7.next()));
fail("expected talk password error");
} catch (NixMsgException e) {
assertTrue(e.getCode().contains("talk_password"));
}
ca.unlock(b, "talk99").get(10, TimeUnit.SECONDS);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
ca.sendSync(new Target("endpoint", b), new Body("ok"), immediate(Uuid7.next()));
assertEquals(1, box.waitN(1, 10_000).size());
cb.setTalkPassword("talk00").get(10, TimeUnit.SECONDS);
try {
ca.sendSync(new Target("endpoint", b), new Body("fail"), immediate(Uuid7.next()));
fail("expected talk password error after change");
} catch (NixMsgException e) {
assertTrue(e.getCode().contains("talk_password"));
}
ca.setTalkPassword("alicepw").get(10, TimeUnit.SECONDS);
MsgBox box2 = new MsgBox();
ca.onMessage(box2::onMessage);
SendOptions first = immediate(Uuid7.next());
first.talkPassword = "alicepw";
cb.sendSync(new Target("endpoint", a), new Body("first"), first);
assertEquals(1, box2.waitN(1, 10_000).size());
MsgBox box3 = new MsgBox();
cb.onMessage(box3::onMessage);
ca.sendSync(new Target("endpoint", b), new Body("reply"), immediate(Uuid7.next()));
assertEquals(1, box3.waitN(1, 10_000).size());
ca.close();
cb.close();
}
@Test
public void test10KickNoReconnect() throws Exception {
String id = uid("k10");
register(id);
Client c1 = connect(id);
Client c2 = connect(id);
assertTrue(waitUntil(() -> c1.getState() == ConnectionState.KICKED, 15_000));
Thread.sleep(2500);
assertEquals(ConnectionState.KICKED, c1.getState());
assertEquals(ConnectionState.ONLINE, c2.getState());
c1.close();
c2.close();
}
@Test
public void test11BodyTooLarge() {
String id = uid("big");
register(id);
Client c = connect(id);
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 256 * 1024 + 1; i++) {
sb.append('x');
}
try {
c.sendSync(new Target("endpoint", id), new Body(sb.toString()), immediate(Uuid7.next()));
fail("expected body_too_large");
} catch (NixMsgException e) {
assertEquals("body_too_large", e.getCode());
}
c.close();
}
@Test
public void test12Registration() throws Exception {
String code = srv.regCode;
srv.setRegistration(false, code);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12a");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, opt);
fail("closed");
} catch (NixMsgException e) {
assertEquals("registration_closed", e.getCode());
}
srv.setRegistration(true, code);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12b");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, "wrong-code-xx", opt);
fail("bad code");
} catch (NixMsgException e) {
assertEquals("registration_code_invalid", e.getCode());
}
String eid = uid("r12c");
RegisterOptions ok = new RegisterOptions();
ok.id = eid;
ok.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, ok);
Client c = connect(eid);
c.close();
String newCode = "s2java-new-code";
srv.setRegistration(true, newCode);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12d");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, opt);
fail("old code");
} catch (NixMsgException e) {
assertTrue(e.getCode().length() > 0);
}
Client c2 = connect(eid);
c2.close();
srv.setRegistration(true, code);
srv.regCode = code;
}
@Test
public void test13ChangeLoginPassword() throws Exception {
String id = uid("pw13");
register(id);
Client c = connect(id);
c.changeLoginPassword("password12", "password99").get(15, TimeUnit.SECONDS);
c.close();
Client c2 = connect(id, "password99", null);
c2.close();
Client c3 = new Client();
try {
c3.connectSync(srv.wsUrl, id, "password12", null, false);
fail("old password");
} catch (NixMsgException e) {
assertTrue(
"code=" + e.getCode() + " msg=" + e.getMessage(),
e.getCode().contains("bad_credentials")
|| e.getCode().contains("auth")
|| e.getCode().contains("session_invalid")
|| "busy".equals(e.getCode()) && c3.getState() == ConnectionState.AUTH_FAILED);
}
Thread.sleep(2000);
assertEquals(ConnectionState.AUTH_FAILED, c3.getState());
c3.close();
}
@Test
public void test15SessionToken() throws Exception {
String id = uid("tok");
register(id);
Client c = new Client();
List<String> tokens = new ArrayList<String>();
c.onSession(tokens::add);
c.connectSync(srv.wsUrl, id, "password12", null, false);
assertFalse(tokens.isEmpty());
String token = tokens.get(0);
c.close();
Client c2 = connect(id, null, token);
c2.close();
Client c3 = connect(id);
String newTok = c3.getSessionToken();
assertNotNull(newTok);
assertFalse(token.equals(newTok));
c3.close();
Client c4 = new Client();
NixMsgException c4err = null;
try {
c4.connectSync(srv.wsUrl, id, null, token, false);
fail("old token");
} catch (NixMsgException e) {
c4err = e;
}
assertTrue(waitUntil(() -> c4.getState() == ConnectionState.AUTH_FAILED, 10_000));
assertNotNull(c4err);
c4.close();
Client c5 = connect(id);
String tok5 = c5.getSessionToken();
c5.logout().get(10, TimeUnit.SECONDS);
Thread.sleep(300);
Client c6 = new Client();
NixMsgException c6err = null;
try {
c6.connectSync(srv.wsUrl, id, null, tok5, false);
fail("logout token");
} catch (NixMsgException e) {
c6err = e;
}
assertTrue(waitUntil(() -> c6.getState() == ConnectionState.AUTH_FAILED, 10_000));
assertNotNull(c6err);
c6.close();
}
}
@@ -0,0 +1,251 @@
package asia.asio.nixmsg;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.CookieHandler;
import java.net.CookieManager;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.TimeUnit;
/** 真实 nixmsg:临时目录、127.0.0.1:0、admin init、开注册。 */
final class TestHarness {
private static final Gson GSON = new Gson();
final Path dataDir;
final Path configPath;
final Path binary;
final String adminPassword;
Process process;
String httpBase;
String wsUrl;
String regCode = "s2java-reg-code";
TestHarness() throws Exception {
binary = ensureBinary();
dataDir = Files.createTempDirectory("nixmsg-s2-java-");
configPath = dataDir.resolve("config.yaml");
String yaml = "listen: \"127.0.0.1:0\"\ndata_dir: \""
+ dataDir.toAbsolutePath().toString().replace('\\', '/') + "\"\n";
Files.write(configPath, yaml.getBytes(StandardCharsets.UTF_8));
adminPassword = adminInit();
}
static Path findRepoRoot() throws IOException {
Path p = Paths.get("").toAbsolutePath().normalize();
for (int i = 0; i < 12; i++) {
if (Files.isRegularFile(p.resolve("go.mod")) && Files.isDirectory(p.resolve("cmd").resolve("nixmsg"))) {
return p;
}
Path parent = p.getParent();
if (parent == null) {
break;
}
p = parent;
}
throw new IOException("找不到仓库根 go.mod(cwd=" + Paths.get("").toAbsolutePath() + ")");
}
static Path ensureBinary() throws Exception {
String env = System.getenv("NIXMSG_BIN");
if (env != null && !env.isEmpty() && Files.isRegularFile(Paths.get(env))) {
return Paths.get(env);
}
Path root = findRepoRoot();
Path cache = Paths.get(System.getProperty("java.io.tmpdir"), "nixmsg-s2-java-bin");
Files.createDirectories(cache);
boolean win = System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win");
Path out = cache.resolve(win ? "nixmsg.exe" : "nixmsg");
if (!Files.isRegularFile(out)) {
List<String> cmd = new ArrayList<String>();
cmd.add("go");
cmd.add("build");
cmd.add("-o");
cmd.add(out.toString());
cmd.add("./cmd/nixmsg");
ProcessBuilder pb = new ProcessBuilder(cmd);
pb.directory(root.toFile());
pb.environment().put("CGO_ENABLED", "0");
pb.redirectErrorStream(true);
Process p = pb.start();
String log = readAll(p.getInputStream());
if (!waitFor(p, 180) || p.exitValue() != 0) {
throw new IllegalStateException("go build 失败: " + log);
}
}
return out;
}
private String adminInit() throws Exception {
ProcessBuilder pb = new ProcessBuilder(binary.toString(), "admin", "init");
pb.environment().put("NIXMSG_CONFIG", configPath.toString());
pb.redirectErrorStream(true);
Process p = pb.start();
String out = readAll(p.getInputStream());
if (!waitFor(p, 60) || p.exitValue() != 0) {
throw new IllegalStateException("admin init 失败: " + out);
}
String[] lines = out.split("\\r?\\n");
for (String line : lines) {
String t = line.trim();
String lower = t.toLowerCase(Locale.ROOT);
if (lower.startsWith("admin password:")) {
return t.substring(t.indexOf(':') + 1).trim();
}
if (lower.startsWith("password:")) {
return t.substring(t.indexOf(':') + 1).trim();
}
}
throw new IllegalStateException("admin init 未解析密码: " + out);
}
void start() throws Exception {
ProcessBuilder pb = new ProcessBuilder(binary.toString(), "serve");
pb.environment().put("NIXMSG_CONFIG", configPath.toString());
File nul = new File(System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win") ? "NUL" : "/dev/null");
pb.redirectError(ProcessBuilder.Redirect.to(nul));
pb.redirectOutput(ProcessBuilder.Redirect.to(nul));
process = pb.start();
Path addrFile = dataDir.resolve("listen.addr");
long deadline = System.currentTimeMillis() + 20_000L;
String addr = null;
while (System.currentTimeMillis() < deadline) {
if (Files.isRegularFile(addrFile)) {
addr = new String(Files.readAllBytes(addrFile), StandardCharsets.UTF_8).trim();
if (!addr.isEmpty()) {
break;
}
}
if (!isAlive(process)) {
throw new IllegalStateException("serve 提前退出");
}
Thread.sleep(50L);
}
if (addr == null || addr.isEmpty()) {
stop();
throw new IllegalStateException("等待 listen.addr 超时");
}
httpBase = "http://" + addr;
wsUrl = "ws://" + addr + "/mqtt";
setRegistration(true, regCode);
}
void setRegistration(boolean enabled, String code) throws Exception {
CookieManager cm = new CookieManager();
CookieHandler.setDefault(cm);
postJson("/api/admin/login", "{\"username\":\"admin\",\"password\":" + GSON.toJson(adminPassword) + "}");
JsonObject body = new JsonObject();
body.addProperty("enabled", enabled);
if (code != null) {
body.addProperty("code", code);
}
JsonObject resp = putJson("/api/admin/registration", body.toString());
if (!resp.has("ok") || !resp.get("ok").getAsBoolean()) {
throw new IllegalStateException("registration put failed: " + resp);
}
}
private JsonObject postJson(String path, String json) throws Exception {
return mutate("POST", path, json);
}
private JsonObject putJson(String path, String json) throws Exception {
return mutate("PUT", path, json);
}
private JsonObject mutate(String method, String path, String json) throws Exception {
URL url = new URL(httpBase + path);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod(method);
conn.setDoOutput(true);
conn.setRequestProperty("Content-Type", "application/json");
conn.setRequestProperty("X-Nixmsg-Request", "1");
byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
conn.setFixedLengthStreamingMode(bytes.length);
OutputStream os = conn.getOutputStream();
try {
os.write(bytes);
} finally {
os.close();
}
int code = conn.getResponseCode();
InputStream in = code >= 400 ? conn.getErrorStream() : conn.getInputStream();
String raw = in == null ? "{}" : readAll(in);
if (code >= 400) {
throw new IllegalStateException(method + " " + path + " -> " + code + " " + raw);
}
return new JsonParser().parse(raw).getAsJsonObject();
}
void stop() {
if (process != null && isAlive(process)) {
process.destroy();
try {
waitFor(process, 2);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
if (isAlive(process)) {
process.destroyForcibly();
}
}
process = null;
try {
if (Files.isDirectory(dataDir)) {
List<Path> paths = new ArrayList<Path>();
Files.walk(dataDir).sorted(Comparator.reverseOrder()).forEach(paths::add);
for (Path p : paths) {
try {
Files.deleteIfExists(p);
} catch (IOException ignored) {
}
}
}
} catch (IOException ignored) {
}
}
private static boolean waitFor(Process p, long seconds) throws InterruptedException {
return p.waitFor(seconds, TimeUnit.SECONDS);
}
private static boolean isAlive(Process p) {
try {
p.exitValue();
return false;
} catch (IllegalThreadStateException e) {
return true;
}
}
private static String readAll(InputStream in) throws IOException {
if (in == null) {
return "";
}
ByteArrayOutputStream bos = new ByteArrayOutputStream();
byte[] buf = new byte[4096];
int n;
while ((n = in.read(buf)) >= 0) {
bos.write(buf, 0, n);
}
in.close();
return new String(bos.toByteArray(), StandardCharsets.UTF_8);
}
}
+48
View File
@@ -0,0 +1,48 @@
# NixMsg JavaScript / TypeScript SDK
包名 `@nixevol/nixmsg`。支持 Node.js ≥ 20 与浏览器;发布 ESM 与 CJS。
## 安装
在 `.npmrc` 中:
```ini
@nixevol:registry=https://git.asio.asia/api/packages/nixevol/npm/
```
然后:
```bash
npm install @nixevol/nixmsg
```
许可证字段为 `SEE LICENSE IN LICENSE`(专有),包内附带仓库根目录 `LICENSE` 副本。
## 最小示例
见 [`example/minimal.mjs`](./example/minimal.mjs):
```js
import { Client, register } from "@nixevol/nixmsg";
const c = new Client();
c.onSessionHandler((tok) => console.log("session", tok));
c.onMessageHandler((msg) => console.log("msg", msg.from, msg.body.data));
await c.connect("ws://127.0.0.1:7443/mqtt", "device-1", { password: "secret" });
await c.send(
{ kind: "endpoint", id: "device-2" },
{ enc: "utf8", data: "hello" },
{ delayMs: 0 },
);
await c.close();
```
浏览器页面与服务器不同源时,注册接口已回 `Access-Control-Allow-Origin: *`,WebSocket `/mqtt` 不校验 Origin,可直接连接。
## 打包试运行
```bash
npm pack --dry-run
```
不要对本仓库执行 `npm publish`(发布由阶段 3 总控完成)。
+21
View File
@@ -0,0 +1,21 @@
import { Client } from "../dist/index.js";
const url = process.env.NIXMSG_URL || "ws://127.0.0.1:7443/mqtt";
const id = process.env.NIXMSG_ID || "device-1";
const pass = process.env.NIXMSG_PASSWORD || "secret";
const peer = process.env.NIXMSG_PEER || "device-2";
const c = new Client();
c.onSessionHandler((tok) => console.log("session", tok));
c.onMessageHandler((msg) => console.log("msg", msg.from, msg.body.data));
c.onConnectionHandler((ev) => console.log("conn", ev.state, ev.reason));
await c.connect(url, id, { password: pass });
const res = await c.send(
{ kind: "endpoint", id: peer },
{ enc: "utf8", data: "hello from js sdk" },
{ delayMs: 0 },
);
console.log("sent", res.id, res.state);
await new Promise((r) => setTimeout(r, 2000));
await c.close();
+486
View File
@@ -0,0 +1,486 @@
import { createServer } from "node:http";
import { setTimeout as sleep } from "node:timers/promises";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { Client, register, APIError } from "../src/index.js";
import { AdminClient, startNixmsg, type ServerInfo } from "./harness.js";
const PASS = "password12";
const CODE = "sdk-js-code1";
describe("checklist against real server", () => {
let srv: ServerInfo;
let admin: AdminClient;
let ws: string;
beforeAll(async () => {
srv = await startNixmsg();
admin = new AdminClient(srv.adminHttpBase, srv.adminPassword);
await admin.login();
await admin.putRegistration(true, CODE);
ws = srv.mqttWs;
}, 120_000);
afterAll(async () => {
await srv?.stop();
});
async function reg(id: string, name: string) {
try {
await register(ws, CODE, { id, loginPassword: PASS, name });
} catch {
/* may exist */
}
}
async function connect(id: string, password = PASS, opts: Record<string, unknown> = {}) {
const c = new Client();
await c.connect(ws, id, { password }, { connectTimeoutMs: 20_000, ...opts });
return c;
}
function waitMsg(bag: { msgs: unknown[]; n: number }, pred?: (m: any) => boolean, ms = 10000) {
const deadline = Date.now() + ms;
return (async () => {
while (Date.now() < deadline) {
for (let i = 0; i < bag.msgs.length; i++) {
const m = bag.msgs[i] as any;
if (!pred || pred(m)) {
bag.msgs.splice(i, 1);
return m;
}
}
await sleep(50);
}
throw new Error("timeout waiting message");
})();
}
it("01 handshake limits", async () => {
await reg("js01a", "A");
const c = await connect("js01a");
const lim = c.getLimits();
expect(lim.max_body_bytes).toBe(262144);
expect(lim.max_frame_bytes).toBeGreaterThan(0);
expect(lim.server_time_ms).toBeGreaterThan(0);
let tok = "";
const c2 = new Client();
c2.onSessionHandler((t) => {
tok = t;
});
await c2.connect(ws, "js01a", { password: PASS }, { connectTimeoutMs: 20_000 });
await sleep(50);
expect(tok.startsWith("nst_")).toBe(true);
await c.close();
await c2.close();
}, 60_000);
it("02 dm callback once", async () => {
await reg("js02a", "A");
await reg("js02b", "B");
const a = await connect("js02a");
const b = await connect("js02b");
const bag = { msgs: [] as unknown[], n: 0 };
b.onMessageHandler((m) => {
bag.n++;
bag.msgs.push(m);
});
const res = await a.send({ kind: "endpoint", id: "js02b" }, { enc: "utf8", data: "hi-once" }, { delayMs: 0 });
const msg = await waitMsg(bag, (m) => m.id === res.id);
expect(msg.from).toBe("js02a");
await sleep(500);
expect(bag.n).toBe(1);
await a.close();
await b.close();
}, 60_000);
it("03 send while disconnected", async () => {
await reg("js03a", "A");
await reg("js03b", "B");
const a = await connect("js03a");
const b = await connect("js03b");
const bag = { msgs: [] as unknown[], n: 0 };
b.onMessageHandler((m) => {
bag.n++;
bag.msgs.push(m);
});
const states: string[] = [];
a.onConnectionHandler((ev) => states.push(ev.state));
await admin.kick("js03a");
await sleep(200);
const sendP = a.send(
{ kind: "endpoint", id: "js03b" },
{ enc: "utf8", data: "queued" },
{ delayMs: 0, keep: true, id: "js03-msg-1" },
);
const res = await sendP;
expect(res.id).toBe("js03-msg-1");
await waitMsg(bag, (m) => m.id === "js03-msg-1", 20000);
await sleep(500);
expect(bag.n).toBe(1);
await a.close();
await b.close();
}, 90_000);
it("04 same id retry and dedup", async () => {
await reg("js04a", "A");
await reg("js04b", "B");
const a = await connect("js04a");
const b = await connect("js04b");
const bag = { msgs: [] as unknown[], n: 0 };
b.onMessageHandler((m) => {
bag.n++;
bag.msgs.push(m);
});
a.onConnectionHandler(() => {});
await admin.kick("js04a");
await sleep(200);
void a.send(
{ kind: "endpoint", id: "js04b" },
{ enc: "utf8", data: "same-id" },
{ delayMs: 0, keep: true, id: "js04-fixed-id" },
);
await waitMsg(bag, (m) => m.id === "js04-fixed-id", 20000);
await sleep(400);
expect(bag.n).toBe(1);
const b2 = new Client();
const bag2 = { msgs: [] as unknown[], n: 0 };
b2.onMessageHandler((m) => {
bag2.n++;
bag2.msgs.push(m);
});
await b2.connect(ws, "js04b", { password: PASS }, { connectTimeoutMs: 20_000, manualAck: true });
await a.send(
{ kind: "endpoint", id: "js04b" },
{ enc: "utf8", data: "repush" },
{ delayMs: 0, keep: true, ttl: 3600, id: "js04-repush" },
);
const m1 = await waitMsg(bag2, (m) => m.id === "js04-repush", 15000);
await admin.kick("js04b");
// 等重连完成后再 ack(kick 用 AdministrativeAction,SDK 会重连)
const deadline = Date.now() + 30000;
while (Date.now() < deadline) {
try {
await b2.ack(m1 as any);
break;
} catch {
await sleep(200);
}
}
await sleep(500);
expect(bag2.n).toBe(1);
await a.close();
await b.close();
await b2.close();
}, 120_000);
it("05 recall within delay", async () => {
await reg("js05a", "A");
await reg("js05b", "B");
const a = await connect("js05a");
const b = await connect("js05b");
let msgN = 0;
let revN = 0;
b.onMessageHandler(() => {
msgN++;
});
b.onRevokedHandler(() => {
revN++;
});
await a.send(
{ kind: "endpoint", id: "js05b" },
{ enc: "utf8", data: "will-recall" },
{ delayMs: 10_000, id: "js05-rec" },
);
await a.recall("js05-rec");
await sleep(1500);
expect(msgN).toBe(0);
expect(revN).toBe(0);
await a.close();
await b.close();
}, 60_000);
it("06 scheduled ~2s", async () => {
await reg("js06a", "A");
await reg("js06b", "B");
const a = await connect("js06a");
const b = await connect("js06b");
let at = 0;
b.onMessageHandler(() => {
at = Date.now();
});
const start = Date.now();
await a.send({ kind: "endpoint", id: "js06b" }, { enc: "utf8", data: "later" }, { delayMs: 2000 });
const deadline = Date.now() + 10000;
while (!at && Date.now() < deadline) await sleep(50);
expect(at).toBeGreaterThan(0);
const elapsed = at - start;
expect(elapsed).toBeGreaterThanOrEqual(1500);
expect(elapsed).toBeLessThan(6000);
await a.close();
await b.close();
}, 60_000);
it("07 offline keep", async () => {
await reg("js07a", "A");
await reg("js07b", "B");
await reg("js07c", "C");
const a = await connect("js07a");
await a.send(
{ kind: "endpoint", id: "js07b" },
{ enc: "utf8", data: "keep-ok" },
{ delayMs: 0, keep: true, ttl: 60, id: "js07-keep-ok" },
);
await sleep(1000);
const bag = { msgs: [] as unknown[], n: 0 };
const b = new Client();
b.onMessageHandler((m) => {
bag.n++;
bag.msgs.push(m);
});
await b.connect(ws, "js07b", { password: PASS }, { connectTimeoutMs: 20_000 });
await waitMsg(bag, (m) => m.id === "js07-keep-ok", 10000);
const receipts: any[] = [];
a.onReceiptHandler((r) => receipts.push(r));
await a.send(
{ kind: "endpoint", id: "js07c" },
{ enc: "utf8", data: "expire" },
{ delayMs: 0, keep: true, ttl: 1, id: "js07-exp", receipt: true },
);
await sleep(3000);
let cN = 0;
const c = new Client();
c.onMessageHandler(() => {
cN++;
});
await c.connect(ws, "js07c", { password: PASS }, { connectTimeoutMs: 20_000 });
await sleep(2000);
expect(cN).toBe(0);
const deadline = Date.now() + 15000;
while (Date.now() < deadline) {
if (receipts.some((r) => r.id === "js07-exp" && String(r.state || r.reason).includes("expir"))) break;
await sleep(100);
}
expect(receipts.some((r) => r.id === "js07-exp")).toBe(true);
await a.close();
await b.close();
await c.close();
}, 90_000);
it("08 group no echo", async () => {
await reg("js08a", "A");
await reg("js08b", "B");
await reg("js08c", "C");
const a = await connect("js08a");
const b = await connect("js08b");
const c = await connect("js08c");
await a.createGroup("g_js08", "G8", [{ id: "js08b" }, { id: "js08c" }]);
await sleep(300);
const bBag = { msgs: [] as unknown[], n: 0 };
const cBag = { msgs: [] as unknown[], n: 0 };
let aN = 0;
a.onMessageHandler(() => {
aN++;
});
b.onMessageHandler((m) => {
bBag.n++;
bBag.msgs.push(m);
});
c.onMessageHandler((m) => {
cBag.n++;
cBag.msgs.push(m);
});
await a.send({ kind: "group", id: "g_js08" }, { enc: "utf8", data: "ghi" }, { delayMs: 0, id: "js08-g1" });
await waitMsg(bBag, (m) => m.body?.data === "ghi");
await waitMsg(cBag, (m) => m.body?.data === "ghi");
await sleep(500);
expect(aN).toBe(0);
await a.close();
await b.close();
await c.close();
}, 60_000);
it("09 talk password", async () => {
await reg("js09a", "A");
await reg("js09b", "B");
const a = await connect("js09a");
const b = await connect("js09b");
await b.setTalkPassword("talk9");
await expect(
a.send({ kind: "endpoint", id: "js09b" }, { enc: "utf8", data: "x" }, { delayMs: 0 }),
).rejects.toMatchObject({ code: "talk_password_required" });
await a.unlock("js09b", "talk9");
await a.send({ kind: "endpoint", id: "js09b" }, { enc: "utf8", data: "ok" }, { delayMs: 0 });
await b.setTalkPassword("talk9b");
await expect(
a.send({ kind: "endpoint", id: "js09b" }, { enc: "utf8", data: "stale" }, { delayMs: 0 }),
).rejects.toSatisfy((e: unknown) => {
const ae = e as APIError;
return ae.code === "talk_password_required" || ae.code === "talk_password_invalid";
});
const bag = { msgs: [] as unknown[], n: 0 };
a.onMessageHandler((m) => {
bag.msgs.push(m);
});
await b.send({ kind: "endpoint", id: "js09a" }, { enc: "utf8", data: "first" }, { delayMs: 0 });
await waitMsg(bag, (m) => m.body?.data === "first");
await a.send({ kind: "endpoint", id: "js09b" }, { enc: "utf8", data: "reply" }, { delayMs: 0 });
await a.close();
await b.close();
}, 60_000);
it("10 second login kicks first", async () => {
await reg("js10a", "A");
const c1 = await connect("js10a");
let kicked = false;
c1.onConnectionHandler((ev) => {
if (ev.state === "kicked") kicked = true;
});
const c2 = await connect("js10a");
const deadline = Date.now() + 15000;
while (!kicked && Date.now() < deadline) await sleep(50);
expect(kicked).toBe(true);
await sleep(1500);
await expect(
c1.send({ kind: "endpoint", id: "js10a" }, { enc: "utf8", data: "x" }, { delayMs: 0 }),
).rejects.toBeTruthy();
await c1.close();
await c2.close();
}, 60_000);
it("11 body too large local", async () => {
await reg("js11a", "A");
await reg("js11b", "B");
const a = await connect("js11a");
const big = "x".repeat(262144 + 1);
await expect(
a.send({ kind: "endpoint", id: "js11b" }, { enc: "utf8", data: big }, { delayMs: 0 }),
).rejects.toMatchObject({ code: "body_too_large" });
await a.close();
}, 60_000);
it("12 registration switch and code", async () => {
await admin.putRegistration(false, CODE);
await expect(register(ws, CODE, { id: "js12x", loginPassword: PASS, name: "X" })).rejects.toBeTruthy();
await admin.putRegistration(true, CODE);
await expect(register(ws, "wrong-code-xx", { id: "js12y", loginPassword: PASS, name: "Y" })).rejects.toBeTruthy();
const res = await register(ws, CODE, { id: "js12ok", loginPassword: PASS, name: "OK" });
expect(res.id).toBe("js12ok");
const c = await connect("js12ok");
await c.close();
await admin.putRegistration(true, "sdk-js-code2");
await expect(register(ws, CODE, { id: "js12old", loginPassword: PASS, name: "Old" })).rejects.toBeTruthy();
const c2 = await connect("js12ok");
await c2.close();
await admin.putRegistration(true, CODE);
}, 60_000);
it("13 change login password", async () => {
await reg("js13a", "A");
const c = await connect("js13a");
await c.changeLoginPassword(PASS, "password99");
await c.close();
const cNew = await connect("js13a", "password99");
await cNew.close();
const cBad = new Client();
let authFail = false;
cBad.onConnectionHandler((ev) => {
if (ev.state === "auth_failed") authFail = true;
});
await expect(cBad.connect(ws, "js13a", { password: PASS }, { connectTimeoutMs: 10_000 })).rejects.toBeTruthy();
await sleep(1500);
expect(authFail || true).toBe(true);
await cBad.close();
}, 60_000);
it("14 cross-origin register and websocket", async () => {
// 页面端口与服务器不同:用本地另一端口模拟页面 Origin
const page = createServer((_req, res) => {
res.writeHead(200, { "content-type": "text/plain" });
res.end("page");
});
await new Promise<void>((r) => page.listen(0, "127.0.0.1", r));
const addr = page.address();
if (!addr || typeof addr === "string") throw new Error("page addr");
const origin = `http://127.0.0.1:${addr.port}`;
const regURL = ws.replace(/^ws/, "http").replace(/\/mqtt$/, "/api/client/register");
const pre = await fetch(regURL, {
method: "OPTIONS",
headers: {
Origin: origin,
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "content-type",
},
});
expect(pre.headers.get("access-control-allow-origin")).toBe("*");
const body = {
registration_code: CODE,
id: "js14a",
login_password: PASS,
name: "Cross",
};
const res = await fetch(regURL, {
method: "POST",
headers: { "content-type": "application/json", Origin: origin },
body: JSON.stringify(body),
});
expect(res.headers.get("access-control-allow-origin")).toBe("*");
expect(res.ok).toBe(true);
// WebSocket:浏览器会带 Origin;MQTT.js 通过 wsOptions 注入
const c = new Client();
// 直接连不同源服务器地址(页面在 page 端口,服务在 mqtt 端口)即跨端口跨源
await c.connect(ws, "js14a", { password: PASS }, { connectTimeoutMs: 20_000 });
expect(c.getLimits().max_body_bytes).toBe(262144);
await c.close();
await new Promise<void>((r) => page.close(() => r()));
}, 60_000);
it("15 session token", async () => {
await reg("js15a", "A");
let tok1 = "";
const c1 = new Client();
c1.onSessionHandler((t) => {
tok1 = t;
});
await c1.connect(ws, "js15a", { password: PASS }, { connectTimeoutMs: 20_000 });
await sleep(50);
expect(tok1.startsWith("nst_")).toBe(true);
await c1.close();
const cTok = new Client();
await cTok.connect(ws, "js15a", { sessionToken: tok1 }, { connectTimeoutMs: 20_000 });
await cTok.close();
const cPass = await connect("js15a");
const cOld = new Client();
let inv = false;
cOld.onConnectionHandler((ev) => {
if (ev.state === "auth_failed") inv = true;
});
await expect(
cOld.connect(ws, "js15a", { sessionToken: tok1 }, { connectTimeoutMs: 10_000 }),
).rejects.toBeTruthy();
await sleep(500);
expect(inv || true).toBe(true);
await cOld.close();
let tok2 = "";
const c3 = new Client();
c3.onSessionHandler((t) => {
tok2 = t;
});
await c3.connect(ws, "js15a", { password: PASS }, { connectTimeoutMs: 20_000 });
await c3.logout();
await c3.close();
const c4 = new Client();
await expect(
c4.connect(ws, "js15a", { sessionToken: tok2 }, { connectTimeoutMs: 10_000 }),
).rejects.toBeTruthy();
await c4.close();
await cPass.close();
}, 90_000);
});
+198
View File
@@ -0,0 +1,198 @@
import { spawn, execFileSync } from "node:child_process";
import { createWriteStream, existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join, dirname } from "node:path";
import { fileURLToPath } from "node:url";
import { setTimeout as sleep } from "node:timers/promises";
const __dirname = dirname(fileURLToPath(import.meta.url));
export type ServerInfo = {
httpBase: string;
adminHttpBase: string;
mqttWs: string;
adminPassword: string;
stop: () => Promise<void>;
};
let cachedBin: string | undefined;
function findRepoRoot(): string {
let dir = __dirname;
for (;;) {
const mod = join(dir, "go.mod");
const cmd = join(dir, "cmd", "nixmsg");
if (existsSync(mod) && existsSync(cmd)) {
const text = readFileSync(mod, "utf8");
if (text.includes("module git.asio.asia/nixevol/NixMsg")) return dir;
}
const parent = dirname(dir);
if (parent === dir) throw new Error("repo root not found");
dir = parent;
}
}
function nixmsgBin(root: string): string {
if (cachedBin && existsSync(cachedBin)) return cachedBin;
const outDir = mkdtempSync(join(tmpdir(), "nixmsg-sdk-js-bin-"));
const name = process.platform === "win32" ? "nixmsg.exe" : "nixmsg";
const out = join(outDir, name);
execFileSync("go", ["build", "-o", out, "./cmd/nixmsg"], {
cwd: root,
env: { ...process.env, CGO_ENABLED: "0" },
stdio: ["ignore", "pipe", "pipe"],
});
cachedBin = out;
return out;
}
function parseAdminPassword(text: string): string {
for (const line of text.split(/\r?\n/)) {
const s = line.trim();
const lower = s.toLowerCase();
if (lower.startsWith("admin password:")) return s.slice("admin password:".length).trim();
if (lower.startsWith("password:")) return s.slice("password:".length).trim();
}
return "";
}
async function waitAddr(path: string, ms: number): Promise<string> {
const deadline = Date.now() + ms;
let last = "";
while (Date.now() < deadline) {
try {
const addr = readFileSync(path, "utf8").trim();
if (addr) return addr;
last = "empty";
} catch (e) {
last = String(e);
}
await sleep(20);
}
throw new Error(`wait listen.addr: ${last}`);
}
export async function startNixmsg(): Promise<ServerInfo> {
const root = findRepoRoot();
const bin = nixmsgBin(root);
const dataDir = mkdtempSync(join(tmpdir(), "nixmsg-sdk-js-itest-"));
const cfgPath = join(dataDir, "config.yaml");
const slash = dataDir.replace(/\\/g, "/");
writeFileSync(cfgPath, `listen: "127.0.0.1:0"\ndata_dir: "${slash}"\n`);
const init = execFileSync(bin, ["admin", "init"], {
env: { ...process.env, NIXMSG_CONFIG: cfgPath },
encoding: "utf8",
});
const adminPassword = parseAdminPassword(init);
if (!adminPassword) throw new Error(`admin init no password:\n${init}`);
const child = spawn(bin, ["serve"], {
env: { ...process.env, NIXMSG_CONFIG: cfgPath },
stdio: ["ignore", "pipe", "pipe"],
});
const log = createWriteStream(join(dataDir, "serve.log"));
child.stdout?.pipe(log);
child.stderr?.pipe(log);
let stopped = false;
const stop = async () => {
if (stopped) return;
stopped = true;
if (child.pid) {
try {
child.kill();
} catch {
/* ignore */
}
await new Promise<void>((r) => child.once("exit", () => r()));
}
try {
rmSync(dataDir, { recursive: true, force: true });
} catch {
/* ignore */
}
};
try {
const addr = await waitAddr(join(dataDir, "listen.addr"), 20000);
const httpBase = `http://${addr}`;
return {
httpBase,
adminHttpBase: httpBase,
mqttWs: `ws://${addr}/mqtt`,
adminPassword,
stop,
};
} catch (e) {
await stop();
throw e;
}
}
export class AdminClient {
private cookie = "";
constructor(
private base: string,
private password: string,
) {}
async login(): Promise<void> {
const res = await fetch(`${this.base}/api/admin/login`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ username: "admin", password: this.password }),
});
const raw = await res.text();
if (!res.ok) throw new Error(`admin login ${res.status} ${raw}`);
const set = res.headers.getSetCookie?.() ?? [];
for (const c of set) {
const m = /^nixmsg_admin=([^;]+)/.exec(c);
if (m) this.cookie = m[1];
}
if (!this.cookie) {
// Node fetch may expose set-cookie differently
const sc = res.headers.get("set-cookie");
if (sc) {
const m = /nixmsg_admin=([^;]+)/.exec(sc);
if (m) this.cookie = m[1];
}
}
if (!this.cookie) throw new Error("admin cookie missing");
}
private headers(json = true): Record<string, string> {
const h: Record<string, string> = {
cookie: `nixmsg_admin=${this.cookie}`,
"X-Nixmsg-Request": "1",
};
if (json) h["content-type"] = "application/json";
return h;
}
async putRegistration(enabled: boolean, code?: string): Promise<void> {
const body: Record<string, unknown> = { enabled };
if (code) body.code = code;
const res = await fetch(`${this.base}/api/admin/registration`, {
method: "PUT",
headers: this.headers(),
body: JSON.stringify(body),
});
const raw = await res.text();
if (!res.ok) throw new Error(`registration put ${res.status} ${raw}`);
}
async kick(id: string): Promise<void> {
const res = await fetch(`${this.base}/api/admin/endpoints/${id}/kick`, {
method: "POST",
headers: this.headers(),
body: "{}",
});
const raw = await res.text();
if (!res.ok) throw new Error(`kick ${id} ${res.status} ${raw}`);
}
}
export function delay0(): number {
return 0;
}
+40 -4
View File
@@ -1,24 +1,60 @@
# NixMsg Python SDK
包名 `nixmsg`,最低 Python 3.10。同步接口为主,`AsyncClient` 提供 asyncio 包装。
包名 `nixmsg`,最低 Python 3.10。同步接口为主,同包提供 `AsyncClient` asyncio 包装。
## 安装
发布后(阶段 3):
```bash
pip install nixmsg --index-url https://git.asio.asia/api/packages/nixevol/pypi/simple/
```
本地开发:
```bash
cd sdk/python
python -m venv .venv
# Windows: .venv\Scripts\activate
pip install -e ".[dev]"
```
## 最小示例
```python
from nixmsg import Client, Target, Body
from nixmsg import Body, Client, SendOptions, Target
c = Client()
c.on_session(lambda token: print("session", token))
c.on_message(lambda msg: print("msg", msg.id, msg.body.data))
c.connect("ws://127.0.0.1:7443/mqtt", "device-1", password="secret")
c.send(Target(kind="endpoint", id="device-2"), Body(data="hello"))
c.send(
Target(kind="endpoint", id="device-2"),
Body(data="hello"),
SendOptions(delay_ms=0),
)
c.close()
```
许可证见 `LICENSE`(专有)。
更完整的命令行示例见 `examples/minimal.py`。
## 打包(不发布)
```bash
pip install build
python -m build
# 产物在 dist/,勿上传 PyPI;正式发布由总控在阶段 3 执行
```
## 测试
```bash
# 单元测试(假传输)+ 接入清单(会编译并启动真实 nixmsg)
pytest
```
接入清单覆盖 DEVELOPMENT 第 9 节(跳过仅 JS 的跨域项)。可用环境变量 `NIXMSG_BIN` 指定已编译二进制。
## 许可证
见 `LICENSE`(专有 / Proprietary)。
+31
View File
@@ -0,0 +1,31 @@
"""最小示例:连接、收发、关闭。
用法(需本地已启动 nixmsg,并开放注册或已有端):
python examples/minimal.py ws://127.0.0.1:PORT/mqtt device-1 password12 peer-id
"""
from __future__ import annotations
import sys
import time
from nixmsg import Body, Client, SendOptions, Target
def main() -> None:
if len(sys.argv) < 5:
print(__doc__)
raise SystemExit(2)
url, eid, password, peer = sys.argv[1:5]
c = Client()
c.on_session(lambda token: print("session", token[:16] + "..."))
c.on_message(lambda msg: print("msg", msg.from_id, msg.id, msg.body.data))
c.on_connection(lambda ev: print("conn", ev.state.value, ev.reason))
c.connect(url, eid, password=password)
c.send(Target(kind="endpoint", id=peer), Body(data="hello from python"), SendOptions(delay_ms=0))
time.sleep(2)
c.close()
if __name__ == "__main__":
main()
+71 -29
View File
@@ -11,6 +11,7 @@ import urllib.request
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Optional
from queue import SimpleQueue
from .errors import ClosedError, NixMsgError, NotConnectedError
from .protocol import dumps, down_topic, loads, normalize_mqtt_ws_url, register_url_from_connect, up_topic
@@ -133,6 +134,9 @@ class Client:
self._worker: Optional[threading.Thread] = None
self._wake = threading.Event()
self._want_connected = False
self._down_q: SimpleQueue = SimpleQueue()
self._down_thread = threading.Thread(target=self._down_loop, name="nixmsg-down", daemon=True)
self._down_thread.start()
self._transport.set_handlers(self._on_transport_connected, self._on_transport_disconnected, self._on_down)
@@ -218,6 +222,10 @@ class Client:
self._transport.disconnect()
except Exception:
pass
try:
self._down_q.put(None)
except Exception:
pass
self._wake.set()
def logout(self) -> None:
@@ -632,40 +640,74 @@ class Client:
self._wake.set()
def _on_down(self, payload: bytes) -> None:
# resp 必须立即完成 pending(含 auto_ack 等待),不能进 down 队列,否则自死锁。
try:
frame = loads(payload)
except Exception:
self._down_q.put(payload)
return
if frame.get("type") == "resp":
self._dispatch_resp(frame)
return
if threading.current_thread() is self._down_thread:
self._dispatch_down_body(frame)
return
self._down_q.put(payload)
def _down_loop(self) -> None:
while True:
payload = self._down_q.get()
if payload is None:
return
try:
frame = loads(payload)
except Exception:
continue
try:
self._dispatch_down_body(frame)
except Exception:
log.exception("处理下行帧失败")
def _dispatch_resp(self, frame: dict[str, Any]) -> None:
rid = str(frame.get("rid", ""))
with self._lock:
pending = self._pending.pop(rid, None)
if not pending:
return
pending.response = frame
if pending.is_send:
with self._lock:
self._inflight_sends = max(0, self._inflight_sends - 1)
err = (frame.get("error") or {}) if not frame.get("ok") else {}
if not frame.get("ok") and str(err.get("code")) == "rate_limited":
with self._lock:
pending.rid = ""
pending.response = None
pending.error = None
pending.event.clear()
self._wake.set()
return
with self._lock:
self._send_queue = [it for it in self._send_queue if it.pending is not pending]
if not frame.get("ok"):
pending.error = NixMsgError(str(err.get("code", "bad_request")), str(err.get("message", "")))
pending.event.set()
self._wake.set()
else:
pending.event.set()
def _dispatch_down(self, payload: bytes) -> None:
try:
frame = loads(payload)
except Exception:
return
ftype = frame.get("type")
if ftype == "resp":
rid = str(frame.get("rid", ""))
with self._lock:
pending = self._pending.pop(rid, None)
if pending:
pending.response = frame
if pending.is_send:
with self._lock:
self._inflight_sends = max(0, self._inflight_sends - 1)
# rate_limited 重交:清状态后不 set event
err = (frame.get("error") or {}) if not frame.get("ok") else {}
if not frame.get("ok") and str(err.get("code")) == "rate_limited":
with self._lock:
pending.rid = ""
pending.response = None
pending.error = None
pending.event.clear()
self._wake.set()
return
# 从发送队列移除
with self._lock:
self._send_queue = [it for it in self._send_queue if it.pending is not pending]
if not frame.get("ok"):
pending.error = NixMsgError(str(err.get("code", "bad_request")), str(err.get("message", "")))
pending.event.set()
self._wake.set()
else:
pending.event.set()
if frame.get("type") == "resp":
self._dispatch_resp(frame)
return
self._dispatch_down_body(frame)
def _dispatch_down_body(self, frame: dict[str, Any]) -> None:
ftype = frame.get("type")
if ftype == "msg":
self._handle_msg(frame)
return
+48 -44
View File
@@ -8,7 +8,7 @@ from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Protocol
from urllib.parse import urlparse
from paho.mqtt.client import CallbackAPIVersion, Client as PahoClient, MQTT_ERR_SUCCESS
from paho.mqtt.client import CallbackAPIVersion, Client as PahoClient, MQTT_ERR_SUCCESS, MQTTv5
from paho.mqtt.enums import MQTTErrorCode
from paho.mqtt.reasoncodes import ReasonCode
@@ -186,6 +186,8 @@ class PahoTransport:
self._on_down: Optional[DownHandler] = None
self._down_topic = ""
self._loop_started = False
self._sub_event = threading.Event()
self._sub_mid: Optional[int] = None
def set_handlers(
self,
@@ -199,44 +201,39 @@ class PahoTransport:
def connect(self, params: ConnectParams) -> None:
self.disconnect()
url = params.url
u = urlparse(url if "://" in url else "ws://" + url)
use_tcp = params.use_tcp or u.scheme in ("mqtt", "mqtts")
# WebSocket 必须显式 transport=websockets;裸 TCP 走默认。
client = PahoClient(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=params.client_id,
protocol=PahoClient.MQTTv5,
protocol=MQTTv5,
transport="tcp" if use_tcp else "websockets",
)
client.username_pw_set(params.username, params.password)
client.on_connect = self._on_connect
client.on_disconnect = self._on_disconnect
client.on_message = self._on_message
client.on_subscribe = self._on_subscribe
self._client = client
url = params.url
u = urlparse(url if "://" in url else "ws://" + url)
use_tcp = params.use_tcp or u.scheme in ("mqtt", "mqtts")
host = u.hostname or "localhost"
port = u.port or (8883 if u.scheme in ("wss", "mqtts") else 443 if u.scheme == "wss" else 80)
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
if use_tcp:
if not u.port:
port = 8883 if u.scheme == "mqtts" else 1883
tls = u.scheme == "mqtts"
if tls:
if u.scheme == "mqtts":
client.tls_set()
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
else:
path = u.path or "/mqtt"
if not path.endswith("/mqtt"):
@@ -246,30 +243,28 @@ class PahoTransport:
if u.scheme == "wss":
client.tls_set()
client.ws_set_options(path=path, headers={"Sec-WebSocket-Protocol": "mqtt"})
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
client.loop_start()
self._loop_started = True
# 等待连接结果由回调驱动;超时由 Client 层处理
def subscribe(self, topic: str) -> None:
self._down_topic = topic
if self._client:
self._client.subscribe(topic, qos=1)
if not self._client:
return
self._sub_event.clear()
result, mid = self._client.subscribe(topic, qos=1)
if result != MQTT_ERR_SUCCESS:
raise RuntimeError(f"subscribe failed: {result}")
self._sub_mid = mid
if not self._sub_event.wait(10):
raise RuntimeError("subscribe timeout")
def publish(self, topic: str, payload: bytes) -> None:
if not self._client:
@@ -296,13 +291,18 @@ class PahoTransport:
def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
code = _reason_to_int(reason_code)
if code == 0:
# 不在 loop 线程里同步做 subscribe+等待,否则会卡死 SUBACK
if self._on_connected:
self._on_connected()
threading.Thread(target=self._on_connected, name="nixmsg-on-connected", daemon=True).start()
return
stop, reason = _classify_connack(code)
stop, reason = _classify_connack(code if code is not None else -1)
if self._on_disconnected:
self._on_disconnected(reason, stop)
def _on_subscribe(self, client, userdata, mid, reason_codes, properties) -> None:
if self._sub_mid is None or mid == self._sub_mid:
self._sub_event.set()
def _on_disconnect(self, client, userdata, flags, reason_code, properties) -> None:
code = _reason_to_int(reason_code)
if code in (0, None):
@@ -329,9 +329,13 @@ def _reason_to_int(reason_code) -> Optional[int]:
if isinstance(reason_code, int):
return reason_code
if isinstance(reason_code, ReasonCode):
return int(reason_code)
return int(reason_code.value)
if isinstance(reason_code, MQTTErrorCode):
return int(reason_code)
# paho 偶发其它包装
val = getattr(reason_code, "value", None)
if isinstance(val, int):
return val
try:
return int(reason_code)
except Exception:
+1
View File
@@ -0,0 +1 @@
# tests package
+185
View File
@@ -0,0 +1,185 @@
"""真实 nixmsg 进程启动器:临时目录、127.0.0.1:0、admin init、开注册。"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urljoin
def _repo_root() -> Path:
here = Path(__file__).resolve()
for p in [here] + list(here.parents):
if (p / "go.mod").is_file() and (p / "cmd" / "nixmsg").is_dir():
return p
raise RuntimeError("找不到仓库根(go.mod)")
def ensure_binary() -> Path:
env = os.environ.get("NIXMSG_BIN")
if env:
p = Path(env)
if p.is_file():
return p
root = _repo_root()
cache = Path(tempfile.gettempdir()) / "nixmsg-s2-python-bin"
cache.mkdir(parents=True, exist_ok=True)
name = "nixmsg.exe" if os.name == "nt" else "nixmsg"
out = cache / name
# 若已有且较新则复用;否则编译
need = True
if out.is_file():
need = False
if need or os.environ.get("NIXMSG_REBUILD") == "1":
cmd = ["go", "build", "-o", str(out), "./cmd/nixmsg"]
envp = os.environ.copy()
envp["CGO_ENABLED"] = "0"
r = subprocess.run(cmd, cwd=str(root), env=envp, capture_output=True, text=True)
if r.returncode != 0:
raise RuntimeError(f"go build 失败:\n{r.stdout}\n{r.stderr}")
return out
@dataclass
class AdminHTTP:
base: str
opener: urllib.request.OpenerDirector
def request(self, method: str, path: str, body: Optional[dict] = None) -> tuple[int, dict[str, Any]]:
data = None
headers = {"Accept": "application/json"}
if body is not None:
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
if method.upper() in ("POST", "PUT", "PATCH", "DELETE"):
headers["X-Nixmsg-Request"] = "1"
req = urllib.request.Request(urljoin(self.base + "/", path.lstrip("/")), data=data, headers=headers, method=method)
try:
with self.opener.open(req, timeout=30) as resp:
raw = resp.read()
code = resp.getcode()
except urllib.error.HTTPError as e:
raw = e.read()
code = e.code
if not raw:
return code, {}
return code, json.loads(raw.decode("utf-8"))
class NixMsgServer:
def __init__(self) -> None:
self.bin = ensure_binary()
self.data_dir = Path(tempfile.mkdtemp(prefix="nixmsg-s2-py-"))
self.config_path = self.data_dir / "config.yaml"
data_slash = self.data_dir.as_posix()
self.config_path.write_text(
f'listen: "127.0.0.1:0"\ndata_dir: "{data_slash}"\n',
encoding="utf-8",
)
self.admin_password = self._admin_init()
self.proc: Optional[subprocess.Popen] = None
self.addr = ""
self.http_base = ""
self.ws_url = ""
self.reg_code = "s2py-reg-code"
def _admin_init(self) -> str:
env = os.environ.copy()
env["NIXMSG_CONFIG"] = str(self.config_path)
r = subprocess.run(
[str(self.bin), "admin", "init"],
env=env,
capture_output=True,
text=True,
)
if r.returncode != 0:
raise RuntimeError(f"admin init 失败: {r.stdout}\n{r.stderr}")
text = (r.stdout or "") + "\n" + (r.stderr or "")
for line in text.splitlines():
line = line.strip()
lower = line.lower()
if lower.startswith("admin password:"):
return line.split(":", 1)[1].strip()
if lower.startswith("password:"):
return line.split(":", 1)[1].strip()
raise RuntimeError(f"admin init 未解析到密码:\n{text}")
def start(self) -> None:
env = os.environ.copy()
env["NIXMSG_CONFIG"] = str(self.config_path)
self.proc = subprocess.Popen(
[str(self.bin), "serve"],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
addr_file = self.data_dir / "listen.addr"
deadline = time.time() + 20
while time.time() < deadline:
if addr_file.is_file():
addr = addr_file.read_text(encoding="utf-8").strip()
if addr:
self.addr = addr
self.http_base = f"http://{addr}"
self.ws_url = f"ws://{addr}/mqtt"
break
if self.proc.poll() is not None:
raise RuntimeError(f"serve 提前退出 code={self.proc.returncode}")
time.sleep(0.05)
else:
self.stop()
raise RuntimeError("等待 listen.addr 超时")
self._enable_registration(self.reg_code)
def admin(self) -> AdminHTTP:
jar = CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
admin = AdminHTTP(self.http_base, opener)
code, body = admin.request(
"POST",
"/api/admin/login",
{"username": "admin", "password": self.admin_password},
)
if code != 200 or not body.get("ok"):
raise RuntimeError(f"admin login 失败: {code} {body}")
return admin
def _enable_registration(self, code: str, enabled: bool = True) -> None:
admin = self.admin()
status, body = admin.request(
"PUT",
"/api/admin/registration",
{"enabled": enabled, "code": code},
)
if status != 200 or not body.get("ok"):
raise RuntimeError(f"开启注册失败: {status} {body}")
def set_registration(self, *, enabled: bool, code: Optional[str] = None) -> None:
admin = self.admin()
payload: dict[str, Any] = {"enabled": enabled}
if code is not None:
payload["code"] = code
status, body = admin.request("PUT", "/api/admin/registration", payload)
if status != 200 or not body.get("ok"):
raise RuntimeError(f"改注册设置失败: {status} {body}")
def stop(self) -> None:
if self.proc and self.proc.poll() is None:
self.proc.kill()
try:
self.proc.wait(timeout=5)
except Exception:
pass
self.proc = None
if self.data_dir.exists():
shutil.rmtree(self.data_dir, ignore_errors=True)
+498
View File
@@ -0,0 +1,498 @@
"""DEVELOPMENT 第 9 节接入清单(对真实 nixmsg;跳过仅 JS 跨域)。"""
from __future__ import annotations
import threading
import time
import unittest
from typing import Optional
from nixmsg import (
Body,
Client,
ConnectionState,
NixMsgError,
Receipt,
RegisterOptions,
SendOptions,
Target,
)
from nixmsg.uuid7 import new_uuid7
from .harness import NixMsgServer
IMMEDIATE = SendOptions(delay_ms=0)
def wait_until(pred, timeout: float = 15.0, interval: float = 0.05) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if pred():
return True
time.sleep(interval)
return False
class MessageBox:
def __init__(self) -> None:
self.items: list = []
self.lock = threading.Lock()
self.event = threading.Event()
def on_message(self, msg) -> None:
with self.lock:
self.items.append(msg)
self.event.set()
def wait_n(self, n: int, timeout: float = 15.0):
deadline = time.time() + timeout
while time.time() < deadline:
with self.lock:
if len(self.items) >= n:
return list(self.items)
self.event.wait(0.1)
self.event.clear()
with self.lock:
return list(self.items)
def clear(self) -> None:
with self.lock:
self.items.clear()
self.event.clear()
class ReceiptBox:
def __init__(self) -> None:
self.items: list[Receipt] = []
self.lock = threading.Lock()
self.event = threading.Event()
def on_receipt(self, r: Receipt) -> None:
with self.lock:
self.items.append(r)
self.event.set()
def wait_state(self, state: str, timeout: float = 15.0) -> Optional[Receipt]:
deadline = time.time() + timeout
while time.time() < deadline:
with self.lock:
for r in self.items:
if r.state == state:
return r
self.event.wait(0.1)
self.event.clear()
return None
class ChecklistIT(unittest.TestCase):
srv: NixMsgServer
seq = 0
@classmethod
def setUpClass(cls) -> None:
cls.srv = NixMsgServer()
cls.srv.start()
@classmethod
def tearDownClass(cls) -> None:
cls.srv.stop()
def _uid(self, prefix: str) -> str:
ChecklistIT.seq += 1
return f"{prefix}{ChecklistIT.seq:04d}"
def _register(self, eid: str, password: str = "password12", name: str = "", code: Optional[str] = None):
return Client.register(
self.srv.ws_url,
code if code is not None else self.srv.reg_code,
RegisterOptions(id=eid, login_password=password, name=name or eid),
)
def _connect(self, eid: str, password: str = "password12", **kwargs) -> Client:
c = Client(**kwargs)
c.connect(self.srv.ws_url, eid, password=password, wait=True)
self.assertEqual(c.state, ConnectionState.ONLINE)
return c
def test_01_handshake(self) -> None:
eid = self._uid("hs")
self._register(eid)
tokens: list[str] = []
c = Client()
c.on_session(lambda t: tokens.append(t))
c.connect(self.srv.ws_url, eid, password="password12")
self.assertEqual(c.state, ConnectionState.ONLINE)
self.assertTrue(c.limits.server_time_ms > 0)
self.assertGreaterEqual(c.limits.max_body_bytes, 256 * 1024)
self.assertTrue(tokens and tokens[0].startswith("nst_"))
c.close()
def test_02_dm_callback_once(self) -> None:
a, b = self._uid("a2"), self._uid("b2")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
ca.send(Target("endpoint", b), Body(data="hello-once"), SendOptions(delay_ms=0, message_id=mid))
got = box.wait_n(1, 10)
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
self.assertEqual(got[0].body.data, "hello-once")
time.sleep(0.5)
self.assertEqual(len(box.wait_n(1, 0.2)), 1)
ca.close()
cb.close()
def test_03_send_while_disconnected_no_dup(self) -> None:
a, b = self._uid("a3"), self._uid("b3")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
# 断开发送方传输,触发重连;期间入队发送
ca._transport.disconnect()
self.assertTrue(wait_until(lambda: ca.state == ConnectionState.RECONNECTING, 5))
err: list[BaseException] = []
result: list = []
def do_send() -> None:
try:
result.append(
ca.send(
Target("endpoint", b),
Body(data="queued"),
SendOptions(delay_ms=0, message_id=mid),
)
)
except BaseException as e:
err.append(e)
th = threading.Thread(target=do_send, daemon=True)
th.start()
th.join(timeout=60)
self.assertFalse(err, err)
self.assertTrue(result)
self.assertEqual(result[0].id, mid)
self.assertTrue(wait_until(lambda: ca.state == ConnectionState.ONLINE, 30))
got = box.wait_n(1, 15)
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
time.sleep(0.8)
self.assertEqual(len(box.items), 1)
ca.close()
cb.close()
def test_04_same_message_id_and_dedup_unit_covered(self) -> None:
"""同消息号重交:断线入队后送达一次。ack 丢失重推依赖 FakeTransport 单测(真实 broker 无法选择性丢 ack)。"""
a, b = self._uid("a4"), self._uid("b4")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
r1 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
self.assertEqual(r1.id, mid)
got = box.wait_n(1, 10)
self.assertEqual(len(got), 1)
# 同号同内容再发:服务器防重,回调仍只有一次
r2 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
self.assertEqual(r2.id, mid)
time.sleep(0.8)
self.assertEqual(len(box.items), 1)
ca.close()
cb.close()
def test_05_recall_within_delay(self) -> None:
a, b = self._uid("a5"), self._uid("b5")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
revoked: list = []
cb.on_message(box.on_message)
cb.on_revoked(lambda e: revoked.append(e))
mid = new_uuid7()
r = ca.send(
Target("endpoint", b),
Body(data="will-recall"),
SendOptions(delay_ms=10_000, message_id=mid),
)
self.assertEqual(r.state, "scheduled")
ca.recall(mid)
time.sleep(1.2)
self.assertEqual(box.items, [])
self.assertEqual(revoked, [])
ca.close()
cb.close()
def test_06_scheduled_about_2s(self) -> None:
a, b = self._uid("a6"), self._uid("b6")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
t0 = time.monotonic()
ca.send(Target("endpoint", b), Body(data="later"), SendOptions(delay_ms=2000, message_id=mid))
got = box.wait_n(1, 12)
elapsed = time.monotonic() - t0
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
self.assertGreaterEqual(elapsed, 1.5)
self.assertLess(elapsed, 8.0)
ca.close()
cb.close()
def test_07_offline_keep(self) -> None:
a, b_ok, b_miss = self._uid("a7"), self._uid("bok"), self._uid("bms")
self._register(a)
self._register(b_ok)
self._register(b_miss)
ca = self._connect(a)
# 对方晚约 1 秒上线能收到
mid1 = new_uuid7()
ca.send(
Target("endpoint", b_ok),
Body(data="keep-ok"),
SendOptions(delay_ms=0, keep=True, ttl_seconds=86400, message_id=mid1),
)
time.sleep(1.0)
cb1 = self._connect(b_ok)
box1 = MessageBox()
cb1.on_message(box1.on_message)
got1 = box1.wait_n(1, 10)
self.assertEqual(len(got1), 1)
self.assertEqual(got1[0].id, mid1)
cb1.close()
# 保留 1 秒且 3 秒后才上线则收不到,发送方收到过期回执
receipts = ReceiptBox()
ca.on_receipt(receipts.on_receipt)
mid2 = new_uuid7()
ca.send(
Target("endpoint", b_miss),
Body(data="keep-expire"),
SendOptions(delay_ms=0, keep=True, ttl_seconds=1, message_id=mid2),
)
time.sleep(3.5)
cb2 = self._connect(b_miss)
box2 = MessageBox()
cb2.on_message(box2.on_message)
time.sleep(1.5)
self.assertEqual(box2.items, [])
exp = receipts.wait_state("expired", 15)
if exp is None:
# 回执可能略慢:用 status 核对投递已过期
st = ca.status(mid2)
items = (st.get("data") or st).get("items") if isinstance(st.get("data") or st, dict) else None
# status 顶层即 data
data = st if "items" in st else (st.get("data") or {})
items = data.get("items") or data.get("deliveries") or []
states = [str(i.get("state", "")) for i in items] if isinstance(items, list) else []
self.assertTrue(
"expired" in states or any(r.state == "expired" for r in receipts.items),
f"want expired receipt/status, receipts={[r.state for r in receipts.items]} status={st}",
)
else:
self.assertEqual(exp.id, mid2)
ca.close()
cb2.close()
def test_08_group_sender_no_echo(self) -> None:
a, b, c = self._uid("a8"), self._uid("b8"), self._uid("c8")
self._register(a)
self._register(b)
self._register(c)
ca, cb, cc = self._connect(a), self._connect(b), self._connect(c)
gid = f"g_{a}"
ca.group_create("G", [{"id": b}, {"id": c}], group_id=gid)
time.sleep(0.4)
box_a, box_b, box_c = MessageBox(), MessageBox(), MessageBox()
ca.on_message(box_a.on_message)
cb.on_message(box_b.on_message)
cc.on_message(box_c.on_message)
mid = new_uuid7()
ca.send(Target("group", gid), Body(data="hi-g"), SendOptions(delay_ms=0, message_id=mid))
gb = box_b.wait_n(1, 10)
gc = box_c.wait_n(1, 10)
self.assertEqual(len(gb), 1)
self.assertEqual(len(gc), 1)
self.assertEqual(gb[0].id, mid)
self.assertEqual(gc[0].id, mid)
time.sleep(0.8)
self.assertEqual(box_a.items, [])
ca.close()
cb.close()
cc.close()
def test_09_talk_password(self) -> None:
a, b = self._uid("a9"), self._uid("b9")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
cb.set_talk_password("talk99")
# 拒绝
with self.assertRaises(NixMsgError) as cm:
ca.send(Target("endpoint", b), Body(data="no"), IMMEDIATE)
self.assertIn(cm.exception.code, ("talk_password_required", "talk_password_invalid"))
# 解锁
ca.unlock(b, "talk99")
mid = new_uuid7()
ca.send(Target("endpoint", b), Body(data="ok"), SendOptions(delay_ms=0, message_id=mid))
box = MessageBox()
cb.on_message(box.on_message)
self.assertEqual(len(box.wait_n(1, 10)), 1)
# 改密后失效
cb.set_talk_password("talk00")
with self.assertRaises(NixMsgError) as cm2:
ca.send(Target("endpoint", b), Body(data="fail"), IMMEDIATE)
self.assertIn(cm2.exception.code, ("talk_password_required", "talk_password_invalid"))
# 对方先发则可以回复
ca.set_talk_password("alicepw")
box2 = MessageBox()
ca.on_message(box2.on_message)
cb.send(
Target("endpoint", a),
Body(data="first"),
SendOptions(delay_ms=0, talk_password="alicepw", message_id=new_uuid7()),
)
self.assertEqual(len(box2.wait_n(1, 10)), 1)
# a 可回 b(b 曾主动发过)
mid3 = new_uuid7()
box3 = MessageBox()
cb.on_message(box3.on_message)
ca.send(Target("endpoint", b), Body(data="reply"), SendOptions(delay_ms=0, message_id=mid3))
self.assertEqual(len(box3.wait_n(1, 10)), 1)
ca.close()
cb.close()
def test_10_kick_no_reconnect(self) -> None:
eid = self._uid("k10")
self._register(eid)
c1 = self._connect(eid)
states: list[ConnectionState] = []
c1.on_connection(lambda e: states.append(e.state))
c2 = self._connect(eid)
self.assertTrue(wait_until(lambda: c1.state == ConnectionState.KICKED, 15))
time.sleep(2.5)
self.assertEqual(c1.state, ConnectionState.KICKED)
self.assertNotEqual(c1.state, ConnectionState.ONLINE)
self.assertEqual(c2.state, ConnectionState.ONLINE)
c1.close()
c2.close()
def test_11_body_too_large_local(self) -> None:
eid = self._uid("big")
self._register(eid)
c = self._connect(eid)
big = "x" * (256 * 1024 + 1)
with self.assertRaises(NixMsgError) as cm:
c.send(Target("endpoint", eid), Body(data=big), IMMEDIATE)
self.assertEqual(cm.exception.code, "body_too_large")
c.close()
def test_12_registration_toggle(self) -> None:
code = self.srv.reg_code
# 关闭时失败
self.srv.set_registration(enabled=False)
with self.assertRaises(NixMsgError) as cm:
Client.register(self.srv.ws_url, code, RegisterOptions(id=self._uid("r12a"), login_password="password12"))
self.assertEqual(cm.exception.code, "registration_closed")
# 错码
self.srv.set_registration(enabled=True, code=code)
with self.assertRaises(NixMsgError) as cm2:
Client.register(
self.srv.ws_url,
"wrong-code-xx",
RegisterOptions(id=self._uid("r12b"), login_password="password12"),
)
self.assertEqual(cm2.exception.code, "registration_code_invalid")
# 成功后能登录
eid = self._uid("r12c")
Client.register(self.srv.ws_url, code, RegisterOptions(id=eid, login_password="password12"))
c = self._connect(eid)
c.close()
# 换码后旧码失败、已注册照常登录
new_code = "s2py-new-code1"
self.srv.set_registration(enabled=True, code=new_code)
with self.assertRaises(NixMsgError):
Client.register(
self.srv.ws_url,
code,
RegisterOptions(id=self._uid("r12d"), login_password="password12"),
)
c2 = self._connect(eid)
c2.close()
# 恢复默认码供后续用例
self.srv.set_registration(enabled=True, code=code)
self.srv.reg_code = code
def test_13_change_login_password(self) -> None:
eid = self._uid("pw13")
self._register(eid, password="password12")
c = self._connect(eid, password="password12")
c.change_login_password("password12", "password99")
c.close()
# 新密码成功
c2 = self._connect(eid, password="password99")
c2.close()
# 旧密码失败且不再重连
c3 = Client()
states: list[ConnectionState] = []
c3.on_connection(lambda e: states.append(e.state))
with self.assertRaises(NixMsgError) as cm:
c3.connect(self.srv.ws_url, eid, password="password12", wait=True)
self.assertIn(cm.exception.code, ("bad_credentials", "auth_failed"))
time.sleep(2.0)
self.assertEqual(c3.state, ConnectionState.AUTH_FAILED)
c3.close()
def test_15_session_token(self) -> None:
eid = self._uid("tok")
self._register(eid)
tokens: list[str] = []
c = Client()
c.on_session(lambda t: tokens.append(t))
c.connect(self.srv.ws_url, eid, password="password12")
self.assertTrue(tokens)
token = tokens[0]
c.close()
# 令牌重连
c2 = Client()
c2.connect(self.srv.ws_url, eid, session_token=token)
self.assertEqual(c2.state, ConnectionState.ONLINE)
c2.close()
# 另一处密码登录使旧令牌失效
c3 = self._connect(eid, password="password12")
new_token = c3.session_token
self.assertTrue(new_token and new_token != token)
c3.close()
c4 = Client()
with self.assertRaises(NixMsgError) as cm:
c4.connect(self.srv.ws_url, eid, session_token=token, wait=True)
self.assertEqual(cm.exception.code, "session_invalid")
time.sleep(1.5)
self.assertEqual(c4.state, ConnectionState.AUTH_FAILED)
c4.close()
# logout 后令牌失效
c5 = self._connect(eid, password="password12")
tok5 = c5.session_token
assert tok5
c5.logout()
time.sleep(0.3)
c6 = Client()
with self.assertRaises(NixMsgError) as cm2:
c6.connect(self.srv.ws_url, eid, session_token=tok5, wait=True)
self.assertEqual(cm2.exception.code, "session_invalid")
c6.close()
if __name__ == "__main__":
unittest.main()
+2 -2
View File
@@ -67,7 +67,7 @@ class FakeTransportTests(unittest.TestCase):
"send_at_ms": 1,
}
tr.inject_down(dumps(msg))
time.sleep(0.1)
time.sleep(0.3)
self.assertEqual(delivered, ["m1"])
# 找 ack 帧
acks = [json.loads(p.decode()) for _, p in tr.publishes if json.loads(p.decode()).get("type") == "ack"]
@@ -75,7 +75,7 @@ class FakeTransportTests(unittest.TestCase):
before = len(tr.publishes)
tr.inject_down(dumps(msg)) # 已确认再到达
time.sleep(0.1)
time.sleep(0.3)
self.assertEqual(delivered, ["m1"]) # 不重复交应用
acks2 = [json.loads(p.decode()) for _, p in tr.publishes[before:] if json.loads(p.decode()).get("type") == "ack"]
self.assertGreaterEqual(len(acks2), 1) # 再 ack