From 226cc0c5768b409436ece162e39e1d8ba4743aba Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 08:50:57 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Python/Java=20SDK=20=E6=8E=A5=E5=85=A5?= =?UTF-8?q?=E6=B8=85=E5=8D=95=E4=B8=8E=E6=89=93=E5=8C=85=E6=96=87=E6=A1=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对真实 nixmsg 跑 DEVELOPMENT 第 9 节接入清单（跳过仅 JS 跨域），补 README/示例，并修 Paho/HiveMQ 真机联调死锁与鉴权分类。 --- docs/DEVIATIONS.md | 37 ++ sdk/java/README.md | 41 +- .../main/java/asia/asio/nixmsg/Client.java | 162 ++++-- .../main/java/asia/asio/nixmsg/Transport.java | 85 ++- .../asio/nixmsg/examples/MinimalExample.java | 38 ++ .../java/asia/asio/nixmsg/ChecklistTest.java | 531 ++++++++++++++++++ .../java/asia/asio/nixmsg/TestHarness.java | 251 +++++++++ sdk/python/README.md | 44 +- sdk/python/examples/minimal.py | 31 + sdk/python/src/nixmsg/client.py | 100 +++- sdk/python/src/nixmsg/transport.py | 92 +-- sdk/python/tests/__init__.py | 1 + sdk/python/tests/harness.py | 185 ++++++ sdk/python/tests/test_checklist.py | 498 ++++++++++++++++ sdk/python/tests/test_client.py | 4 +- 15 files changed, 1952 insertions(+), 148 deletions(-) create mode 100644 sdk/java/src/main/java/asia/asio/nixmsg/examples/MinimalExample.java create mode 100644 sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java create mode 100644 sdk/java/src/test/java/asia/asio/nixmsg/TestHarness.java create mode 100644 sdk/python/examples/minimal.py create mode 100644 sdk/python/tests/__init__.py create mode 100644 sdk/python/tests/harness.py create mode 100644 sdk/python/tests/test_checklist.py diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 6f82f17..c0827ee 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -816,6 +816,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 diff --git a/sdk/java/README.md b/sdk/java/README.md index c75d282..1b994cd 100644 --- a/sdk/java/README.md +++ b/sdk/java/README.md @@ -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 + + asia.asio.nixmsg + nixmsg-sdk + 0.1.0 + +``` + +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）。 diff --git a/sdk/java/src/main/java/asia/asio/nixmsg/Client.java b/sdk/java/src/main/java/asia/asio/nixmsg/Client.java index 19e9347..c541001 100644 --- a/sdk/java/src/main/java/asia/asio/nixmsg/Client.java +++ b/sdk/java/src/main/java/asia/asio/nixmsg/Client.java @@ -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 downQueue = new LinkedBlockingQueue(); + private final Thread downWorker; private Consumer sessionHandler; private Consumer 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 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 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 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 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 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 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 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 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 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 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 frame) { + String type = str(frame.get("type"), ""); if ("msg".equals(type)) { handleMsg(frame); return; diff --git a/sdk/java/src/main/java/asia/asio/nixmsg/Transport.java b/sdk/java/src/main/java/asia/asio/nixmsg/Transport.java index ed30519..42af4e1 100644 --- a/sdk/java/src/main/java/asia/asio/nixmsg/Transport.java +++ b/sdk/java/src/main/java/asia/asio/nixmsg/Transport.java @@ -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 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 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 diff --git a/sdk/java/src/main/java/asia/asio/nixmsg/examples/MinimalExample.java b/sdk/java/src/main/java/asia/asio/nixmsg/examples/MinimalExample.java new file mode 100644 index 0000000..111604d --- /dev/null +++ b/sdk/java/src/main/java/asia/asio/nixmsg/examples/MinimalExample.java @@ -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; + +/** + * 最小示例：连接、发送、关闭。 + *

+ * 运行（需本机已有 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 "); + 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(); + } +} diff --git a/sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java b/sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java new file mode 100644 index 0000000..a1ee220 --- /dev/null +++ b/sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java @@ -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 items = Collections.synchronizedList(new ArrayList()); + private final CountDownLatch latch = new CountDownLatch(1); + + void onMessage(IncomingMessage m) { + items.add(m); + latch.countDown(); + } + + List waitN(int n, long timeoutMs) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + if (items.size() >= n) { + return new ArrayList(items); + } + Thread.sleep(50L); + } + return new ArrayList(items); + } + } + + @Test + public void test01Handshake() { + String id = uid("hs"); + register(id); + Client c = new Client(); + final List tokens = new ArrayList(); + 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 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 err = new AtomicReference(); + AtomicReference result = new AtomicReference(); + 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 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 revoked = Collections.synchronizedList(new ArrayList()); + 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 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 receipts = Collections.synchronizedList(new ArrayList()); + 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> members = new ArrayList>(); + 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 tokens = new ArrayList(); + 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(); + } +} diff --git a/sdk/java/src/test/java/asia/asio/nixmsg/TestHarness.java b/sdk/java/src/test/java/asia/asio/nixmsg/TestHarness.java new file mode 100644 index 0000000..6c4c986 --- /dev/null +++ b/sdk/java/src/test/java/asia/asio/nixmsg/TestHarness.java @@ -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 cmd = new ArrayList(); + 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 paths = new ArrayList(); + 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); + } +} diff --git a/sdk/python/README.md b/sdk/python/README.md index cb65249..2ee04e3 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -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）。 diff --git a/sdk/python/examples/minimal.py b/sdk/python/examples/minimal.py new file mode 100644 index 0000000..1c0e456 --- /dev/null +++ b/sdk/python/examples/minimal.py @@ -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() diff --git a/sdk/python/src/nixmsg/client.py b/sdk/python/src/nixmsg/client.py index eb7f522..59c038f 100644 --- a/sdk/python/src/nixmsg/client.py +++ b/sdk/python/src/nixmsg/client.py @@ -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 diff --git a/sdk/python/src/nixmsg/transport.py b/sdk/python/src/nixmsg/transport.py index b92d7cf..f9b46d6 100644 --- a/sdk/python/src/nixmsg/transport.py +++ b/sdk/python/src/nixmsg/transport.py @@ -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: diff --git a/sdk/python/tests/__init__.py b/sdk/python/tests/__init__.py new file mode 100644 index 0000000..65140f2 --- /dev/null +++ b/sdk/python/tests/__init__.py @@ -0,0 +1 @@ +# tests package diff --git a/sdk/python/tests/harness.py b/sdk/python/tests/harness.py new file mode 100644 index 0000000..d315e6e --- /dev/null +++ b/sdk/python/tests/harness.py @@ -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) diff --git a/sdk/python/tests/test_checklist.py b/sdk/python/tests/test_checklist.py new file mode 100644 index 0000000..51e400f --- /dev/null +++ b/sdk/python/tests/test_checklist.py @@ -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() diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 706cf4d..b8928f2 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -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