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

feat: Python/Java SDK 接入清单与打包文档

Browse Source 
对真实 nixmsg 跑 DEVELOPMENT 第 9 节接入清单（跳过仅 JS 跨域），补 README/示例，并修 Paho/HiveMQ 真机联调死锁与鉴权分类。
This commit is contained in:
Nixevol
2026-09-30 08:57:06 +08:00
parent 2f750bcb0f
commit 1f715e4e51
15 changed files with 1952 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/DEVIATIONS.md
+37
View File
@@ -853,6 +853,43 @@
- 备选方案：依赖库自带重连再改 Clean Start（易漏）。
- 影响：无。
### S2-PY/JAVA 4–5 2026-09-30
1. **接入清单对真实 nixmsg，跳过仅 JS 跨域**
- 原条款：DEVELOPMENT 第 9 节 15 条；任务 4 用 T0.5 启动器起真实服务端。
- 实际做法：Python `tests/harness.py` + `test_checklist.py`、Java `TestHarness` + `ChecklistTest` 自行 `go build`/`admin init`/`serve`（临时目录、`127.0.0.1:0`），管理登录后 `PUT /api/admin/registration` 开注册。覆盖清单 1–13、15；第 14 条（仅 JS 跨域）不做。
- 原因：总控指示跳过 JS 专属跨域；不改服务器业务代码。
- 备选方案：复用 Go `test/harness` 包（SDK 测试不便依赖）。
- 影响：无。
2. **清单第 4 条「ack 丢失后服务器重推」未在真机选择性复现**
- 原条款：模拟 ack 丢失后服务器重推，SDK 自动再确认且不重复回调。
- 实际做法：集成测验证同消息号防重与断线入队重交送达一次；「选择性丢弃 SDK 发出的 ack 帧」在真实 broker 上做不到，去重再 ack 仍由假传输单测覆盖。
- 原因：不改服务器、无中间代理注入丢包。
- 备选方案：toxiproxy 按包过滤（超出本任务、且难按 MQTT 应用帧过滤）。
- 影响：清单 4 真机为部分通过；假传输路径完整。
3. **下行 `resp` 与业务帧分流，避免 auto_ack 自死锁**
- 原条款：自动模式回调后发 ack；回调串行。
- 实际做法：MQTT/`publishes` 回调里对 `resp` 立即完成 pending；`msg` 等进单线程队列再处理（可在队列线程里同步 `ack`/`request`）。Paho 使用 `MQTTv5` 常量与 `transport=websockets`，并等待 SUBACK。
- 原因：若 `resp` 与 `msg` 同队列，auto_ack 等待 `resp` 会永久卡住。
- 备选方案：ack 只发布不等待（弱化协议确认）。
- 影响：与 DEVELOPMENT 行为一致，修复真机联调阻塞。
4. **HiveMQ 鉴权失败与顶号原因码解析**
- 原条款：CONNACK 鉴权失败停重连；`0x8E` 顶号停重连。
- 实际做法：`connect().get()` 抛出的 `Mqtt5ConnAckException` / 文案含 `BAD_USER_*` 时归为 `bad_credentials`（令牌场景 Client 层改为 `session_invalid`）；断开原因从 `Mqtt5DisconnectException` 读 `SESSION_TAKEN_OVER`。`connectSync` 等到终态再返回，避免与 attemptConnect 竞态报 `busy/RECONNECTING`。
- 原因：HiveMQ 失败路径多为异常而非成功返回的 CONNACK 对象。
- 备选方案：无。
- 影响：无。
5. **任务 5：README/示例与打包试跑，不发布**
- 原条款：包名与许可证；工具试跑确认能打包；README 与最小示例；真正发布在 Z3。
- 实际做法：更新两端 README；Python `examples/minimal.py`；Java `asia.asio.nixmsg.examples.MinimalExample`；`python -m build` 产出 wheel/sdist；`mvn package -DskipTests` 产出 jar；`javap` major version 52（Java 8）。未上传 PyPI/Maven。
- 原因：本波范围。
- 备选方案：无。
- 影响：无。
## 测试交付 Q
### Q1 / Q4 骨架 2026-09-30
sdk/java/README.md
+37 -4
View File
@@ -2,7 +2,7 @@
坐标：`asia.asio.nixmsg:nixmsg-sdk`
包名：`asia.asio.nixmsg`
字节码目标：Java 8
字节码目标：Java 8（`maven.compiler.release=8`）
接口：`CompletableFuture`
## Android
@@ -17,7 +17,15 @@ Maven / Gradle 仓库：
https://git.asio.asia/api/packages/nixevol/maven
```
HiveMQ MQTT Client（含 WebSocket：`webSocketConfig` + `netty-codec-http`）。
```xml
<dependency>
<groupId>asia.asio.nixmsg</groupId>
<artifactId>nixmsg-sdk</artifactId>
<version>0.1.0</version>
</dependency>
```
HiveMQ MQTT Client（WebSocket：`webSocketConfig` + `netty-codec-http`）。
## 最小示例
@@ -26,9 +34,34 @@ Client c = new Client();
c.onSession(token -> { /* 应用保存 */ });
c.onMessage(msg -> System.out.println(msg.id + " " + msg.body.data));
c.connect("ws://127.0.0.1:7443/mqtt", "device-1", "secret", null)
.thenCompose(v -> c.send(new Types.Target("endpoint", "device-2"), new Types.Body("hello"), new Types.SendOptions()))
.thenCompose(v -> {
Types.SendOptions opt = new Types.SendOptions();
opt.delayMs = 0L;
return c.send(new Types.Target("endpoint", "device-2"), new Types.Body("hello"), opt);
})
.join();
c.close();
```
许可证见 `LICENSE`（专有）。
命令行示例类：`asia.asio.nixmsg.examples.MinimalExample`。
## 打包（不发布）
```bash
mvn package -DskipTests
# 产物 target/nixmsg-sdk-0.1.0.jar；勿部署到 Maven 仓库；正式发布由总控在阶段 3 执行
```
确认字节码为 8：`javap -v target/classes/asia/asio/nixmsg/Client.class | findstr major`（应为 52）。
## 测试
```bash
mvn test
```
含假传输单元测试与 DEVELOPMENT 第 9 节接入清单（会编译并启动真实 `nixmsg`）。可用环境变量 `NIXMSG_BIN` 指定已编译二进制。跳过仅 JS 的跨域项。
## 许可证
见 `LICENSE`（Proprietary）。
sdk/java/src/main/java/asia/asio/nixmsg/Client.java
+117 -45
View File
@@ -27,7 +27,9 @@ import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
@@ -82,6 +84,8 @@ public final class Client {
private final Object connWait = new Object();
private Thread worker;
private final Object wake = new Object();
private final BlockingQueue<byte[]> downQueue = new LinkedBlockingQueue<byte[]>();
private final Thread downWorker;
private Consumer<String> sessionHandler;
private Consumer<IncomingMessage> messageHandler;
@@ -106,6 +110,9 @@ public final class Client {
this.clientName = clientName;
this.connectTimeoutMs = connectTimeoutMs;
this.transport.setHandlers(this::onTransportConnected, this::onTransportDisconnected, this::onDown);
this.downWorker = new Thread(this::downLoop, "nixmsg-down");
this.downWorker.setDaemon(true);
this.downWorker.start();
}
public void onSession(Consumer<String> handler) { this.sessionHandler = handler; }
@@ -121,6 +128,11 @@ public final class Client {
public String getSessionToken() { return sessionToken; }
public long getClockSkewMs() { return clockSkewMs; }
/** 同包测试用：断开底层传输以触发重连与发送队列重交。 */
void dropTransportForTest() {
transport.disconnect();
}
public CompletableFuture<Void> connect(String url, String endpointId, String password, String sessionToken) {
return connect(url, endpointId, password, sessionToken, false);
}
@@ -157,9 +169,17 @@ public final class Client {
}
long deadline = System.currentTimeMillis() + connectTimeoutMs + 5000;
synchronized (connWait) {
while (!connReady.get() && System.currentTimeMillis() < deadline) {
while (System.currentTimeMillis() < deadline) {
ConnectionState s = state;
if (s == ConnectionState.ONLINE
|| s == ConnectionState.AUTH_FAILED
|| s == ConnectionState.KICKED
|| s == ConnectionState.OFFLINE
|| handshakeError != null) {
break;
}
try {
connWait.wait(200);
connWait.wait(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
break;
@@ -200,6 +220,7 @@ public final class Client {
transport.disconnect();
} catch (Exception ignored) {
}
downQueue.offer(new byte[0]); // 空载荷哨兵：downLoop 见 closed 退出
wakeUp();
}
@@ -716,57 +737,108 @@ public final class Client {
}
private void onDown(byte[] payload) {
// resp 立即完成 pending，避免 down 工作线程在 autoAck 等待时自死锁。
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
downQueue.offer(payload);
return;
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
return;
}
downQueue.offer(payload);
}
private void downLoop() {
while (true) {
try {
byte[] payload = downQueue.take();
if (payload.length == 0 && closed) {
return;
}
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
continue;
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
} else {
dispatchDownBody(frame);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
} catch (Exception e) {
LOG.log(Level.WARNING, "处理下行帧失败", e);
}
}
}
private void dispatchResp(Map<String, Object> frame) {
String rid = str(frame.get("rid"), "");
Pending p;
synchronized (lock) {
p = pending.remove(rid);
}
if (p == null) {
return;
}
if (p.isSend) {
synchronized (lock) {
inflightSends = Math.max(0, inflightSends - 1);
}
if (!Boolean.TRUE.equals(frame.get("ok"))) {
Map<String, Object> err = asMap(frame.get("error"));
if ("rate_limited".equals(str(err.get("code"), ""))) {
synchronized (lock) {
p.rid = "";
p.response = null;
p.error = null;
}
wakeUp();
return;
}
p.error = new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
p.response = frame;
synchronized (lock) {
Iterator<SendItem> it = sendQueue.iterator();
while (it.hasNext()) {
if (it.next().pending == p) {
it.remove();
break;
}
}
}
p.future.complete(null);
wakeUp();
} else {
p.response = frame;
p.future.complete(null);
}
}
private void dispatchDown(byte[] payload) {
Map<String, Object> frame;
try {
frame = Protocol.loads(payload);
} catch (Exception e) {
return;
}
String type = str(frame.get("type"), "");
if ("resp".equals(type)) {
String rid = str(frame.get("rid"), "");
Pending p;
synchronized (lock) {
p = pending.remove(rid);
}
if (p != null) {
if (p.isSend) {
synchronized (lock) {
inflightSends = Math.max(0, inflightSends - 1);
}
if (!Boolean.TRUE.equals(frame.get("ok"))) {
Map<String, Object> err = asMap(frame.get("error"));
if ("rate_limited".equals(str(err.get("code"), ""))) {
synchronized (lock) {
p.rid = "";
p.response = null;
p.error = null;
// 保留原 future，重交成功后再 complete
}
wakeUp();
return;
}
p.error = new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
p.response = frame;
synchronized (lock) {
Iterator<SendItem> it = sendQueue.iterator();
while (it.hasNext()) {
if (it.next().pending == p) {
it.remove();
break;
}
}
}
p.future.complete(null);
wakeUp();
} else {
p.response = frame;
p.future.complete(null);
}
}
if ("resp".equals(str(frame.get("type"), ""))) {
dispatchResp(frame);
return;
}
dispatchDownBody(frame);
}
private void dispatchDownBody(Map<String, Object> frame) {
String type = str(frame.get("type"), "");
if ("msg".equals(type)) {
handleMsg(frame);
return;
sdk/java/src/main/java/asia/asio/nixmsg/Transport.java
+65 -20
View File
@@ -3,9 +3,10 @@ package asia.asio.nixmsg;
import com.hivemq.client.mqtt.MqttClient;
import com.hivemq.client.mqtt.MqttGlobalPublishFilter;
import com.hivemq.client.mqtt.datatypes.MqttQos;
import com.hivemq.client.mqtt.lifecycle.MqttDisconnectSource;
import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5ConnAckException;
import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5DisconnectException;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAckReasonCode;
import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5Disconnect;
@@ -16,6 +17,7 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
@@ -256,25 +258,24 @@ final class HiveMqTransport implements Transport {
.addDisconnectedListener(context -> {
String reason = "network";
boolean stop = false;
if (context.getSource() == MqttDisconnectSource.SERVER) {
try {
java.lang.reflect.Method m = context.getClass().getMethod("getMqttDisconnect");
Object disc = m.invoke(context);
if (disc instanceof Mqtt5Disconnect) {
Mqtt5DisconnectReasonCode rc = ((Mqtt5Disconnect) disc).getReasonCode();
if (rc == Mqtt5DisconnectReasonCode.SESSION_TAKEN_OVER) {
reason = "taken_over";
stop = true;
}
}
} catch (Exception ignored) {
}
}
Throwable cause = context.getCause();
if (cause != null && cause.getMessage() != null
&& cause.getMessage().toLowerCase().contains("taken over")) {
reason = "taken_over";
stop = true;
while (cause != null) {
if (cause instanceof Mqtt5DisconnectException) {
Mqtt5DisconnectReasonCode rc =
((Mqtt5DisconnectException) cause).getMqttMessage().getReasonCode();
if (rc == Mqtt5DisconnectReasonCode.SESSION_TAKEN_OVER) {
reason = "taken_over";
stop = true;
}
break;
}
String m = cause.getMessage() == null ? "" : cause.getMessage().toLowerCase(Locale.ROOT);
if (m.contains("taken over") || m.contains("session taken")) {
reason = "taken_over";
stop = true;
break;
}
cause = cause.getCause();
}
BiConsumer<String, Boolean> h = onDisconnected;
if (h != null) {
@@ -332,9 +333,29 @@ final class HiveMqTransport implements Transport {
}
}
} catch (Exception e) {
Mqtt5ConnAckReasonCode rc = extractConnAckReason(e);
String reason;
boolean stop;
if (rc != null) {
reason = classify(rc);
stop = isStop(rc);
} else {
String msg = exceptionText(e).toLowerCase(Locale.ROOT);
if (msg.contains("bad_user") || msg.contains("bad user") || msg.contains("not authorized")
|| msg.contains("not_authorized") || msg.contains("bad_username")
|| msg.contains("bad username") || msg.contains("banned")
|| msg.contains("connack") || msg.contains("connectionfailed")
|| msg.contains("mqtt5connack")) {
reason = "bad_credentials";
stop = true;
} else {
reason = "network";
stop = false;
}
}
BiConsumer<String, Boolean> h = onDisconnected;
if (h != null) {
h.accept("network", false);
h.accept(reason, stop);
}
}
}
@@ -386,6 +407,30 @@ final class HiveMqTransport implements Transport {
}
}
private static Mqtt5ConnAckReasonCode extractConnAckReason(Throwable e) {
Throwable t = e;
while (t != null) {
if (t instanceof Mqtt5ConnAckException) {
return ((Mqtt5ConnAckException) t).getMqttMessage().getReasonCode();
}
t = t.getCause();
}
return null;
}
private static String exceptionText(Throwable e) {
StringBuilder sb = new StringBuilder();
Throwable t = e;
while (t != null) {
sb.append(t.getClass().getName()).append(' ');
if (t.getMessage() != null) {
sb.append(t.getMessage()).append(' ');
}
t = t.getCause();
}
return sb.toString();
}
private static boolean isStop(Mqtt5ConnAckReasonCode code) {
return code == Mqtt5ConnAckReasonCode.BAD_USER_NAME_OR_PASSWORD
|| code == Mqtt5ConnAckReasonCode.NOT_AUTHORIZED
sdk/java/src/main/java/asia/asio/nixmsg/examples/MinimalExample.java
+38
View File
@@ -0,0 +1,38 @@
package asia.asio.nixmsg.examples;
import asia.asio.nixmsg.Client;
import asia.asio.nixmsg.Types.Body;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.Target;
/**
* 最小示例：连接、发送、关闭。
* <p>
* 运行（需本机已有 nixmsg，并准备好端号与密码）：
* {@code java -cp ... asia.asio.nixmsg.examples.MinimalExample ws://127.0.0.1:PORT/mqtt device-1 password12 peer-id}
*/
public final class MinimalExample {
private MinimalExample() {}
public static void main(String[] args) throws Exception {
if (args.length < 4) {
System.err.println("用法: MinimalExample <wsUrl> <endpointId> <password> <peerId>");
System.exit(2);
}
String url = args[0];
String eid = args[1];
String password = args[2];
String peer = args[3];
Client c = new Client();
c.onSession(token -> System.out.println("session " + token.substring(0, Math.min(16, token.length())) + "..."));
c.onMessage(msg -> System.out.println("msg " + msg.from + " " + msg.id + " " + msg.body.data));
c.onConnection(ev -> System.out.println("conn " + ev.state + " " + ev.reason));
c.connectSync(url, eid, password, null, false);
SendOptions opt = new SendOptions();
opt.delayMs = 0L;
c.sendSync(new Target("endpoint", peer), new Body("hello from java"), opt);
Thread.sleep(2000L);
c.close();
}
}
sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java
+531
View File
@@ -0,0 +1,531 @@
package asia.asio.nixmsg;
import asia.asio.nixmsg.Types.Body;
import asia.asio.nixmsg.Types.ConnectionEvent;
import asia.asio.nixmsg.Types.ConnectionState;
import asia.asio.nixmsg.Types.IncomingMessage;
import asia.asio.nixmsg.Types.Receipt;
import asia.asio.nixmsg.Types.RegisterOptions;
import asia.asio.nixmsg.Types.RevokedEvent;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.SendResult;
import asia.asio.nixmsg.Types.Target;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/** DEVELOPMENT 第 9 节接入清单（真实服务器；跳过仅 JS 跨域）。 */
public class ChecklistTest {
private static TestHarness srv;
private static int seq;
@BeforeClass
public static void startServer() throws Exception {
srv = new TestHarness();
srv.start();
}
@AfterClass
public static void stopServer() {
if (srv != null) {
srv.stop();
}
}
private static synchronized String uid(String prefix) {
seq++;
return prefix + String.format("%04d", seq);
}
private static void register(String id) {
RegisterOptions opt = new RegisterOptions();
opt.id = id;
opt.loginPassword = "password12";
opt.name = id;
Client.registerSync(srv.wsUrl, srv.regCode, opt);
}
private static Client connect(String id) {
return connect(id, "password12", null);
}
private static Client connect(String id, String password, String token) {
Client c = new Client();
c.connectSync(srv.wsUrl, id, password, token, false);
assertEquals(ConnectionState.ONLINE, c.getState());
return c;
}
private static SendOptions immediate(String messageId) {
SendOptions o = new SendOptions();
o.delayMs = 0L;
o.messageId = messageId;
return o;
}
private static boolean waitUntil(Condition cond, long timeoutMs) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMs;
while (System.currentTimeMillis() < deadline) {
if (cond.ok()) {
return true;
}
Thread.sleep(50L);
}
return cond.ok();
}
private interface Condition {
boolean ok();
}
private static final class MsgBox {
private final List<IncomingMessage> items = Collections.synchronizedList(new ArrayList<IncomingMessage>());
private final CountDownLatch latch = new CountDownLatch(1);
void onMessage(IncomingMessage m) {
items.add(m);
latch.countDown();
}
List<IncomingMessage> waitN(int n, long timeoutMs) throws InterruptedException {
long deadline = System.currentTimeMillis() + timeoutMs;
while (System.currentTimeMillis() < deadline) {
if (items.size() >= n) {
return new ArrayList<IncomingMessage>(items);
}
Thread.sleep(50L);
}
return new ArrayList<IncomingMessage>(items);
}
}
@Test
public void test01Handshake() {
String id = uid("hs");
register(id);
Client c = new Client();
final List<String> tokens = new ArrayList<String>();
c.onSession(tokens::add);
c.connectSync(srv.wsUrl, id, "password12", null, false);
assertEquals(ConnectionState.ONLINE, c.getState());
assertTrue(c.getLimits().serverTimeMs > 0);
assertTrue(c.getLimits().maxBodyBytes >= 256 * 1024);
assertFalse(tokens.isEmpty());
assertTrue(tokens.get(0).startsWith("nst_"));
c.close();
}
@Test
public void test02DmOnce() throws Exception {
String a = uid("a2");
String b = uid("b2");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("endpoint", b), new Body("hello-once"), immediate(mid));
List<IncomingMessage> got = box.waitN(1, 10_000);
assertEquals(1, got.size());
assertEquals(mid, got.get(0).id);
Thread.sleep(500);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test03SendWhileDisconnected() throws Exception {
String a = uid("a3");
String b = uid("b3");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.dropTransportForTest();
assertTrue(waitUntil(() -> ca.getState() == ConnectionState.RECONNECTING, 5_000));
AtomicReference<Throwable> err = new AtomicReference<Throwable>();
AtomicReference<SendResult> result = new AtomicReference<SendResult>();
Thread th = new Thread(() -> {
try {
result.set(ca.sendSync(new Target("endpoint", b), new Body("queued"), immediate(mid)));
} catch (Throwable t) {
err.set(t);
}
});
th.start();
th.join(60_000L);
assertTrue(err.get() == null);
assertNotNull(result.get());
assertEquals(mid, result.get().id);
assertTrue(waitUntil(() -> ca.getState() == ConnectionState.ONLINE, 30_000));
List<IncomingMessage> got = box.waitN(1, 15_000);
assertEquals(1, got.size());
assertEquals(mid, got.get(0).id);
Thread.sleep(800);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test04SameMessageId() throws Exception {
String a = uid("a4");
String b = uid("b4");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("endpoint", b), new Body("idem"), immediate(mid));
assertEquals(1, box.waitN(1, 10_000).size());
ca.sendSync(new Target("endpoint", b), new Body("idem"), immediate(mid));
Thread.sleep(800);
assertEquals(1, box.items.size());
ca.close();
cb.close();
}
@Test
public void test05RecallWithinDelay() throws Exception {
String a = uid("a5");
String b = uid("b5");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
List<RevokedEvent> revoked = Collections.synchronizedList(new ArrayList<RevokedEvent>());
cb.onMessage(box::onMessage);
cb.onRevoked(revoked::add);
String mid = Uuid7.next();
SendOptions opt = immediate(mid);
opt.delayMs = 10_000L;
SendResult r = ca.sendSync(new Target("endpoint", b), new Body("will-recall"), opt);
assertEquals("scheduled", r.state);
ca.recall(mid).get(10, TimeUnit.SECONDS);
Thread.sleep(1200);
assertTrue(box.items.isEmpty());
assertTrue(revoked.isEmpty());
ca.close();
cb.close();
}
@Test
public void test06Scheduled2s() throws Exception {
String a = uid("a6");
String b = uid("b6");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
String mid = Uuid7.next();
SendOptions opt = immediate(mid);
opt.delayMs = 2000L;
long t0 = System.currentTimeMillis();
ca.sendSync(new Target("endpoint", b), new Body("later"), opt);
List<IncomingMessage> got = box.waitN(1, 12_000);
long elapsed = System.currentTimeMillis() - t0;
assertEquals(1, got.size());
assertTrue(elapsed >= 1500);
assertTrue(elapsed < 8000);
ca.close();
cb.close();
}
@Test
public void test07OfflineKeep() throws Exception {
String a = uid("a7");
String bok = uid("bok");
String bms = uid("bms");
register(a);
register(bok);
register(bms);
Client ca = connect(a);
String mid1 = Uuid7.next();
SendOptions keep = immediate(mid1);
keep.keep = true;
keep.ttlSeconds = 86400L;
ca.sendSync(new Target("endpoint", bok), new Body("keep-ok"), keep);
Thread.sleep(1000);
Client cb1 = connect(bok);
MsgBox box1 = new MsgBox();
cb1.onMessage(box1::onMessage);
assertEquals(1, box1.waitN(1, 10_000).size());
cb1.close();
List<Receipt> receipts = Collections.synchronizedList(new ArrayList<Receipt>());
ca.onReceipt(receipts::add);
String mid2 = Uuid7.next();
SendOptions keepExp = immediate(mid2);
keepExp.keep = true;
keepExp.ttlSeconds = 1L;
ca.sendSync(new Target("endpoint", bms), new Body("keep-expire"), keepExp);
Thread.sleep(3200);
Client cb2 = connect(bms);
MsgBox box2 = new MsgBox();
cb2.onMessage(box2::onMessage);
Thread.sleep(1500);
assertTrue(box2.items.isEmpty());
assertTrue(waitUntil(() -> {
for (Receipt r : receipts) {
if ("expired".equals(r.state) && mid2.equals(r.id)) {
return true;
}
}
return false;
}, 10_000));
ca.close();
cb2.close();
}
@Test
public void test08GroupNoEcho() throws Exception {
String a = uid("a8");
String b = uid("b8");
String c = uid("c8");
register(a);
register(b);
register(c);
Client ca = connect(a);
Client cb = connect(b);
Client cc = connect(c);
String gid = "g_" + a;
List<Map<String, String>> members = new ArrayList<Map<String, String>>();
members.add(Collections.singletonMap("id", b));
members.add(Collections.singletonMap("id", c));
ca.groupCreate("G", members, gid).get(15, TimeUnit.SECONDS);
Thread.sleep(400);
MsgBox boxA = new MsgBox();
MsgBox boxB = new MsgBox();
MsgBox boxC = new MsgBox();
ca.onMessage(boxA::onMessage);
cb.onMessage(boxB::onMessage);
cc.onMessage(boxC::onMessage);
String mid = Uuid7.next();
ca.sendSync(new Target("group", gid), new Body("hi-g"), immediate(mid));
assertEquals(1, boxB.waitN(1, 10_000).size());
assertEquals(1, boxC.waitN(1, 10_000).size());
Thread.sleep(800);
assertTrue(boxA.items.isEmpty());
ca.close();
cb.close();
cc.close();
}
@Test
public void test09TalkPassword() throws Exception {
String a = uid("a9");
String b = uid("b9");
register(a);
register(b);
Client ca = connect(a);
Client cb = connect(b);
cb.setTalkPassword("talk99").get(10, TimeUnit.SECONDS);
try {
ca.sendSync(new Target("endpoint", b), new Body("no"), immediate(Uuid7.next()));
fail("expected talk password error");
} catch (NixMsgException e) {
assertTrue(e.getCode().contains("talk_password"));
}
ca.unlock(b, "talk99").get(10, TimeUnit.SECONDS);
MsgBox box = new MsgBox();
cb.onMessage(box::onMessage);
ca.sendSync(new Target("endpoint", b), new Body("ok"), immediate(Uuid7.next()));
assertEquals(1, box.waitN(1, 10_000).size());
cb.setTalkPassword("talk00").get(10, TimeUnit.SECONDS);
try {
ca.sendSync(new Target("endpoint", b), new Body("fail"), immediate(Uuid7.next()));
fail("expected talk password error after change");
} catch (NixMsgException e) {
assertTrue(e.getCode().contains("talk_password"));
}
ca.setTalkPassword("alicepw").get(10, TimeUnit.SECONDS);
MsgBox box2 = new MsgBox();
ca.onMessage(box2::onMessage);
SendOptions first = immediate(Uuid7.next());
first.talkPassword = "alicepw";
cb.sendSync(new Target("endpoint", a), new Body("first"), first);
assertEquals(1, box2.waitN(1, 10_000).size());
MsgBox box3 = new MsgBox();
cb.onMessage(box3::onMessage);
ca.sendSync(new Target("endpoint", b), new Body("reply"), immediate(Uuid7.next()));
assertEquals(1, box3.waitN(1, 10_000).size());
ca.close();
cb.close();
}
@Test
public void test10KickNoReconnect() throws Exception {
String id = uid("k10");
register(id);
Client c1 = connect(id);
Client c2 = connect(id);
assertTrue(waitUntil(() -> c1.getState() == ConnectionState.KICKED, 15_000));
Thread.sleep(2500);
assertEquals(ConnectionState.KICKED, c1.getState());
assertEquals(ConnectionState.ONLINE, c2.getState());
c1.close();
c2.close();
}
@Test
public void test11BodyTooLarge() {
String id = uid("big");
register(id);
Client c = connect(id);
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 256 * 1024 + 1; i++) {
sb.append('x');
}
try {
c.sendSync(new Target("endpoint", id), new Body(sb.toString()), immediate(Uuid7.next()));
fail("expected body_too_large");
} catch (NixMsgException e) {
assertEquals("body_too_large", e.getCode());
}
c.close();
}
@Test
public void test12Registration() throws Exception {
String code = srv.regCode;
srv.setRegistration(false, code);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12a");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, opt);
fail("closed");
} catch (NixMsgException e) {
assertEquals("registration_closed", e.getCode());
}
srv.setRegistration(true, code);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12b");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, "wrong-code-xx", opt);
fail("bad code");
} catch (NixMsgException e) {
assertEquals("registration_code_invalid", e.getCode());
}
String eid = uid("r12c");
RegisterOptions ok = new RegisterOptions();
ok.id = eid;
ok.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, ok);
Client c = connect(eid);
c.close();
String newCode = "s2java-new-code";
srv.setRegistration(true, newCode);
try {
RegisterOptions opt = new RegisterOptions();
opt.id = uid("r12d");
opt.loginPassword = "password12";
Client.registerSync(srv.wsUrl, code, opt);
fail("old code");
} catch (NixMsgException e) {
assertTrue(e.getCode().length() > 0);
}
Client c2 = connect(eid);
c2.close();
srv.setRegistration(true, code);
srv.regCode = code;
}
@Test
public void test13ChangeLoginPassword() throws Exception {
String id = uid("pw13");
register(id);
Client c = connect(id);
c.changeLoginPassword("password12", "password99").get(15, TimeUnit.SECONDS);
c.close();
Client c2 = connect(id, "password99", null);
c2.close();
Client c3 = new Client();
try {
c3.connectSync(srv.wsUrl, id, "password12", null, false);
fail("old password");
} catch (NixMsgException e) {
assertTrue(
"code=" + e.getCode() + " msg=" + e.getMessage(),
e.getCode().contains("bad_credentials")
|| e.getCode().contains("auth")
|| e.getCode().contains("session_invalid")
|| "busy".equals(e.getCode()) && c3.getState() == ConnectionState.AUTH_FAILED);
}
Thread.sleep(2000);
assertEquals(ConnectionState.AUTH_FAILED, c3.getState());
c3.close();
}
@Test
public void test15SessionToken() throws Exception {
String id = uid("tok");
register(id);
Client c = new Client();
List<String> tokens = new ArrayList<String>();
c.onSession(tokens::add);
c.connectSync(srv.wsUrl, id, "password12", null, false);
assertFalse(tokens.isEmpty());
String token = tokens.get(0);
c.close();
Client c2 = connect(id, null, token);
c2.close();
Client c3 = connect(id);
String newTok = c3.getSessionToken();
assertNotNull(newTok);
assertFalse(token.equals(newTok));
c3.close();
Client c4 = new Client();
NixMsgException c4err = null;
try {
c4.connectSync(srv.wsUrl, id, null, token, false);
fail("old token");
} catch (NixMsgException e) {
c4err = e;
}
assertTrue(waitUntil(() -> c4.getState() == ConnectionState.AUTH_FAILED, 10_000));
assertNotNull(c4err);
c4.close();
Client c5 = connect(id);
String tok5 = c5.getSessionToken();
c5.logout().get(10, TimeUnit.SECONDS);
Thread.sleep(300);
Client c6 = new Client();
NixMsgException c6err = null;
try {
c6.connectSync(srv.wsUrl, id, null, tok5, false);
fail("logout token");
} catch (NixMsgException e) {
c6err = e;
}
assertTrue(waitUntil(() -> c6.getState() == ConnectionState.AUTH_FAILED, 10_000));
assertNotNull(c6err);
c6.close();
}
}
sdk/java/src/test/java/asia/asio/nixmsg/TestHarness.java
+251
View File
@@ -0,0 +1,251 @@
package asia.asio.nixmsg;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.CookieHandler;
import java.net.CookieManager;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.TimeUnit;
/** 真实 nixmsg：临时目录、127.0.0.1:0、admin init、开注册。 */
final class TestHarness {
private static final Gson GSON = new Gson();
final Path dataDir;
final Path configPath;
final Path binary;
final String adminPassword;
Process process;
String httpBase;
String wsUrl;
String regCode = "s2java-reg-code";
TestHarness() throws Exception {
binary = ensureBinary();
dataDir = Files.createTempDirectory("nixmsg-s2-java-");
configPath = dataDir.resolve("config.yaml");
String yaml = "listen: \"127.0.0.1:0\"\ndata_dir: \""
+ dataDir.toAbsolutePath().toString().replace('\\', '/') + "\"\n";
Files.write(configPath, yaml.getBytes(StandardCharsets.UTF_8));
adminPassword = adminInit();
}
static Path findRepoRoot() throws IOException {
Path p = Paths.get("").toAbsolutePath().normalize();
for (int i = 0; i < 12; i++) {
if (Files.isRegularFile(p.resolve("go.mod")) && Files.isDirectory(p.resolve("cmd").resolve("nixmsg"))) {
return p;
}
Path parent = p.getParent();
if (parent == null) {
break;
}
p = parent;
}
throw new IOException("找不到仓库根 go.mod（cwd=" + Paths.get("").toAbsolutePath() + ")");
}
static Path ensureBinary() throws Exception {
String env = System.getenv("NIXMSG_BIN");
if (env != null && !env.isEmpty() && Files.isRegularFile(Paths.get(env))) {
return Paths.get(env);
}
Path root = findRepoRoot();
Path cache = Paths.get(System.getProperty("java.io.tmpdir"), "nixmsg-s2-java-bin");
Files.createDirectories(cache);
boolean win = System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win");
Path out = cache.resolve(win ? "nixmsg.exe" : "nixmsg");
if (!Files.isRegularFile(out)) {
List<String> cmd = new ArrayList<String>();
cmd.add("go");
cmd.add("build");
cmd.add("-o");
cmd.add(out.toString());
cmd.add("./cmd/nixmsg");
ProcessBuilder pb = new ProcessBuilder(cmd);
pb.directory(root.toFile());
pb.environment().put("CGO_ENABLED", "0");
pb.redirectErrorStream(true);
Process p = pb.start();
String log = readAll(p.getInputStream());
if (!waitFor(p, 180) || p.exitValue() != 0) {
throw new IllegalStateException("go build 失败: " + log);
}
}
return out;
}
private String adminInit() throws Exception {
ProcessBuilder pb = new ProcessBuilder(binary.toString(), "admin", "init");
pb.environment().put("NIXMSG_CONFIG", configPath.toString());
pb.redirectErrorStream(true);
Process p = pb.start();
String out = readAll(p.getInputStream());
if (!waitFor(p, 60) || p.exitValue() != 0) {
throw new IllegalStateException("admin init 失败: " + out);
}
String[] lines = out.split("\\r?\\n");
for (String line : lines) {
String t = line.trim();
String lower = t.toLowerCase(Locale.ROOT);
if (lower.startsWith("admin password:")) {
return t.substring(t.indexOf(':') + 1).trim();
}
if (lower.startsWith("password:")) {
return t.substring(t.indexOf(':') + 1).trim();
}
}
throw new IllegalStateException("admin init 未解析密码: " + out);
}
void start() throws Exception {
ProcessBuilder pb = new ProcessBuilder(binary.toString(), "serve");
pb.environment().put("NIXMSG_CONFIG", configPath.toString());
File nul = new File(System.getProperty("os.name", "").toLowerCase(Locale.ROOT).contains("win") ? "NUL" : "/dev/null");
pb.redirectError(ProcessBuilder.Redirect.to(nul));
pb.redirectOutput(ProcessBuilder.Redirect.to(nul));
process = pb.start();
Path addrFile = dataDir.resolve("listen.addr");
long deadline = System.currentTimeMillis() + 20_000L;
String addr = null;
while (System.currentTimeMillis() < deadline) {
if (Files.isRegularFile(addrFile)) {
addr = new String(Files.readAllBytes(addrFile), StandardCharsets.UTF_8).trim();
if (!addr.isEmpty()) {
break;
}
}
if (!isAlive(process)) {
throw new IllegalStateException("serve 提前退出");
}
Thread.sleep(50L);
}
if (addr == null || addr.isEmpty()) {
stop();
throw new IllegalStateException("等待 listen.addr 超时");
}
httpBase = "http://" + addr;
wsUrl = "ws://" + addr + "/mqtt";
setRegistration(true, regCode);
}
void setRegistration(boolean enabled, String code) throws Exception {
CookieManager cm = new CookieManager();
CookieHandler.setDefault(cm);
postJson("/api/admin/login", "{\"username\":\"admin\",\"password\":" + GSON.toJson(adminPassword) + "}");
JsonObject body = new JsonObject();
body.addProperty("enabled", enabled);
if (code != null) {
body.addProperty("code", code);
}
JsonObject resp = putJson("/api/admin/registration", body.toString());
if (!resp.has("ok") || !resp.get("ok").getAsBoolean()) {
throw new IllegalStateException("registration put failed: " + resp);
}
}
private JsonObject postJson(String path, String json) throws Exception {
return mutate("POST", path, json);
}
private JsonObject putJson(String path, String json) throws Exception {
return mutate("PUT", path, json);
}
private JsonObject mutate(String method, String path, String json) throws Exception {
URL url = new URL(httpBase + path);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod(method);
conn.setDoOutput(true);
conn.setRequestProperty("Content-Type", "application/json");
conn.setRequestProperty("X-Nixmsg-Request", "1");
byte[] bytes = json.getBytes(StandardCharsets.UTF_8);
conn.setFixedLengthStreamingMode(bytes.length);
OutputStream os = conn.getOutputStream();
try {
os.write(bytes);
} finally {
os.close();
}
int code = conn.getResponseCode();
InputStream in = code >= 400 ? conn.getErrorStream() : conn.getInputStream();
String raw = in == null ? "{}" : readAll(in);
if (code >= 400) {
throw new IllegalStateException(method + " " + path + " -> " + code + " " + raw);
}
return new JsonParser().parse(raw).getAsJsonObject();
}
void stop() {
if (process != null && isAlive(process)) {
process.destroy();
try {
waitFor(process, 2);
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
if (isAlive(process)) {
process.destroyForcibly();
}
}
process = null;
try {
if (Files.isDirectory(dataDir)) {
List<Path> paths = new ArrayList<Path>();
Files.walk(dataDir).sorted(Comparator.reverseOrder()).forEach(paths::add);
for (Path p : paths) {
try {
Files.deleteIfExists(p);
} catch (IOException ignored) {
}
}
}
} catch (IOException ignored) {
}
}
private static boolean waitFor(Process p, long seconds) throws InterruptedException {
return p.waitFor(seconds, TimeUnit.SECONDS);
}
private static boolean isAlive(Process p) {
try {
p.exitValue();
return false;
} catch (IllegalThreadStateException e) {
return true;
}
}
private static String readAll(InputStream in) throws IOException {
if (in == null) {
return "";
}
ByteArrayOutputStream bos = new ByteArrayOutputStream();
byte[] buf = new byte[4096];
int n;
while ((n = in.read(buf)) >= 0) {
bos.write(buf, 0, n);
}
in.close();
return new String(bos.toByteArray(), StandardCharsets.UTF_8);
}
}
sdk/python/README.md
+40 -4
View File
@@ -1,24 +1,60 @@
# NixMsg Python SDK
包名 `nixmsg`，最低 Python 3.10。同步接口为主，`AsyncClient` 提供 asyncio 包装。
包名 `nixmsg`，最低 Python 3.10。同步接口为主，同包提供 `AsyncClient` asyncio 包装。
## 安装
发布后（阶段 3）：
```bash
pip install nixmsg --index-url https://git.asio.asia/api/packages/nixevol/pypi/simple/
```
本地开发：
```bash
cd sdk/python
python -m venv .venv
# Windows: .venv\Scripts\activate
pip install -e ".[dev]"
```
## 最小示例
```python
from nixmsg import Client, Target, Body
from nixmsg import Body, Client, SendOptions, Target
c = Client()
c.on_session(lambda token: print("session", token))
c.on_message(lambda msg: print("msg", msg.id, msg.body.data))
c.connect("ws://127.0.0.1:7443/mqtt", "device-1", password="secret")
c.send(Target(kind="endpoint", id="device-2"), Body(data="hello"))
c.send(
Target(kind="endpoint", id="device-2"),
Body(data="hello"),
SendOptions(delay_ms=0),
)
c.close()
```
许可证见 `LICENSE`（专有）。
更完整的命令行示例见 `examples/minimal.py`。
## 打包（不发布）
```bash
pip install build
python -m build
# 产物在 dist/，勿上传 PyPI；正式发布由总控在阶段 3 执行
```
## 测试
```bash
# 单元测试（假传输）+ 接入清单（会编译并启动真实 nixmsg）
pytest
```
接入清单覆盖 DEVELOPMENT 第 9 节（跳过仅 JS 的跨域项）。可用环境变量 `NIXMSG_BIN` 指定已编译二进制。
## 许可证
见 `LICENSE`（专有 / Proprietary）。
sdk/python/examples/minimal.py
+31
View File
@@ -0,0 +1,31 @@
"""最小示例：连接、收发、关闭。
用法（需本地已启动 nixmsg，并开放注册或已有端）：
python examples/minimal.py ws://127.0.0.1:PORT/mqtt device-1 password12 peer-id
"""
from __future__ import annotations
import sys
import time
from nixmsg import Body, Client, SendOptions, Target
def main() -> None:
if len(sys.argv) < 5:
print(__doc__)
raise SystemExit(2)
url, eid, password, peer = sys.argv[1:5]
c = Client()
c.on_session(lambda token: print("session", token[:16] + "..."))
c.on_message(lambda msg: print("msg", msg.from_id, msg.id, msg.body.data))
c.on_connection(lambda ev: print("conn", ev.state.value, ev.reason))
c.connect(url, eid, password=password)
c.send(Target(kind="endpoint", id=peer), Body(data="hello from python"), SendOptions(delay_ms=0))
time.sleep(2)
c.close()
if __name__ == "__main__":
main()
sdk/python/src/nixmsg/client.py
+71 -29
View File
@@ -11,6 +11,7 @@ import urllib.request
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Optional
from queue import SimpleQueue
from .errors import ClosedError, NixMsgError, NotConnectedError
from .protocol import dumps, down_topic, loads, normalize_mqtt_ws_url, register_url_from_connect, up_topic
@@ -133,6 +134,9 @@ class Client:
self._worker: Optional[threading.Thread] = None
self._wake = threading.Event()
self._want_connected = False
self._down_q: SimpleQueue = SimpleQueue()
self._down_thread = threading.Thread(target=self._down_loop, name="nixmsg-down", daemon=True)
self._down_thread.start()
self._transport.set_handlers(self._on_transport_connected, self._on_transport_disconnected, self._on_down)
@@ -218,6 +222,10 @@ class Client:
self._transport.disconnect()
except Exception:
pass
try:
self._down_q.put(None)
except Exception:
pass
self._wake.set()
def logout(self) -> None:
@@ -632,40 +640,74 @@ class Client:
self._wake.set()
def _on_down(self, payload: bytes) -> None:
# resp 必须立即完成 pending（含 auto_ack 等待），不能进 down 队列，否则自死锁。
try:
frame = loads(payload)
except Exception:
self._down_q.put(payload)
return
if frame.get("type") == "resp":
self._dispatch_resp(frame)
return
if threading.current_thread() is self._down_thread:
self._dispatch_down_body(frame)
return
self._down_q.put(payload)
def _down_loop(self) -> None:
while True:
payload = self._down_q.get()
if payload is None:
return
try:
frame = loads(payload)
except Exception:
continue
try:
self._dispatch_down_body(frame)
except Exception:
log.exception("处理下行帧失败")
def _dispatch_resp(self, frame: dict[str, Any]) -> None:
rid = str(frame.get("rid", ""))
with self._lock:
pending = self._pending.pop(rid, None)
if not pending:
return
pending.response = frame
if pending.is_send:
with self._lock:
self._inflight_sends = max(0, self._inflight_sends - 1)
err = (frame.get("error") or {}) if not frame.get("ok") else {}
if not frame.get("ok") and str(err.get("code")) == "rate_limited":
with self._lock:
pending.rid = ""
pending.response = None
pending.error = None
pending.event.clear()
self._wake.set()
return
with self._lock:
self._send_queue = [it for it in self._send_queue if it.pending is not pending]
if not frame.get("ok"):
pending.error = NixMsgError(str(err.get("code", "bad_request")), str(err.get("message", "")))
pending.event.set()
self._wake.set()
else:
pending.event.set()
def _dispatch_down(self, payload: bytes) -> None:
try:
frame = loads(payload)
except Exception:
return
ftype = frame.get("type")
if ftype == "resp":
rid = str(frame.get("rid", ""))
with self._lock:
pending = self._pending.pop(rid, None)
if pending:
pending.response = frame
if pending.is_send:
with self._lock:
self._inflight_sends = max(0, self._inflight_sends - 1)
# rate_limited 重交：清状态后不 set event
err = (frame.get("error") or {}) if not frame.get("ok") else {}
if not frame.get("ok") and str(err.get("code")) == "rate_limited":
with self._lock:
pending.rid = ""
pending.response = None
pending.error = None
pending.event.clear()
self._wake.set()
return
# 从发送队列移除
with self._lock:
self._send_queue = [it for it in self._send_queue if it.pending is not pending]
if not frame.get("ok"):
pending.error = NixMsgError(str(err.get("code", "bad_request")), str(err.get("message", "")))
pending.event.set()
self._wake.set()
else:
pending.event.set()
if frame.get("type") == "resp":
self._dispatch_resp(frame)
return
self._dispatch_down_body(frame)
def _dispatch_down_body(self, frame: dict[str, Any]) -> None:
ftype = frame.get("type")
if ftype == "msg":
self._handle_msg(frame)
return
sdk/python/src/nixmsg/transport.py
+48 -44
View File
@@ -8,7 +8,7 @@ from dataclasses import dataclass, field
from typing import Any, Callable, Optional, Protocol
from urllib.parse import urlparse
from paho.mqtt.client import CallbackAPIVersion, Client as PahoClient, MQTT_ERR_SUCCESS
from paho.mqtt.client import CallbackAPIVersion, Client as PahoClient, MQTT_ERR_SUCCESS, MQTTv5
from paho.mqtt.enums import MQTTErrorCode
from paho.mqtt.reasoncodes import ReasonCode
@@ -186,6 +186,8 @@ class PahoTransport:
self._on_down: Optional[DownHandler] = None
self._down_topic = ""
self._loop_started = False
self._sub_event = threading.Event()
self._sub_mid: Optional[int] = None
def set_handlers(
self,
@@ -199,44 +201,39 @@ class PahoTransport:
def connect(self, params: ConnectParams) -> None:
self.disconnect()
url = params.url
u = urlparse(url if "://" in url else "ws://" + url)
use_tcp = params.use_tcp or u.scheme in ("mqtt", "mqtts")
# WebSocket 必须显式 transport=websockets；裸 TCP 走默认。
client = PahoClient(
callback_api_version=CallbackAPIVersion.VERSION2,
client_id=params.client_id,
protocol=PahoClient.MQTTv5,
protocol=MQTTv5,
transport="tcp" if use_tcp else "websockets",
)
client.username_pw_set(params.username, params.password)
client.on_connect = self._on_connect
client.on_disconnect = self._on_disconnect
client.on_message = self._on_message
client.on_subscribe = self._on_subscribe
self._client = client
url = params.url
u = urlparse(url if "://" in url else "ws://" + url)
use_tcp = params.use_tcp or u.scheme in ("mqtt", "mqtts")
host = u.hostname or "localhost"
port = u.port or (8883 if u.scheme in ("wss", "mqtts") else 443 if u.scheme == "wss" else 80)
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
if use_tcp:
if not u.port:
port = 8883 if u.scheme == "mqtts" else 1883
tls = u.scheme == "mqtts"
if tls:
if u.scheme == "mqtts":
client.tls_set()
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
else:
path = u.path or "/mqtt"
if not path.endswith("/mqtt"):
@@ -246,30 +243,28 @@ class PahoTransport:
if u.scheme == "wss":
client.tls_set()
client.ws_set_options(path=path, headers={"Sec-WebSocket-Protocol": "mqtt"})
props = None
try:
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = params.session_expiry
except Exception:
props = None
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
client.connect(
host,
port,
keepalive=params.keep_alive,
clean_start=params.clean_start,
properties=props,
)
client.loop_start()
self._loop_started = True
# 等待连接结果由回调驱动；超时由 Client 层处理
def subscribe(self, topic: str) -> None:
self._down_topic = topic
if self._client:
self._client.subscribe(topic, qos=1)
if not self._client:
return
self._sub_event.clear()
result, mid = self._client.subscribe(topic, qos=1)
if result != MQTT_ERR_SUCCESS:
raise RuntimeError(f"subscribe failed: {result}")
self._sub_mid = mid
if not self._sub_event.wait(10):
raise RuntimeError("subscribe timeout")
def publish(self, topic: str, payload: bytes) -> None:
if not self._client:
@@ -296,13 +291,18 @@ class PahoTransport:
def _on_connect(self, client, userdata, flags, reason_code, properties) -> None:
code = _reason_to_int(reason_code)
if code == 0:
# 不在 loop 线程里同步做 subscribe+等待，否则会卡死 SUBACK
if self._on_connected:
self._on_connected()
threading.Thread(target=self._on_connected, name="nixmsg-on-connected", daemon=True).start()
return
stop, reason = _classify_connack(code)
stop, reason = _classify_connack(code if code is not None else -1)
if self._on_disconnected:
self._on_disconnected(reason, stop)
def _on_subscribe(self, client, userdata, mid, reason_codes, properties) -> None:
if self._sub_mid is None or mid == self._sub_mid:
self._sub_event.set()
def _on_disconnect(self, client, userdata, flags, reason_code, properties) -> None:
code = _reason_to_int(reason_code)
if code in (0, None):
@@ -329,9 +329,13 @@ def _reason_to_int(reason_code) -> Optional[int]:
if isinstance(reason_code, int):
return reason_code
if isinstance(reason_code, ReasonCode):
return int(reason_code)
return int(reason_code.value)
if isinstance(reason_code, MQTTErrorCode):
return int(reason_code)
# paho 偶发其它包装
val = getattr(reason_code, "value", None)
if isinstance(val, int):
return val
try:
return int(reason_code)
except Exception:
sdk/python/tests/__init__.py
+1
View File
@@ -0,0 +1 @@
# tests package
sdk/python/tests/harness.py
+185
View File
@@ -0,0 +1,185 @@
"""真实 nixmsg 进程启动器：临时目录、127.0.0.1:0、admin init、开注册。"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Any, Optional
from urllib.parse import urljoin
def _repo_root() -> Path:
here = Path(__file__).resolve()
for p in [here] + list(here.parents):
if (p / "go.mod").is_file() and (p / "cmd" / "nixmsg").is_dir():
return p
raise RuntimeError("找不到仓库根（go.mod）")
def ensure_binary() -> Path:
env = os.environ.get("NIXMSG_BIN")
if env:
p = Path(env)
if p.is_file():
return p
root = _repo_root()
cache = Path(tempfile.gettempdir()) / "nixmsg-s2-python-bin"
cache.mkdir(parents=True, exist_ok=True)
name = "nixmsg.exe" if os.name == "nt" else "nixmsg"
out = cache / name
# 若已有且较新则复用；否则编译
need = True
if out.is_file():
need = False
if need or os.environ.get("NIXMSG_REBUILD") == "1":
cmd = ["go", "build", "-o", str(out), "./cmd/nixmsg"]
envp = os.environ.copy()
envp["CGO_ENABLED"] = "0"
r = subprocess.run(cmd, cwd=str(root), env=envp, capture_output=True, text=True)
if r.returncode != 0:
raise RuntimeError(f"go build 失败:\n{r.stdout}\n{r.stderr}")
return out
@dataclass
class AdminHTTP:
base: str
opener: urllib.request.OpenerDirector
def request(self, method: str, path: str, body: Optional[dict] = None) -> tuple[int, dict[str, Any]]:
data = None
headers = {"Accept": "application/json"}
if body is not None:
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
if method.upper() in ("POST", "PUT", "PATCH", "DELETE"):
headers["X-Nixmsg-Request"] = "1"
req = urllib.request.Request(urljoin(self.base + "/", path.lstrip("/")), data=data, headers=headers, method=method)
try:
with self.opener.open(req, timeout=30) as resp:
raw = resp.read()
code = resp.getcode()
except urllib.error.HTTPError as e:
raw = e.read()
code = e.code
if not raw:
return code, {}
return code, json.loads(raw.decode("utf-8"))
class NixMsgServer:
def __init__(self) -> None:
self.bin = ensure_binary()
self.data_dir = Path(tempfile.mkdtemp(prefix="nixmsg-s2-py-"))
self.config_path = self.data_dir / "config.yaml"
data_slash = self.data_dir.as_posix()
self.config_path.write_text(
f'listen: "127.0.0.1:0"\ndata_dir: "{data_slash}"\n',
encoding="utf-8",
)
self.admin_password = self._admin_init()
self.proc: Optional[subprocess.Popen] = None
self.addr = ""
self.http_base = ""
self.ws_url = ""
self.reg_code = "s2py-reg-code"
def _admin_init(self) -> str:
env = os.environ.copy()
env["NIXMSG_CONFIG"] = str(self.config_path)
r = subprocess.run(
[str(self.bin), "admin", "init"],
env=env,
capture_output=True,
text=True,
)
if r.returncode != 0:
raise RuntimeError(f"admin init 失败: {r.stdout}\n{r.stderr}")
text = (r.stdout or "") + "\n" + (r.stderr or "")
for line in text.splitlines():
line = line.strip()
lower = line.lower()
if lower.startswith("admin password:"):
return line.split(":", 1)[1].strip()
if lower.startswith("password:"):
return line.split(":", 1)[1].strip()
raise RuntimeError(f"admin init 未解析到密码:\n{text}")
def start(self) -> None:
env = os.environ.copy()
env["NIXMSG_CONFIG"] = str(self.config_path)
self.proc = subprocess.Popen(
[str(self.bin), "serve"],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
addr_file = self.data_dir / "listen.addr"
deadline = time.time() + 20
while time.time() < deadline:
if addr_file.is_file():
addr = addr_file.read_text(encoding="utf-8").strip()
if addr:
self.addr = addr
self.http_base = f"http://{addr}"
self.ws_url = f"ws://{addr}/mqtt"
break
if self.proc.poll() is not None:
raise RuntimeError(f"serve 提前退出 code={self.proc.returncode}")
time.sleep(0.05)
else:
self.stop()
raise RuntimeError("等待 listen.addr 超时")
self._enable_registration(self.reg_code)
def admin(self) -> AdminHTTP:
jar = CookieJar()
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
admin = AdminHTTP(self.http_base, opener)
code, body = admin.request(
"POST",
"/api/admin/login",
{"username": "admin", "password": self.admin_password},
)
if code != 200 or not body.get("ok"):
raise RuntimeError(f"admin login 失败: {code} {body}")
return admin
def _enable_registration(self, code: str, enabled: bool = True) -> None:
admin = self.admin()
status, body = admin.request(
"PUT",
"/api/admin/registration",
{"enabled": enabled, "code": code},
)
if status != 200 or not body.get("ok"):
raise RuntimeError(f"开启注册失败: {status} {body}")
def set_registration(self, *, enabled: bool, code: Optional[str] = None) -> None:
admin = self.admin()
payload: dict[str, Any] = {"enabled": enabled}
if code is not None:
payload["code"] = code
status, body = admin.request("PUT", "/api/admin/registration", payload)
if status != 200 or not body.get("ok"):
raise RuntimeError(f"改注册设置失败: {status} {body}")
def stop(self) -> None:
if self.proc and self.proc.poll() is None:
self.proc.kill()
try:
self.proc.wait(timeout=5)
except Exception:
pass
self.proc = None
if self.data_dir.exists():
shutil.rmtree(self.data_dir, ignore_errors=True)
sdk/python/tests/test_checklist.py
+498
View File
@@ -0,0 +1,498 @@
"""DEVELOPMENT 第 9 节接入清单（对真实 nixmsg；跳过仅 JS 跨域）。"""
from __future__ import annotations
import threading
import time
import unittest
from typing import Optional
from nixmsg import (
Body,
Client,
ConnectionState,
NixMsgError,
Receipt,
RegisterOptions,
SendOptions,
Target,
)
from nixmsg.uuid7 import new_uuid7
from .harness import NixMsgServer
IMMEDIATE = SendOptions(delay_ms=0)
def wait_until(pred, timeout: float = 15.0, interval: float = 0.05) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if pred():
return True
time.sleep(interval)
return False
class MessageBox:
def __init__(self) -> None:
self.items: list = []
self.lock = threading.Lock()
self.event = threading.Event()
def on_message(self, msg) -> None:
with self.lock:
self.items.append(msg)
self.event.set()
def wait_n(self, n: int, timeout: float = 15.0):
deadline = time.time() + timeout
while time.time() < deadline:
with self.lock:
if len(self.items) >= n:
return list(self.items)
self.event.wait(0.1)
self.event.clear()
with self.lock:
return list(self.items)
def clear(self) -> None:
with self.lock:
self.items.clear()
self.event.clear()
class ReceiptBox:
def __init__(self) -> None:
self.items: list[Receipt] = []
self.lock = threading.Lock()
self.event = threading.Event()
def on_receipt(self, r: Receipt) -> None:
with self.lock:
self.items.append(r)
self.event.set()
def wait_state(self, state: str, timeout: float = 15.0) -> Optional[Receipt]:
deadline = time.time() + timeout
while time.time() < deadline:
with self.lock:
for r in self.items:
if r.state == state:
return r
self.event.wait(0.1)
self.event.clear()
return None
class ChecklistIT(unittest.TestCase):
srv: NixMsgServer
seq = 0
@classmethod
def setUpClass(cls) -> None:
cls.srv = NixMsgServer()
cls.srv.start()
@classmethod
def tearDownClass(cls) -> None:
cls.srv.stop()
def _uid(self, prefix: str) -> str:
ChecklistIT.seq += 1
return f"{prefix}{ChecklistIT.seq:04d}"
def _register(self, eid: str, password: str = "password12", name: str = "", code: Optional[str] = None):
return Client.register(
self.srv.ws_url,
code if code is not None else self.srv.reg_code,
RegisterOptions(id=eid, login_password=password, name=name or eid),
)
def _connect(self, eid: str, password: str = "password12", **kwargs) -> Client:
c = Client(**kwargs)
c.connect(self.srv.ws_url, eid, password=password, wait=True)
self.assertEqual(c.state, ConnectionState.ONLINE)
return c
def test_01_handshake(self) -> None:
eid = self._uid("hs")
self._register(eid)
tokens: list[str] = []
c = Client()
c.on_session(lambda t: tokens.append(t))
c.connect(self.srv.ws_url, eid, password="password12")
self.assertEqual(c.state, ConnectionState.ONLINE)
self.assertTrue(c.limits.server_time_ms > 0)
self.assertGreaterEqual(c.limits.max_body_bytes, 256 * 1024)
self.assertTrue(tokens and tokens[0].startswith("nst_"))
c.close()
def test_02_dm_callback_once(self) -> None:
a, b = self._uid("a2"), self._uid("b2")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
ca.send(Target("endpoint", b), Body(data="hello-once"), SendOptions(delay_ms=0, message_id=mid))
got = box.wait_n(1, 10)
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
self.assertEqual(got[0].body.data, "hello-once")
time.sleep(0.5)
self.assertEqual(len(box.wait_n(1, 0.2)), 1)
ca.close()
cb.close()
def test_03_send_while_disconnected_no_dup(self) -> None:
a, b = self._uid("a3"), self._uid("b3")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
# 断开发送方传输，触发重连；期间入队发送
ca._transport.disconnect()
self.assertTrue(wait_until(lambda: ca.state == ConnectionState.RECONNECTING, 5))
err: list[BaseException] = []
result: list = []
def do_send() -> None:
try:
result.append(
ca.send(
Target("endpoint", b),
Body(data="queued"),
SendOptions(delay_ms=0, message_id=mid),
)
)
except BaseException as e:
err.append(e)
th = threading.Thread(target=do_send, daemon=True)
th.start()
th.join(timeout=60)
self.assertFalse(err, err)
self.assertTrue(result)
self.assertEqual(result[0].id, mid)
self.assertTrue(wait_until(lambda: ca.state == ConnectionState.ONLINE, 30))
got = box.wait_n(1, 15)
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
time.sleep(0.8)
self.assertEqual(len(box.items), 1)
ca.close()
cb.close()
def test_04_same_message_id_and_dedup_unit_covered(self) -> None:
"""同消息号重交：断线入队后送达一次。ack 丢失重推依赖 FakeTransport 单测（真实 broker 无法选择性丢 ack）。"""
a, b = self._uid("a4"), self._uid("b4")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
r1 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
self.assertEqual(r1.id, mid)
got = box.wait_n(1, 10)
self.assertEqual(len(got), 1)
# 同号同内容再发：服务器防重，回调仍只有一次
r2 = ca.send(Target("endpoint", b), Body(data="idem"), SendOptions(delay_ms=0, message_id=mid))
self.assertEqual(r2.id, mid)
time.sleep(0.8)
self.assertEqual(len(box.items), 1)
ca.close()
cb.close()
def test_05_recall_within_delay(self) -> None:
a, b = self._uid("a5"), self._uid("b5")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
revoked: list = []
cb.on_message(box.on_message)
cb.on_revoked(lambda e: revoked.append(e))
mid = new_uuid7()
r = ca.send(
Target("endpoint", b),
Body(data="will-recall"),
SendOptions(delay_ms=10_000, message_id=mid),
)
self.assertEqual(r.state, "scheduled")
ca.recall(mid)
time.sleep(1.2)
self.assertEqual(box.items, [])
self.assertEqual(revoked, [])
ca.close()
cb.close()
def test_06_scheduled_about_2s(self) -> None:
a, b = self._uid("a6"), self._uid("b6")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
box = MessageBox()
cb.on_message(box.on_message)
mid = new_uuid7()
t0 = time.monotonic()
ca.send(Target("endpoint", b), Body(data="later"), SendOptions(delay_ms=2000, message_id=mid))
got = box.wait_n(1, 12)
elapsed = time.monotonic() - t0
self.assertEqual(len(got), 1)
self.assertEqual(got[0].id, mid)
self.assertGreaterEqual(elapsed, 1.5)
self.assertLess(elapsed, 8.0)
ca.close()
cb.close()
def test_07_offline_keep(self) -> None:
a, b_ok, b_miss = self._uid("a7"), self._uid("bok"), self._uid("bms")
self._register(a)
self._register(b_ok)
self._register(b_miss)
ca = self._connect(a)
# 对方晚约 1 秒上线能收到
mid1 = new_uuid7()
ca.send(
Target("endpoint", b_ok),
Body(data="keep-ok"),
SendOptions(delay_ms=0, keep=True, ttl_seconds=86400, message_id=mid1),
)
time.sleep(1.0)
cb1 = self._connect(b_ok)
box1 = MessageBox()
cb1.on_message(box1.on_message)
got1 = box1.wait_n(1, 10)
self.assertEqual(len(got1), 1)
self.assertEqual(got1[0].id, mid1)
cb1.close()
# 保留 1 秒且 3 秒后才上线则收不到，发送方收到过期回执
receipts = ReceiptBox()
ca.on_receipt(receipts.on_receipt)
mid2 = new_uuid7()
ca.send(
Target("endpoint", b_miss),
Body(data="keep-expire"),
SendOptions(delay_ms=0, keep=True, ttl_seconds=1, message_id=mid2),
)
time.sleep(3.5)
cb2 = self._connect(b_miss)
box2 = MessageBox()
cb2.on_message(box2.on_message)
time.sleep(1.5)
self.assertEqual(box2.items, [])
exp = receipts.wait_state("expired", 15)
if exp is None:
# 回执可能略慢：用 status 核对投递已过期
st = ca.status(mid2)
items = (st.get("data") or st).get("items") if isinstance(st.get("data") or st, dict) else None
# status 顶层即 data
data = st if "items" in st else (st.get("data") or {})
items = data.get("items") or data.get("deliveries") or []
states = [str(i.get("state", "")) for i in items] if isinstance(items, list) else []
self.assertTrue(
"expired" in states or any(r.state == "expired" for r in receipts.items),
f"want expired receipt/status, receipts={[r.state for r in receipts.items]} status={st}",
)
else:
self.assertEqual(exp.id, mid2)
ca.close()
cb2.close()
def test_08_group_sender_no_echo(self) -> None:
a, b, c = self._uid("a8"), self._uid("b8"), self._uid("c8")
self._register(a)
self._register(b)
self._register(c)
ca, cb, cc = self._connect(a), self._connect(b), self._connect(c)
gid = f"g_{a}"
ca.group_create("G", [{"id": b}, {"id": c}], group_id=gid)
time.sleep(0.4)
box_a, box_b, box_c = MessageBox(), MessageBox(), MessageBox()
ca.on_message(box_a.on_message)
cb.on_message(box_b.on_message)
cc.on_message(box_c.on_message)
mid = new_uuid7()
ca.send(Target("group", gid), Body(data="hi-g"), SendOptions(delay_ms=0, message_id=mid))
gb = box_b.wait_n(1, 10)
gc = box_c.wait_n(1, 10)
self.assertEqual(len(gb), 1)
self.assertEqual(len(gc), 1)
self.assertEqual(gb[0].id, mid)
self.assertEqual(gc[0].id, mid)
time.sleep(0.8)
self.assertEqual(box_a.items, [])
ca.close()
cb.close()
cc.close()
def test_09_talk_password(self) -> None:
a, b = self._uid("a9"), self._uid("b9")
self._register(a)
self._register(b)
ca, cb = self._connect(a), self._connect(b)
cb.set_talk_password("talk99")
# 拒绝
with self.assertRaises(NixMsgError) as cm:
ca.send(Target("endpoint", b), Body(data="no"), IMMEDIATE)
self.assertIn(cm.exception.code, ("talk_password_required", "talk_password_invalid"))
# 解锁
ca.unlock(b, "talk99")
mid = new_uuid7()
ca.send(Target("endpoint", b), Body(data="ok"), SendOptions(delay_ms=0, message_id=mid))
box = MessageBox()
cb.on_message(box.on_message)
self.assertEqual(len(box.wait_n(1, 10)), 1)
# 改密后失效
cb.set_talk_password("talk00")
with self.assertRaises(NixMsgError) as cm2:
ca.send(Target("endpoint", b), Body(data="fail"), IMMEDIATE)
self.assertIn(cm2.exception.code, ("talk_password_required", "talk_password_invalid"))
# 对方先发则可以回复
ca.set_talk_password("alicepw")
box2 = MessageBox()
ca.on_message(box2.on_message)
cb.send(
Target("endpoint", a),
Body(data="first"),
SendOptions(delay_ms=0, talk_password="alicepw", message_id=new_uuid7()),
)
self.assertEqual(len(box2.wait_n(1, 10)), 1)
# a 可回 b（b 曾主动发过）
mid3 = new_uuid7()
box3 = MessageBox()
cb.on_message(box3.on_message)
ca.send(Target("endpoint", b), Body(data="reply"), SendOptions(delay_ms=0, message_id=mid3))
self.assertEqual(len(box3.wait_n(1, 10)), 1)
ca.close()
cb.close()
def test_10_kick_no_reconnect(self) -> None:
eid = self._uid("k10")
self._register(eid)
c1 = self._connect(eid)
states: list[ConnectionState] = []
c1.on_connection(lambda e: states.append(e.state))
c2 = self._connect(eid)
self.assertTrue(wait_until(lambda: c1.state == ConnectionState.KICKED, 15))
time.sleep(2.5)
self.assertEqual(c1.state, ConnectionState.KICKED)
self.assertNotEqual(c1.state, ConnectionState.ONLINE)
self.assertEqual(c2.state, ConnectionState.ONLINE)
c1.close()
c2.close()
def test_11_body_too_large_local(self) -> None:
eid = self._uid("big")
self._register(eid)
c = self._connect(eid)
big = "x" * (256 * 1024 + 1)
with self.assertRaises(NixMsgError) as cm:
c.send(Target("endpoint", eid), Body(data=big), IMMEDIATE)
self.assertEqual(cm.exception.code, "body_too_large")
c.close()
def test_12_registration_toggle(self) -> None:
code = self.srv.reg_code
# 关闭时失败
self.srv.set_registration(enabled=False)
with self.assertRaises(NixMsgError) as cm:
Client.register(self.srv.ws_url, code, RegisterOptions(id=self._uid("r12a"), login_password="password12"))
self.assertEqual(cm.exception.code, "registration_closed")
# 错码
self.srv.set_registration(enabled=True, code=code)
with self.assertRaises(NixMsgError) as cm2:
Client.register(
self.srv.ws_url,
"wrong-code-xx",
RegisterOptions(id=self._uid("r12b"), login_password="password12"),
)
self.assertEqual(cm2.exception.code, "registration_code_invalid")
# 成功后能登录
eid = self._uid("r12c")
Client.register(self.srv.ws_url, code, RegisterOptions(id=eid, login_password="password12"))
c = self._connect(eid)
c.close()
# 换码后旧码失败、已注册照常登录
new_code = "s2py-new-code1"
self.srv.set_registration(enabled=True, code=new_code)
with self.assertRaises(NixMsgError):
Client.register(
self.srv.ws_url,
code,
RegisterOptions(id=self._uid("r12d"), login_password="password12"),
)
c2 = self._connect(eid)
c2.close()
# 恢复默认码供后续用例
self.srv.set_registration(enabled=True, code=code)
self.srv.reg_code = code
def test_13_change_login_password(self) -> None:
eid = self._uid("pw13")
self._register(eid, password="password12")
c = self._connect(eid, password="password12")
c.change_login_password("password12", "password99")
c.close()
# 新密码成功
c2 = self._connect(eid, password="password99")
c2.close()
# 旧密码失败且不再重连
c3 = Client()
states: list[ConnectionState] = []
c3.on_connection(lambda e: states.append(e.state))
with self.assertRaises(NixMsgError) as cm:
c3.connect(self.srv.ws_url, eid, password="password12", wait=True)
self.assertIn(cm.exception.code, ("bad_credentials", "auth_failed"))
time.sleep(2.0)
self.assertEqual(c3.state, ConnectionState.AUTH_FAILED)
c3.close()
def test_15_session_token(self) -> None:
eid = self._uid("tok")
self._register(eid)
tokens: list[str] = []
c = Client()
c.on_session(lambda t: tokens.append(t))
c.connect(self.srv.ws_url, eid, password="password12")
self.assertTrue(tokens)
token = tokens[0]
c.close()
# 令牌重连
c2 = Client()
c2.connect(self.srv.ws_url, eid, session_token=token)
self.assertEqual(c2.state, ConnectionState.ONLINE)
c2.close()
# 另一处密码登录使旧令牌失效
c3 = self._connect(eid, password="password12")
new_token = c3.session_token
self.assertTrue(new_token and new_token != token)
c3.close()
c4 = Client()
with self.assertRaises(NixMsgError) as cm:
c4.connect(self.srv.ws_url, eid, session_token=token, wait=True)
self.assertEqual(cm.exception.code, "session_invalid")
time.sleep(1.5)
self.assertEqual(c4.state, ConnectionState.AUTH_FAILED)
c4.close()
# logout 后令牌失效
c5 = self._connect(eid, password="password12")
tok5 = c5.session_token
assert tok5
c5.logout()
time.sleep(0.3)
c6 = Client()
with self.assertRaises(NixMsgError) as cm2:
c6.connect(self.srv.ws_url, eid, session_token=tok5, wait=True)
self.assertEqual(cm2.exception.code, "session_invalid")
c6.close()
if __name__ == "__main__":
unittest.main()
sdk/python/tests/test_client.py
+2 -2
View File
@@ -67,7 +67,7 @@ class FakeTransportTests(unittest.TestCase):
"send_at_ms": 1,
}
tr.inject_down(dumps(msg))
time.sleep(0.1)
time.sleep(0.3)
self.assertEqual(delivered, ["m1"])
# 找 ack 帧
acks = [json.loads(p.decode()) for _, p in tr.publishes if json.loads(p.decode()).get("type") == "ack"]
@@ -75,7 +75,7 @@ class FakeTransportTests(unittest.TestCase):
before = len(tr.publishes)
tr.inject_down(dumps(msg)) # 已确认再到达
time.sleep(0.1)
time.sleep(0.3)
self.assertEqual(delivered, ["m1"]) # 不重复交应用
acks2 = [json.loads(p.decode()) for _, p in tr.publishes[before:] if json.loads(p.decode()).get("type") == "ack"]
self.assertGreaterEqual(len(acks2), 1) # 再 ack