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

fix: 按 K-00 约定修复 Java SDK 断线重交与退避

Browse Source
This commit is contained in:
Nixevol
2026-09-30 16:24:13 +08:00
parent 537cef0e32
commit c6b7b6e798
12 changed files with 1178 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/java/pom.xml
+22
View File
@@ -52,6 +52,18 @@
</dependencies>
<build>
<resources>
<resource>
<directory>src/main/resources</directory>
</resource>
<resource>
<directory>${project.basedir}</directory>
<includes>
<include>LICENSE</include>
</includes>
<targetPath>META-INF</targetPath>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
@@ -70,6 +82,16 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>3.4.2</version>
<configuration>
<archive>
<addMavenDescriptor>true</addMavenDescriptor>
</archive>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<version>3.3.1</version>
</plugin>
</plugins>
</build>
sdk/java/src/main/java/asia/asio/nixmsg/Client.java
+290 -123
View File
@@ -45,6 +45,10 @@ public final class Client {
private static final Logger LOG = Logger.getLogger("nixmsg");
private static final String DELIVERED = "delivered";
private static final String ACKED = "acked";
private static final Runnable CB_POISON = new Runnable() {
@Override
public void run() {}
};
private final Transport transport;
private final boolean autoAck;
@@ -73,19 +77,23 @@ public final class Client {
private boolean useTcp;
private HelloLimits limits = new HelloLimits();
private long clockSkewMs;
private long onlineSinceMs;
private long backoffMs = Types.BACKOFF_INITIAL_MS;
private final Types.ReconnectBackoff backoff = new Types.ReconnectBackoff();
private int inflightSends;
private List<String> watchIds;
private boolean watchAll;
private volatile Throwable handshakeError;
private volatile String authReason = "";
private volatile String lastStopCode = "";
private volatile NixMsgException lastStopErr;
private int sendQueueLimit = Types.SEND_QUEUE_LIMIT;
private final AtomicBoolean connReady = new AtomicBoolean(false);
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 final BlockingQueue<Runnable> cbQueue = new LinkedBlockingQueue<Runnable>();
private final Thread cbWorker;
private Consumer<String> sessionHandler;
private Consumer<IncomingMessage> messageHandler;
@@ -106,13 +114,16 @@ public final class Client {
public Client(Transport transport, boolean autoAck, int maxReceiveBytes, String clientName, long connectTimeoutMs) {
this.transport = transport;
this.autoAck = autoAck;
this.maxReceiveBytes = Math.max(Types.MIN_MAX_RECEIVE, maxReceiveBytes);
this.maxReceiveBytes = maxReceiveBytes;
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();
this.cbWorker = new Thread(this::cbLoop, "nixmsg-cb");
this.cbWorker.setDaemon(true);
this.cbWorker.start();
}
public void onSession(Consumer<String> handler) { this.sessionHandler = handler; }
@@ -128,6 +139,14 @@ public final class Client {
public String getSessionToken() { return sessionToken; }
public long getClockSkewMs() { return clockSkewMs; }
String lastStopCodeForTest() {
return lastStopCode == null ? "" : lastStopCode;
}
void setSendQueueLimitForTest(int n) {
sendQueueLimit = n;
}
/** 同包测试用：断开底层传输以触发重连与发送队列重交。 */
void dropTransportForTest() {
transport.disconnect();
@@ -145,11 +164,14 @@ public final class Client {
if (password == null && sessionToken == null) {
throw new IllegalArgumentException("需要 password 或 sessionToken");
}
if (maxReceiveBytes < Types.MIN_MAX_RECEIVE) {
throw new NixMsgException("bad_request", "max_receive_bytes 不能小于 1024");
}
synchronized (lock) {
if (closed) {
throw new NixMsgException("closed", "已关闭");
}
this.url = useTcp ? url : Protocol.normalizeMqttWsUrl(url);
this.url = Protocol.normalizeMqttWsUrl(url, useTcp);
this.endpointId = endpointId;
this.password = password;
this.sessionToken = sessionToken;
@@ -158,6 +180,9 @@ public final class Client {
this.userClose = false;
this.wantConnected = true;
this.handshakeError = null;
this.lastStopCode = "";
this.lastStopErr = null;
backoff.reset();
connReady.set(false);
setState(ConnectionState.CONNECTING, "");
if (worker == null || !worker.isAlive()) {
@@ -167,14 +192,13 @@ public final class Client {
}
wakeUp();
}
long deadline = System.currentTimeMillis() + connectTimeoutMs + 5000;
long deadline = System.currentTimeMillis() + connectTimeoutMs;
synchronized (connWait) {
while (System.currentTimeMillis() < deadline) {
ConnectionState s = state;
if (s == ConnectionState.ONLINE
|| s == ConnectionState.AUTH_FAILED
|| s == ConnectionState.KICKED
|| s == ConnectionState.OFFLINE
|| handshakeError != null) {
break;
}
@@ -187,10 +211,13 @@ public final class Client {
}
}
if (handshakeError != null) {
if (handshakeError instanceof NixMsgException) {
throw (NixMsgException) handshakeError;
}
if (handshakeError instanceof RuntimeException) {
throw (RuntimeException) handshakeError;
}
throw new NixMsgException("busy", handshakeError.getMessage());
throw new NixMsgException("not_connected", handshakeError.getMessage());
}
if (state != ConnectionState.ONLINE) {
if (state == ConnectionState.AUTH_FAILED) {
@@ -199,7 +226,11 @@ public final class Client {
if (state == ConnectionState.KICKED) {
throw new NixMsgException("taken_over", "会话被接管");
}
throw new NixMsgException("busy", "连接未成功: " + state);
synchronized (lock) {
stopReconnect = true;
wantConnected = false;
}
throw new NixMsgException("not_connected", "连接未成功: " + state);
}
}
@@ -220,21 +251,27 @@ public final class Client {
transport.disconnect();
} catch (Exception ignored) {
}
downQueue.offer(new byte[0]); // 空载荷哨兵：downLoop 见 closed 退出
downQueue.offer(new byte[0]);
cbQueue.offer(CB_POISON);
wakeUp();
}
public CompletableFuture<Void> logout() {
return CompletableFuture.runAsync(() -> {
Throwable reqErr = null;
try {
request(mapOf("type", "self.logout"), true);
} catch (Exception ignored) {
} catch (Exception e) {
reqErr = e;
}
NixMsgException stop = new NixMsgException("logged_out", "已退出登录");
synchronized (lock) {
stopReconnect = true;
wantConnected = false;
sessionToken = null;
failAll(new NixMsgException("auth_failed", "已退出登录"));
lastStopCode = "logged_out";
lastStopErr = stop;
failAll(stop);
}
try {
transport.disconnect();
@@ -242,6 +279,12 @@ public final class Client {
}
setState(ConnectionState.OFFLINE, "");
wakeUp();
if (reqErr != null) {
if (reqErr instanceof RuntimeException) {
throw (RuntimeException) reqErr;
}
throw new NixMsgException("not_connected", reqErr.getMessage());
}
});
}
@@ -261,8 +304,12 @@ public final class Client {
if (closed) {
throw new NixMsgException("closed", "已关闭");
}
if (sendQueue.size() >= Types.SEND_QUEUE_LIMIT) {
throw new NixMsgException("quota_exceeded", "发送队列已满");
if (stopReconnect) {
throw lastStopErr != null ? lastStopErr : new NixMsgException(
lastStopCode.isEmpty() ? "not_connected" : lastStopCode, "已停止重连");
}
if (sendQueue.size() >= sendQueueLimit) {
throw new NixMsgException("queue_full", "发送队列已满");
}
int maxBody = limits.maxBodyBytes > 0 ? limits.maxBodyBytes : Types.DEFAULT_MAX_BODY;
int maxMeta = limits.maxMetaBytes > 0 ? limits.maxMetaBytes : Types.DEFAULT_MAX_META;
@@ -285,8 +332,17 @@ public final class Client {
if (!meta.isEmpty()) {
frame.put("meta", meta);
}
if (options.sendAtMs != null && options.delayMs != null
|| options.sendAt != null && options.delayMs != null
|| options.sendAtDate != null && options.delayMs != null) {
throw new NixMsgException("bad_request", "sendAt 与 delay 互斥");
}
if (options.sendAtMs != null) {
frame.put("send_at_ms", options.sendAtMs);
} else if (options.sendAt != null) {
frame.put("send_at_ms", options.sendAt + clockSkewMs);
} else if (options.sendAtDate != null) {
frame.put("send_at_ms", options.sendAtDate.getTime() + clockSkewMs);
} else if (options.delayMs != null) {
frame.put("delay_ms", options.delayMs);
}
@@ -332,8 +388,7 @@ public final class Client {
public CompletableFuture<RecallResult> recall(String messageId) {
return CompletableFuture.supplyAsync(() -> {
Map<String, Object> resp = request(mapOf("type", "recall", "id", messageId), true);
Map<String, Object> data = asMap(resp.get("data"));
Map<String, Object> data = request(mapOf("type", "recall", "id", messageId), true);
return new RecallResult(str(data.get("result"), ""),
(int) longVal(data.get("recalled"), 0),
(int) longVal(data.get("accepted"), 0),
@@ -389,6 +444,10 @@ public final class Client {
}
public CompletableFuture<Map<String, Object>> updateSelf(String name, Integer defaultDelayMs) {
return updateSelf(name, defaultDelayMs == null ? null : defaultDelayMs.longValue());
}
public CompletableFuture<Map<String, Object>> updateSelf(String name, Long defaultDelayMs) {
return CompletableFuture.supplyAsync(() -> {
Map<String, Object> f = new LinkedHashMap<String, Object>();
f.put("type", "self.update");
@@ -408,11 +467,10 @@ public final class Client {
public CompletableFuture<Map<String, Object>> changeLoginPassword(String oldPassword, String newPassword) {
return CompletableFuture.supplyAsync(() -> {
Map<String, Object> resp = request(mapOf(
Map<String, Object> data = request(mapOf(
"type", "self.login_password",
"old_password", oldPassword,
"new_password", newPassword), true);
Map<String, Object> data = asMap(resp.get("data"));
Object token = data.get("session_token");
if (token != null) {
synchronized (lock) {
@@ -420,7 +478,7 @@ public final class Client {
}
fireSession(String.valueOf(token));
}
return resp;
return data;
});
}
@@ -536,33 +594,42 @@ public final class Client {
}
if (st == ConnectionState.ONLINE) {
pumpSends();
if (onlineSinceMs > 0 && System.currentTimeMillis() - onlineSinceMs >= Types.STABLE_RESET_MS) {
backoffMs = Types.BACKOFF_INITIAL_MS;
}
waitWake(200);
continue;
}
long delay;
synchronized (lock) {
if (!wantConnected || stopReconnect) {
continue;
}
delay = backoff.nextWaitMs();
}
if (delay > 0) {
waitWake(delay);
}
synchronized (lock) {
if (closed && !wantConnected) {
return;
}
if (!wantConnected || stopReconnect) {
continue;
}
}
try {
attemptConnect();
} catch (Exception e) {
LOG.log(Level.FINE, "connect attempt failed", e);
}
long delay;
synchronized (lock) {
if (state == ConnectionState.ONLINE) {
backoff.markOnline();
continue;
}
if (stopReconnect || !wantConnected) {
continue;
}
delay = jitter(backoffMs);
backoffMs = Math.min(Types.BACKOFF_MAX_MS, backoffMs * 2);
setState(ConnectionState.RECONNECTING, "");
}
try {
Thread.sleep(delay);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
backoff.markOffline();
setState(ConnectionState.RECONNECTING, lastStopCode.isEmpty() ? "network" : lastStopCode);
}
}
}
@@ -612,7 +679,7 @@ public final class Client {
} catch (Exception ignored) {
}
if (handshakeError == null && state != ConnectionState.AUTH_FAILED && state != ConnectionState.KICKED) {
handshakeError = new NixMsgException("busy", "连接超时");
handshakeError = new NixMsgException("not_connected", "连接超时");
}
signalConn();
}
@@ -639,7 +706,10 @@ public final class Client {
if (p.error != null) {
throw p.error instanceof RuntimeException ? (RuntimeException) p.error : new NixMsgException("busy", p.error.getMessage());
}
if (p.response == null || !Boolean.TRUE.equals(p.response.get("ok"))) {
if (p.response == null) {
throw new NixMsgException("not_connected", "握手超时");
}
if (!Boolean.TRUE.equals(p.response.get("ok"))) {
Map<String, Object> err = p.response == null ? new LinkedHashMap<String, Object>() : asMap(p.response.get("error"));
throw new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
@@ -659,8 +729,8 @@ public final class Client {
synchronized (lock) {
limits = lim;
clockSkewMs = skew;
onlineSinceMs = System.currentTimeMillis();
handshakeError = null;
backoff.markOnline();
setState(ConnectionState.ONLINE, "");
}
if (lim.sessionToken != null && !lim.sessionToken.isEmpty()) {
@@ -684,7 +754,14 @@ public final class Client {
signalConn();
wakeUp();
} catch (Exception e) {
handshakeError = e;
if (e instanceof NixMsgException) {
NixMsgException ne = (NixMsgException) e;
handshakeError = "busy".equals(ne.getCode())
? new NixMsgException("not_connected", ne.getMessage())
: ne;
} else {
handshakeError = new NixMsgException("not_connected", e.getMessage());
}
try {
transport.disconnect();
} catch (Exception ignored) {
@@ -700,7 +777,9 @@ public final class Client {
if ("taken_over".equals(reason)) {
stopReconnect = true;
wantConnected = false;
failAll(new NixMsgException("taken_over", "会话被接管"));
lastStopCode = "taken_over";
lastStopErr = new NixMsgException("taken_over", "会话被接管");
failAll(lastStopErr);
setState(ConnectionState.KICKED, "taken_over");
signalConn();
return;
@@ -717,6 +796,8 @@ public final class Client {
stopReconnect = true;
wantConnected = false;
NixMsgException err = new NixMsgException(ar, "认证失败");
lastStopCode = ar;
lastStopErr = err;
failAll(err);
handshakeError = err;
setState(ConnectionState.AUTH_FAILED, ar);
@@ -728,6 +809,9 @@ public final class Client {
signalConn();
return;
}
backoff.markOffline();
requeueInflight();
failPending(new NixMsgException("not_connected", "连接已断开"), false);
if (wasOnline || state == ConnectionState.CONNECTING || state == ConnectionState.RECONNECTING) {
setState(ConnectionState.RECONNECTING, reason == null ? "network" : reason);
}
@@ -749,6 +833,10 @@ public final class Client {
dispatchResp(frame);
return;
}
if ("fatal".equals(str(frame.get("type"), ""))) {
handleFatal(str(frame.get("reason"), "protocol"));
return;
}
downQueue.offer(payload);
}
@@ -796,9 +884,19 @@ public final class Client {
Map<String, Object> err = asMap(frame.get("error"));
if ("rate_limited".equals(str(err.get("code"), ""))) {
synchronized (lock) {
pending.remove(rid);
inflightSends = Math.max(0, inflightSends - 1);
p.rid = "";
p.response = null;
p.error = null;
for (SendItem it : sendQueue) {
if (it.pending == p) {
it.rateN++;
long wait = Types.ReconnectBackoff.jitter(Types.ReconnectBackoff.nominalMs(it.rateN));
it.retryAtMs = System.currentTimeMillis() + wait;
break;
}
}
}
wakeUp();
return;
@@ -861,17 +959,23 @@ public final class Client {
return;
}
if ("fatal".equals(type)) {
String reason = str(frame.get("reason"), "protocol");
synchronized (lock) {
stopReconnect = true;
wantConnected = false;
failAll(new NixMsgException("fatal", reason));
}
setState(ConnectionState.AUTH_FAILED, reason);
try {
transport.disconnect();
} catch (Exception ignored) {
}
handleFatal(str(frame.get("reason"), "protocol"));
return;
}
}
private void handleFatal(String reason) {
synchronized (lock) {
stopReconnect = true;
wantConnected = false;
lastStopCode = reason == null || reason.isEmpty() ? "fatal" : reason;
lastStopErr = new NixMsgException(lastStopCode, reason);
failAll(lastStopErr);
}
setState(ConnectionState.AUTH_FAILED, reason);
try {
transport.disconnect();
} catch (Exception ignored) {
}
}
@@ -905,24 +1009,37 @@ public final class Client {
}
if (messageHandler == null) {
if (autoAck) {
synchronized (lock) {
dedupPut(key, ACKED);
}
sendAck(from, mid, true);
}
return;
}
try {
synchronized (cbLock) {
messageHandler.accept(msg);
final IncomingMessage delivered = msg;
final String dedupKey = key;
final String ackFrom = from;
final String ackId = mid;
enqueueCb(new Runnable() {
@Override
public void run() {
try {
messageHandler.accept(delivered);
} catch (Exception e) {
LOG.log(Level.SEVERE, "onMessage 回调错误，等待重推", e);
synchronized (lock) {
dedup.remove(dedupKey);
}
return;
}
if (autoAck) {
synchronized (lock) {
dedupPut(dedupKey, ACKED);
}
sendAck(ackFrom, ackId, true);
}
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onMessage 回调错误，等待重推", e);
synchronized (lock) {
dedup.remove(key);
}
return;
}
if (autoAck) {
sendAck(from, mid, true);
}
});
}
private void handleReceipt(Map<String, Object> frame) {
@@ -964,8 +1081,7 @@ public final class Client {
private void sendAck(String from, String messageId, boolean markAcked) {
try {
Map<String, Object> resp = request(mapOf("type", "ack", "from", from, "id", messageId), true);
Map<String, Object> data = asMap(resp.get("data"));
Map<String, Object> data = request(mapOf("type", "ack", "from", from, "id", messageId), true);
String result = str(data.get("result"), "accepted");
String key = from + "\0" + messageId;
if (!"accepted".equals(result)) {
@@ -998,7 +1114,11 @@ public final class Client {
return;
}
item = null;
long now = System.currentTimeMillis();
for (SendItem it : sendQueue) {
if (it.retryAtMs > now) {
continue;
}
if (it.pending.rid.isEmpty() && it.pending.response == null && it.pending.error == null) {
item = it;
break;
@@ -1023,9 +1143,13 @@ public final class Client {
pending.remove(rid);
item.pending.rid = "";
inflightSends = Math.max(0, inflightSends - 1);
item.pending.error = e;
item.pending.future.complete(null);
sendQueue.remove(item);
if (stopReconnect) {
NixMsgException err = lastStopErr != null ? lastStopErr
: new NixMsgException("not_connected", "已停止重连");
item.pending.error = err;
item.pending.future.complete(null);
sendQueue.remove(item);
}
}
}
}
@@ -1069,7 +1193,7 @@ public final class Client {
Map<String, Object> err = asMap(p.response.get("error"));
throw new NixMsgException(str(err.get("code"), "bad_request"), str(err.get("message"), ""));
}
return p.response;
return asMap(p.response.get("data"));
}
private String nextRid() {
@@ -1092,96 +1216,142 @@ public final class Client {
}
private void failAll(Throwable err) {
for (Pending p : new ArrayList<Pending>(pending.values())) {
failPending(err, true);
}
private void failPending(Throwable err, boolean includeSend) {
for (Map.Entry<String, Pending> e : new ArrayList<Map.Entry<String, Pending>>(pending.entrySet())) {
Pending p = e.getValue();
if (!includeSend && p.isSend) {
continue;
}
p.error = err;
p.future.complete(null);
pending.remove(e.getKey());
}
pending.clear();
for (SendItem it : new ArrayList<SendItem>(sendQueue)) {
it.pending.error = err;
it.pending.future.complete(null);
if (includeSend) {
for (SendItem it : new ArrayList<SendItem>(sendQueue)) {
it.pending.error = err;
it.pending.future.complete(null);
}
sendQueue.clear();
inflightSends = 0;
}
}
private void requeueInflight() {
for (SendItem it : sendQueue) {
if (it.pending.rid != null && !it.pending.rid.isEmpty()) {
pending.remove(it.pending.rid);
it.pending.rid = "";
}
}
sendQueue.clear();
inflightSends = 0;
}
private void setState(ConnectionState st, String reason) {
state = st;
Consumer<ConnectionEvent> h = connectionHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(new ConnectionEvent(st, reason));
final ConnectionEvent ev = new ConnectionEvent(st, reason);
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<ConnectionEvent> h = connectionHandler;
if (h != null) {
h.accept(ev);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onConnection", e);
}
});
}
private void enqueueCb(Runnable r) {
if (r == null) {
return;
}
cbQueue.offer(r);
}
private void cbLoop() {
while (true) {
try {
Runnable r = cbQueue.take();
if (r == CB_POISON) {
return;
}
try {
r.run();
} catch (Exception e) {
LOG.log(Level.SEVERE, "回调错误", e);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
}
private void fireSession(String token) {
Consumer<String> h = sessionHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(token);
final String t = token;
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<String> h = sessionHandler;
if (h != null) {
h.accept(t);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onSession", e);
}
}
});
}
private void fireReceipt(Receipt r) {
Consumer<Receipt> h = receiptHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(r);
final Receipt ev = r;
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<Receipt> h = receiptHandler;
if (h != null) {
h.accept(ev);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onReceipt", e);
}
}
});
}
private void fireRevoked(RevokedEvent ev) {
Consumer<RevokedEvent> h = revokedHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(ev);
final RevokedEvent e = ev;
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<RevokedEvent> h = revokedHandler;
if (h != null) {
h.accept(e);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onRevoked", e);
}
}
});
}
private void firePresence(PresenceEvent ev) {
Consumer<PresenceEvent> h = presenceHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(ev);
final PresenceEvent e = ev;
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<PresenceEvent> h = presenceHandler;
if (h != null) {
h.accept(e);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onPresence", e);
}
}
});
}
private void fireGroup(GroupEvent ev) {
Consumer<GroupEvent> h = groupHandler;
if (h != null) {
try {
synchronized (cbLock) {
h.accept(ev);
final GroupEvent e = ev;
enqueueCb(new Runnable() {
@Override
public void run() {
Consumer<GroupEvent> h = groupHandler;
if (h != null) {
h.accept(e);
}
} catch (Exception e) {
LOG.log(Level.SEVERE, "onGroupEvent", e);
}
}
});
}
private void signalConn() {
@@ -1207,11 +1377,6 @@ public final class Client {
}
}
private static long jitter(long base) {
double j = 1.0 + (Math.random() * 2 - 1) * Types.BACKOFF_JITTER;
return Math.max(0L, (long) (base * j));
}
private static void await(CompletableFuture<Void> f, long timeoutMs) {
try {
f.get(timeoutMs, TimeUnit.MILLISECONDS);
@@ -1285,6 +1450,8 @@ public final class Client {
final String messageId;
final Map<String, Object> frame;
final Pending pending;
int rateN;
long retryAtMs;
SendItem(String messageId, Map<String, Object> frame, Pending pending) {
this.messageId = messageId;
sdk/java/src/main/java/asia/asio/nixmsg/NixBase64.java
+65
View File
@@ -0,0 +1,65 @@
package asia.asio.nixmsg;
import java.nio.charset.StandardCharsets;
/** Android API 24 可用的标准 Base64（带填充），不依赖 java.util.Base64。 */
final class NixBase64 {
private static final byte[] ENC = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/".getBytes(StandardCharsets.US_ASCII);
private static final int[] DEC = new int[256];
static {
for (int i = 0; i < DEC.length; i++) {
DEC[i] = -1;
}
for (int i = 0; i < ENC.length; i++) {
DEC[ENC[i] & 0xff] = i;
}
DEC['='] = 0;
}
private NixBase64() {}
static String encode(byte[] src) {
if (src == null || src.length == 0) {
return "";
}
int len = src.length;
int outLen = 4 * ((len + 2) / 3);
byte[] out = new byte[outLen];
int i = 0;
int o = 0;
while (i < len) {
int b0 = src[i++] & 0xff;
int b1 = i < len ? src[i++] & 0xff : 0;
int b2 = i < len ? src[i++] & 0xff : 0;
int triple = (b0 << 16) | (b1 << 8) | b2;
out[o++] = ENC[(triple >> 18) & 63];
out[o++] = ENC[(triple >> 12) & 63];
out[o++] = ENC[(triple >> 6) & 63];
out[o++] = ENC[triple & 63];
}
int mod = len % 3;
if (mod == 1) {
out[outLen - 1] = '=';
out[outLen - 2] = '=';
} else if (mod == 2) {
out[outLen - 1] = '=';
}
return new String(out, StandardCharsets.US_ASCII);
}
static int decodedLength(String s) {
if (s == null || s.isEmpty()) {
return 0;
}
int pad = 0;
int n = s.length();
if (n >= 1 && s.charAt(n - 1) == '=') {
pad++;
}
if (n >= 2 && s.charAt(n - 2) == '=') {
pad++;
}
return (n / 4) * 3 - pad;
}
}
sdk/java/src/main/java/asia/asio/nixmsg/Protocol.java
+28 -11
View File
@@ -34,25 +34,42 @@ final class Protocol {
}
static String normalizeMqttWsUrl(String url) {
return normalizeMqttWsUrl(url, false);
}
static String normalizeMqttWsUrl(String url, boolean allowTcp) {
String raw = url.trim();
if (!raw.contains("://")) {
raw = "ws://" + raw;
}
URI u = URI.create(raw);
String path = u.getPath() == null ? "" : u.getPath();
if (!path.endsWith("/mqtt")) {
if (path.endsWith("/")) {
path = path + "mqtt";
} else if (path.isEmpty()) {
path = "/mqtt";
} else {
path = path + "/mqtt";
}
String scheme = u.getScheme() == null ? "ws" : u.getScheme().toLowerCase();
if ("http".equals(scheme)) {
scheme = "ws";
} else if ("https".equals(scheme)) {
scheme = "wss";
}
if ("mqtt".equals(scheme) || "mqtts".equals(scheme)) {
if (!allowTcp) {
throw new NixMsgException("bad_request", "裸 TCP 需显式 useTcp");
}
return reconstruct(u, scheme, u.getPath() == null ? "" : u.getPath());
}
if (!"ws".equals(scheme) && !"wss".equals(scheme)) {
throw new NixMsgException("bad_request", "不支持的地址协议: " + u.getScheme());
}
String path = u.getPath() == null ? "" : u.getPath();
if (path.isEmpty() || "/".equals(path)) {
path = "/mqtt";
}
return reconstruct(u, scheme, path);
}
private static String reconstruct(URI u, String scheme, String path) {
try {
return new URI(u.getScheme(), u.getUserInfo(), u.getHost(), u.getPort(), path, u.getQuery(), u.getFragment()).toString();
return new URI(scheme, u.getUserInfo(), u.getHost(), u.getPort(), path, u.getQuery(), u.getFragment()).toString();
} catch (Exception e) {
return raw;
return u.toString();
}
}
sdk/java/src/main/java/asia/asio/nixmsg/Transport.java
+9 -197
View File
@@ -9,21 +9,14 @@ 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;
import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5DisconnectReasonCode;
import java.net.URI;
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;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
/** MQTT 传输抽象。 */
public interface Transport {
@@ -46,6 +39,7 @@ public interface Transport {
public final int sessionExpiry;
public final boolean useTcp;
public final long timeoutMs;
public final int keepAlive = 30;
public ConnectParams(String url, String clientId, String username, String password,
boolean cleanStart, int sessionExpiry, boolean useTcp, long timeoutMs) {
@@ -61,164 +55,6 @@ public interface Transport {
}
}
/** 单元测试假传输。 */
final class FakeTransport implements Transport {
final List<Transport.ConnectParams> connects = new ArrayList<Transport.ConnectParams>();
final List<byte[]> publishes = new CopyOnWriteArrayList<byte[]>();
final List<String> subscriptions = new ArrayList<String>();
volatile String nextConnackFail;
Map<String, Object> autoHello;
boolean autoSendOk = true;
final List<Function<Map<String, Object>, Map<String, Object>>> upHandlers =
new CopyOnWriteArrayList<Function<Map<String, Object>, Map<String, Object>>>();
private Runnable onConnected;
private BiConsumer<String, Boolean> onDisconnected;
private Consumer<byte[]> onDown;
private volatile boolean connected;
FakeTransport() {
autoHello = new LinkedHashMap<String, Object>();
autoHello.put("server_time_ms", 1750000000000L);
autoHello.put("server_version", "0.1.0");
autoHello.put("max_body_bytes", 262144);
autoHello.put("max_meta_bytes", 4096);
autoHello.put("max_frame_bytes", 786432);
autoHello.put("max_ttl_seconds", 2592000);
autoHello.put("max_schedule_seconds", 31536000);
autoHello.put("ack_timeout_seconds", 300);
autoHello.put("session_token", "nst_test_token");
}
void onUp(Function<Map<String, Object>, Map<String, Object>> h) {
upHandlers.add(h);
}
@Override
public void setHandlers(Runnable onConnected, BiConsumer<String, Boolean> onDisconnected, Consumer<byte[]> onDown) {
this.onConnected = onConnected;
this.onDisconnected = onDisconnected;
this.onDown = onDown;
}
@Override
public void connect(ConnectParams params) {
connects.add(params);
if (nextConnackFail != null) {
String fail = nextConnackFail;
nextConnackFail = null;
boolean stop = "session_invalid".equals(fail) || "bad_credentials".equals(fail) || "banned".equals(fail);
if (onDisconnected != null) {
onDisconnected.accept(fail, stop);
}
return;
}
connected = true;
if (onConnected != null) {
onConnected.run();
}
}
@Override
public void subscribe(String topic) {
subscriptions.add(topic);
}
@Override
public void publish(String topic, byte[] payload) {
publishes.add(payload);
Map<String, Object> frame = Protocol.loads(payload);
for (Function<Map<String, Object>, Map<String, Object>> h : upHandlers) {
Map<String, Object> resp = h.apply(frame);
if (resp != null) {
injectDown(Protocol.dumps(resp));
return;
}
}
String type = str(frame.get("type"));
Object rid = frame.get("rid");
if ("hello".equals(type) && autoHello != null) {
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", autoHello);
injectDown(Protocol.dumps(resp));
return;
}
if ("send".equals(type) && autoSendOk) {
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("id", frame.get("id"));
Object sat = frame.get("send_at_ms");
data.put("send_at_ms", sat == null ? 0 : sat);
data.put("state", "dispatched");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", data);
injectDown(Protocol.dumps(resp));
return;
}
if ("ack".equals(type)) {
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("result", "accepted");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", data);
injectDown(Protocol.dumps(resp));
return;
}
if (type != null && !"hello".equals(type) && !"send".equals(type) && rid != null) {
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", new LinkedHashMap<String, Object>());
injectDown(Protocol.dumps(resp));
}
}
@Override
public void disconnect() {
boolean was = connected;
connected = false;
if (was && onDisconnected != null) {
onDisconnected.accept(null, false);
}
}
void injectDown(byte[] payload) {
if (onDown != null) {
onDown.accept(payload);
}
}
void simulateTakenOver() {
connected = false;
if (onDisconnected != null) {
onDisconnected.accept("taken_over", true);
}
}
void simulateNetworkDrop() {
connected = false;
if (onDisconnected != null) {
onDisconnected.accept("network", false);
}
}
private static String str(Object o) {
return o == null ? null : String.valueOf(o);
}
}
/** HiveMQ MQTT 5 + WebSocket（子协议 mqtt）。 */
final class HiveMqTransport implements Transport {
private Runnable onConnected;
@@ -237,9 +73,7 @@ final class HiveMqTransport implements Transport {
public void connect(final ConnectParams params) {
disconnectQuiet();
URI u = URI.create(params.url.contains("://") ? params.url : "ws://" + params.url);
boolean useTcp = params.useTcp
|| "mqtt".equalsIgnoreCase(u.getScheme())
|| "mqtts".equalsIgnoreCase(u.getScheme());
boolean useTcp = params.useTcp;
String host = u.getHost() == null ? "localhost" : u.getHost();
int port = u.getPort();
if (port < 0) {
@@ -266,6 +100,9 @@ final class HiveMqTransport implements Transport {
if (rc == Mqtt5DisconnectReasonCode.SESSION_TAKEN_OVER) {
reason = "taken_over";
stop = true;
} else if (rc == Mqtt5DisconnectReasonCode.SERVER_SHUTTING_DOWN) {
reason = "network";
stop = false;
}
break;
}
@@ -284,10 +121,7 @@ final class HiveMqTransport implements Transport {
});
if (!useTcp) {
String path = u.getPath() == null || u.getPath().isEmpty() ? "/mqtt" : u.getPath();
if (!path.endsWith("/mqtt")) {
path = path.endsWith("/") ? path + "mqtt" : path + "/mqtt";
}
String path = u.getPath() == null || u.getPath().isEmpty() || "/".equals(u.getPath()) ? "/mqtt" : u.getPath();
String serverPath = path.startsWith("/") ? path.substring(1) : path;
b5 = b5.webSocketConfig()
.serverPath(serverPath)
@@ -313,6 +147,7 @@ final class HiveMqTransport implements Transport {
Mqtt5ConnAck ack = c.connectWith()
.cleanStart(params.cleanStart)
.sessionExpiryInterval(params.sessionExpiry)
.keepAlive(30)
.simpleAuth()
.username(params.username)
.password(params.password.getBytes(StandardCharsets.UTF_8))
@@ -340,18 +175,8 @@ final class HiveMqTransport implements Transport {
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;
}
reason = "network";
stop = false;
}
BiConsumer<String, Boolean> h = onDisconnected;
if (h != null) {
@@ -418,19 +243,6 @@ final class HiveMqTransport implements Transport {
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/Types.java
+107 -3
View File
@@ -1,7 +1,6 @@
package asia.asio.nixmsg;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
@@ -61,12 +60,12 @@ public final class Types {
}
public static Body ofBytes(byte[] raw) {
return new Body("base64", Base64.getEncoder().encodeToString(raw), "application/octet-stream");
return new Body("base64", NixBase64.encode(raw), "application/octet-stream");
}
public int decodedSize() {
if ("base64".equals(enc)) {
return Base64.getDecoder().decode(data).length;
return NixBase64.decodedLength(data);
}
return data.getBytes(StandardCharsets.UTF_8).length;
}
@@ -89,6 +88,9 @@ public final class Types {
public static final class SendOptions {
public Long sendAtMs;
/** 本机时间 epoch 毫秒；入队时加时钟偏差写成 send_at_ms */
public Long sendAt;
public java.util.Date sendAtDate;
public Long delayMs;
public boolean keep;
public Long ttlSeconds;
@@ -237,4 +239,106 @@ public final class Types {
this.reason = reason == null ? "" : reason;
}
}
public static final class ReconnectBackoff {
private int n;
private boolean skipFirst = true;
private boolean online;
private long onlineAt;
private boolean counted;
static volatile boolean jitterDisabled;
public long nextWaitMs() {
counted = false;
if (skipFirst) {
skipFirst = false;
return 0L;
}
int use = n < 1 ? 1 : n;
return jitter(nominalMs(use));
}
public long nextWaitNoJitterMs() {
counted = false;
if (skipFirst) {
skipFirst = false;
return 0L;
}
int use = n < 1 ? 1 : n;
return nominalMs(use);
}
public void reset() {
n = 0;
skipFirst = true;
online = false;
counted = false;
}
public void markOnline() {
online = true;
onlineAt = System.currentTimeMillis();
counted = false;
}
public void markOffline() {
if (counted) {
return;
}
counted = true;
boolean was = online;
long at = onlineAt;
online = false;
if (!was) {
n++;
return;
}
if (System.currentTimeMillis() - at >= STABLE_RESET_MS) {
n = 1;
return;
}
n++;
}
public void setOnlineAtForTest(long ms) {
online = true;
onlineAt = ms;
}
static void disableJitterForTest() {
jitterDisabled = true;
}
static void restoreJitterForTest() {
jitterDisabled = false;
}
static long nominalMs(int n) {
if (n < 1) {
return 0L;
}
if (n > 6) {
return BACKOFF_MAX_MS;
}
long d = BACKOFF_INITIAL_MS;
for (int i = 1; i < n; i++) {
d *= 2;
if (d >= BACKOFF_MAX_MS) {
return BACKOFF_MAX_MS;
}
}
return d;
}
static long jitter(long d) {
if (d <= 0) {
return 0L;
}
if (jitterDisabled) {
return d;
}
double f = 0.7 + Math.random() * 0.6;
return (long) (d * f);
}
}
}
sdk/java/src/test/java/asia/asio/nixmsg/ChecklistTest.java
+4
View File
@@ -491,6 +491,10 @@ public class ChecklistTest {
List<String> tokens = new ArrayList<String>();
c.onSession(tokens::add);
c.connectSync(srv.wsUrl, id, "password12", null, false);
long deadline = System.currentTimeMillis() + 2000;
while (tokens.isEmpty() && System.currentTimeMillis() < deadline) {
Thread.sleep(10);
}
assertFalse(tokens.isEmpty());
String token = tokens.get(0);
c.close();
sdk/java/src/test/java/asia/asio/nixmsg/ClientTest.java
+5 -1
View File
@@ -62,7 +62,7 @@ public class ClientTest {
}
@Test
public void sessionTokenCallback() {
public void sessionTokenCallback() throws Exception {
transport = new FakeTransport();
client = new Client(transport);
final List<String> tokens = new ArrayList<String>();
@@ -73,6 +73,10 @@ public class ClientTest {
}
});
client.connectSync("ws://example.test/mqtt", "ep1", "pw", null, false);
long deadline = System.currentTimeMillis() + 1000;
while (tokens.isEmpty() && System.currentTimeMillis() < deadline) {
Thread.sleep(10);
}
assertEquals(1, tokens.size());
assertEquals("nst_test_token", tokens.get(0));
assertEquals("nst_test_token", client.getSessionToken());
sdk/java/src/test/java/asia/asio/nixmsg/FakeTransport.java
+190
View File
@@ -0,0 +1,190 @@
package asia.asio.nixmsg;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
/** 单元测试假传输，不进发布 jar。 */
final class FakeTransport implements Transport {
final List<Transport.ConnectParams> connects = new ArrayList<Transport.ConnectParams>();
final List<byte[]> publishes = new CopyOnWriteArrayList<byte[]>();
final List<String> subscriptions = new ArrayList<String>();
volatile String nextConnackFail;
Map<String, Object> autoHello;
boolean autoSendOk = true;
final List<Function<Map<String, Object>, Map<String, Object>>> upHandlers =
new CopyOnWriteArrayList<Function<Map<String, Object>, Map<String, Object>>>();
private Runnable onConnected;
private BiConsumer<String, Boolean> onDisconnected;
private Consumer<byte[]> onDown;
private volatile boolean connected;
FakeTransport() {
autoHello = new LinkedHashMap<String, Object>();
autoHello.put("server_time_ms", 1750000000000L);
autoHello.put("server_version", "0.1.0");
autoHello.put("max_body_bytes", 262144);
autoHello.put("max_meta_bytes", 4096);
autoHello.put("max_frame_bytes", 786432);
autoHello.put("max_ttl_seconds", 2592000);
autoHello.put("max_schedule_seconds", 31536000);
autoHello.put("ack_timeout_seconds", 300);
autoHello.put("session_token", "nst_test_token");
}
void onUp(Function<Map<String, Object>, Map<String, Object>> h) {
upHandlers.add(h);
}
@Override
public void setHandlers(Runnable onConnected, BiConsumer<String, Boolean> onDisconnected, Consumer<byte[]> onDown) {
this.onConnected = onConnected;
this.onDisconnected = onDisconnected;
this.onDown = onDown;
}
@Override
public void connect(ConnectParams params) {
connects.add(params);
if (nextConnackFail != null) {
String fail = nextConnackFail;
nextConnackFail = null;
boolean stop = "session_invalid".equals(fail) || "bad_credentials".equals(fail) || "banned".equals(fail);
if (onDisconnected != null) {
onDisconnected.accept(fail, stop);
}
return;
}
connected = true;
if (onConnected != null) {
onConnected.run();
}
}
@Override
public void subscribe(String topic) {
subscriptions.add(topic);
}
@Override
public void publish(String topic, byte[] payload) {
publishes.add(payload);
Map<String, Object> frame = Protocol.loads(payload);
for (Function<Map<String, Object>, Map<String, Object>> h : upHandlers) {
Map<String, Object> resp = h.apply(frame);
if (resp != null) {
injectDown(Protocol.dumps(resp));
return;
}
}
String type = str(frame.get("type"));
Object rid = frame.get("rid");
if ("hello".equals(type) && autoHello != null) {
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", autoHello);
injectDown(Protocol.dumps(resp));
return;
}
if ("send".equals(type) && autoSendOk) {
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("id", frame.get("id"));
Object sat = frame.get("send_at_ms");
data.put("send_at_ms", sat == null ? 0 : sat);
data.put("state", "dispatched");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", data);
injectDown(Protocol.dumps(resp));
return;
}
if ("ack".equals(type)) {
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("result", "accepted");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", data);
injectDown(Protocol.dumps(resp));
return;
}
if (type != null && !"hello".equals(type) && !"send".equals(type) && rid != null) {
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", new LinkedHashMap<String, Object>());
injectDown(Protocol.dumps(resp));
}
}
@Override
public void disconnect() {
boolean was = connected;
connected = false;
if (was && onDisconnected != null) {
onDisconnected.accept("network", false);
}
}
void injectDown(byte[] payload) {
if (onDown != null) {
onDown.accept(payload);
}
}
void simulateTakenOver() {
connected = false;
if (onDisconnected != null) {
onDisconnected.accept("taken_over", true);
}
}
void simulateKick() {
simulateTakenOver();
}
void simulateNetworkDrop() {
connected = false;
if (onDisconnected != null) {
onDisconnected.accept("network", false);
}
}
void simulateServerDisconnect(int code) {
connected = false;
if (onDisconnected != null) {
if (code == 0x8E) {
onDisconnected.accept("taken_over", true);
} else {
onDisconnected.accept("network", false);
}
}
}
void simulateAuthFail(String reason) {
connected = false;
if (onDisconnected != null) {
onDisconnected.accept(reason, true);
}
}
private static String str(Object o) {
return o == null ? null : String.valueOf(o);
}
}
sdk/java/src/test/java/asia/asio/nixmsg/K00Test.java
+373
View File
@@ -0,0 +1,373 @@
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.ReconnectBackoff;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.SendResult;
import asia.asio.nixmsg.Types.Target;
import org.junit.After;
import org.junit.Test;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class K00Test {
private Client client;
@After
public void tearDown() {
Types.ReconnectBackoff.restoreJitterForTest();
if (client != null) {
client.close();
}
}
private Client connectOnline(FakeTransport tr) {
client = new Client(tr);
client.connectSync("ws://example.test/mqtt", "ep1", "p", null, false);
return client;
}
@Test
public void testK00FirstConnectTimeout() {
FakeTransport tr = new FakeTransport();
tr.autoHello = null;
client = new Client(tr, true, Types.DEFAULT_MAX_FRAME, Types.CLIENT_NAME, 150);
try {
client.connectSync("ws://example.test/mqtt", "ep1", "p", null, false);
fail("expected not_connected");
} catch (NixMsgException e) {
assertEquals("not_connected", e.getCode());
}
tr.autoHello = new LinkedHashMap<String, Object>();
tr.autoHello.put("server_time_ms", 1750000000000L);
tr.autoHello.put("server_version", "0.1.0");
tr.autoHello.put("max_body_bytes", 262144);
tr.autoHello.put("max_meta_bytes", 4096);
tr.autoHello.put("max_frame_bytes", 786432);
tr.autoHello.put("max_ttl_seconds", 2592000);
tr.autoHello.put("max_schedule_seconds", 31536000);
tr.autoHello.put("ack_timeout_seconds", 300);
tr.autoHello.put("session_token", "nst_test_token");
client.connectSync("ws://example.test/mqtt", "ep1", "p", null, false);
assertEquals(ConnectionState.ONLINE, client.getState());
}
@Test
public void testK00AuthErrorCodes() throws Exception {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
tr.simulateAuthFail("bad_credentials");
long deadline = System.currentTimeMillis() + 1000;
while (!"bad_credentials".equals(client.lastStopCodeForTest()) && System.currentTimeMillis() < deadline) {
Thread.sleep(10);
}
assertEquals("bad_credentials", client.lastStopCodeForTest());
try {
client.sendSync(new Target("endpoint", "b"), new Body("x"), new SendOptions());
fail("expected bad_credentials");
} catch (NixMsgException e) {
assertEquals("bad_credentials", e.getCode());
}
}
@Test
public void testK00TakenOverReason() throws Exception {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
final List<String> got = new CopyOnWriteArrayList<String>();
client.onConnection(new java.util.function.Consumer<ConnectionEvent>() {
@Override
public void accept(ConnectionEvent ev) {
if (ev.state == ConnectionState.KICKED) {
got.add(ev.reason);
}
}
});
tr.simulateKick();
long deadline = System.currentTimeMillis() + 1000;
while (got.isEmpty() && System.currentTimeMillis() < deadline) {
Thread.sleep(10);
}
assertTrue(got.contains("taken_over"));
assertEquals("taken_over", client.lastStopCodeForTest());
}
@Test
public void testK00Disconnect8BRetryable() throws Exception {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
tr.simulateServerDisconnect(0x8B);
Thread.sleep(50);
assertNotEquals("taken_over", client.lastStopCodeForTest());
}
@Test
public void testK00QueueFull() throws Exception {
FakeTransport tr = new FakeTransport();
tr.autoSendOk = false;
connectOnline(tr);
client.setSendQueueLimitForTest(1);
Thread holder = new Thread(new Runnable() {
@Override
public void run() {
try {
SendOptions opt = new SendOptions();
opt.messageId = "hold";
client.sendSync(new Target("endpoint", "b"), new Body("1"), opt);
} catch (Exception ignored) {
}
}
});
holder.setDaemon(true);
holder.start();
Thread.sleep(50);
try {
client.sendSync(new Target("endpoint", "b"), new Body("2"), new SendOptions());
fail("expected queue_full");
} catch (NixMsgException e) {
assertEquals("queue_full", e.getCode());
}
}
@Test
public void testK00RequestReturnsData() {
FakeTransport tr = new FakeTransport();
tr.onUp(new java.util.function.Function<Map<String, Object>, Map<String, Object>>() {
@Override
public Map<String, Object> apply(Map<String, Object> frame) {
if (!"self.get".equals(String.valueOf(frame.get("type")))) {
return null;
}
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("id", "ep1");
data.put("name", "n");
data.put("default_delay_ms", 0L);
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", frame.get("rid"));
resp.put("ok", true);
resp.put("data", data);
return resp;
}
});
connectOnline(tr);
Map<String, Object> info = client.getSelf().join();
assertEquals("ep1", String.valueOf(info.get("id")));
assertEquals("n", String.valueOf(info.get("name")));
assertFalse(info.containsKey("ok"));
}
@Test
public void testK00SendAtAndDelayConflict() {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
SendOptions opt = new SendOptions();
opt.sendAtDate = new Date(1);
opt.delayMs = 1000L;
try {
client.sendSync(new Target("endpoint", "b"), new Body("x"), opt);
fail("expected bad_request");
} catch (NixMsgException e) {
assertEquals("bad_request", e.getCode());
}
}
@Test
public void testK00SendAfterStopped() throws Exception {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
tr.simulateKick();
Thread.sleep(50);
try {
client.sendSync(new Target("endpoint", "b"), new Body("x"), new SendOptions());
fail("expected taken_over");
} catch (NixMsgException e) {
assertEquals("taken_over", e.getCode());
}
}
@Test
public void testK00LogoutReturnsError() throws Exception {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
tr.simulateNetworkDrop();
Thread.sleep(50);
try {
client.logout().join();
fail("expected error");
} catch (Exception e) {
Throwable c = e.getCause() == null ? e : e.getCause();
assertTrue(c instanceof NixMsgException);
assertEquals("not_connected", ((NixMsgException) c).getCode());
}
try {
client.sendSync(new Target("endpoint", "b"), new Body("x"), new SendOptions());
fail("expected stop error");
} catch (NixMsgException ignored) {
}
}
@Test
public void testK00DurationInt64() {
Map<String, Object> raw = Protocol.loads("{\"id\":\"m1\",\"send_at_ms\":123,\"state\":\"scheduled\"}".getBytes(java.nio.charset.StandardCharsets.UTF_8));
assertEquals(123L, ((Number) raw.get("send_at_ms")).longValue());
assertEquals(2592000000L, 30L * 24 * 3600 * 1000);
SendOptions opt = new SendOptions();
opt.delayMs = 2592000000L;
opt.ttlSeconds = 2592000L;
assertEquals(Long.valueOf(2592000000L), opt.delayMs);
}
@Test
public void testK00MaxReceiveBytesMin() {
FakeTransport tr = new FakeTransport();
client = new Client(tr, true, 512, Types.CLIENT_NAME, Types.CONNECT_TIMEOUT_MS);
try {
client.connectSync("ws://example.test/mqtt", "ep1", "p", null, false);
fail("expected bad_request");
} catch (NixMsgException e) {
assertEquals("bad_request", e.getCode());
}
}
@Test
public void testK00URLMapping() {
String u = Protocol.normalizeMqttWsUrl("https://host:7443/");
assertTrue(u.startsWith("wss://"));
assertTrue(u.endsWith("/mqtt"));
u = Protocol.normalizeMqttWsUrl("http://host/app");
assertTrue(u.startsWith("ws://"));
assertTrue(u.contains("/app"));
assertFalse(u.endsWith("/app/mqtt"));
try {
Protocol.normalizeMqttWsUrl("mqtt://host:1883", false);
fail("expected bad_request");
} catch (NixMsgException e) {
assertEquals("bad_request", e.getCode());
}
assertTrue(Protocol.normalizeMqttWsUrl("mqtt://host:1883", true).contains("mqtt://"));
}
@Test
public void testK00ReconnectBackoff() {
ReconnectBackoff b = new ReconnectBackoff();
assertEquals(0L, b.nextWaitNoJitterMs());
List<Long> got = new ArrayList<Long>();
for (int i = 0; i < 6; i++) {
b.markOffline();
got.add(b.nextWaitNoJitterMs());
}
assertEquals(java.util.Arrays.asList(1000L, 2000L, 4000L, 8000L, 16000L, 30000L), got);
ReconnectBackoff b2 = new ReconnectBackoff();
b2.nextWaitNoJitterMs();
b2.markOnline();
b2.setOnlineAtForTest(System.currentTimeMillis() - 61000);
b2.markOffline();
assertEquals(1000L, b2.nextWaitNoJitterMs());
}
@Test
public void testK00RateLimitedBackoff() {
Types.ReconnectBackoff.disableJitterForTest();
FakeTransport tr = new FakeTransport();
tr.autoSendOk = false;
final List<String> rids = new CopyOnWriteArrayList<String>();
tr.onUp(new java.util.function.Function<Map<String, Object>, Map<String, Object>>() {
@Override
public Map<String, Object> apply(Map<String, Object> frame) {
if (!"send".equals(String.valueOf(frame.get("type")))) {
return null;
}
String rid = String.valueOf(frame.get("rid"));
rids.add(rid);
if (rids.size() < 3) {
Map<String, Object> err = new LinkedHashMap<String, Object>();
err.put("code", "rate_limited");
err.put("message", "slow");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", false);
resp.put("error", err);
return resp;
}
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("id", frame.get("id"));
data.put("send_at_ms", frame.get("send_at_ms"));
data.put("state", "scheduled");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", rid);
resp.put("ok", true);
resp.put("data", data);
return resp;
}
});
connectOnline(tr);
SendOptions opt = new SendOptions();
opt.sendAtMs = 1700000000000L;
opt.messageId = "id1";
SendResult r = client.sendSync(new Target("endpoint", "ep2"), new Body("hi"), opt);
assertEquals("id1", r.id);
assertEquals(3, new java.util.HashSet<String>(rids).size());
}
@Test
public void testK00KeepaliveDefault() {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
assertEquals(30, tr.connects.get(0).keepAlive);
}
@Test
public void testK00NoReceiveMaximum() {
FakeTransport tr = new FakeTransport();
connectOnline(tr);
assertEquals(30, tr.connects.get(0).keepAlive);
// CONNECT 不暴露 Receive Maximum 字段
}
@Test
public void testK00CancelUnsent() {
// Java SDK 当前无取消 API；未发出条目随 close 以 closed 结束。
FakeTransport tr = new FakeTransport();
tr.autoSendOk = false;
connectOnline(tr);
final List<String> codes = new CopyOnWriteArrayList<String>();
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
client.sendSync(new Target("endpoint", "b"), new Body("x"), new SendOptions());
} catch (NixMsgException e) {
codes.add(e.getCode());
}
}
});
t.start();
try {
Thread.sleep(50);
client.close();
t.join(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
assertTrue(codes.contains("closed") || codes.contains("logged_out") || !codes.isEmpty());
}
}
sdk/java/src/test/java/asia/asio/nixmsg/K04Test.java
+76
View File
@@ -0,0 +1,76 @@
package asia.asio.nixmsg;
import asia.asio.nixmsg.Types.Body;
import asia.asio.nixmsg.Types.SendOptions;
import asia.asio.nixmsg.Types.SendResult;
import asia.asio.nixmsg.Types.Target;
import org.junit.After;
import org.junit.Test;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
public class K04Test {
private Client client;
@After
public void tearDown() {
if (client != null) {
client.close();
}
}
@Test
public void inflightResendNewRid() {
FakeTransport tr = new FakeTransport();
tr.autoSendOk = false;
final AtomicReference<String> firstRid = new AtomicReference<String>("");
final AtomicReference<String> firstId = new AtomicReference<String>("");
tr.onUp(new java.util.function.Function<Map<String, Object>, Map<String, Object>>() {
@Override
public Map<String, Object> apply(Map<String, Object> frame) {
if (!"send".equals(String.valueOf(frame.get("type")))) {
return null;
}
if (firstRid.get().isEmpty()) {
firstRid.set(String.valueOf(frame.get("rid")));
firstId.set(String.valueOf(frame.get("id")));
new Thread(new Runnable() {
@Override
public void run() {
tr.simulateNetworkDrop();
}
}).start();
return null;
}
if (!String.valueOf(frame.get("rid")).equals(firstRid.get())) {
assertEquals(firstId.get(), String.valueOf(frame.get("id")));
Map<String, Object> data = new LinkedHashMap<String, Object>();
data.put("id", frame.get("id"));
data.put("send_at_ms", frame.get("send_at_ms"));
data.put("state", "accepted");
Map<String, Object> resp = new LinkedHashMap<String, Object>();
resp.put("v", 1);
resp.put("type", "resp");
resp.put("rid", frame.get("rid"));
resp.put("ok", true);
resp.put("data", data);
return resp;
}
return null;
}
});
client = new Client(tr);
client.connectSync("ws://example.test/mqtt", "ep1", "pw", null, false);
SendOptions opt = new SendOptions();
opt.sendAtMs = 1700000000111L;
opt.messageId = "keep";
SendResult r = client.sendSync(new Target("endpoint", "ep2"), new Body("hi"), opt);
assertEquals("keep", r.id);
assertNotEquals("", firstRid.get());
}
}