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