Skip to content

Commit bdb88df

Browse files
cuihairucuihairu
committed
fix(sdk/java)+test: 首连不挂 inbound listener 修复 + 入站/编解码缺口补测
产品修复（测试暴露的真 bug）：connect() 首次连接从不调用 attachInboundListener——仅重连路径（recoverConnection）挂载。首连的 Java Provider 对所有 Agent 主动调用（invoke 等）无响应，直到发生一次 重连才恢复。修复：connect 成功路径在 connected 置位前挂载 listener。 顺带删除死代码：SdkWireMessages.writeMap 无任何调用方（encoder 均 内联 map 循环）。 补测（覆盖 92→74 missed，97%）： - TCPTransportInboundTest：agent 推送 invoke 到达 listener 并回写/ handler 异常回空且连接存活/无 listener 丢弃帧/16 并发全应答 （TCPTransport 71→14 missed——dispatchInbound/inboundPool/ writeResponseSilently/readLoop 入站分支全通） - CroupierClientInboundTcpTest：真实 TCP 全链路（client 级）—— agent invoke 经 TCPTransport 派发到本地 handler；未知函数回空 - SdkWireMessagesEdgeTest：10 个 decoder 的垃圾输入 catch 分支、 unknown-field skipField 全覆盖、非法 map entry、各 record round-trip
1 parent b3a0736 commit bdb88df

5 files changed

Lines changed: 471 additions & 169 deletions

File tree

‎sdks/java/src/main/java/io/github/cuihairu/croupier/sdk/CroupierClientImpl.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -133,6 +133,9 @@ public CompletableFuture<Void> connect() {
133133
}
134134
transport = nextTransport;
135135
sessionId = response.sessionId;
136+
// 首连也必须挂入站 listener（Agent→Provider 调用）——此前仅
137+
// 重连路径挂载，首连客户端对所有 agent 主动调用无响应。
138+
attachInboundListener(nextTransport);
136139
connected.set(true);
137140
startHeartbeatLoop();
138141

‎sdks/java/src/main/java/io/github/cuihairu/croupier/sdk/wire/SdkWireMessages.java‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -441,15 +441,6 @@ private static ProviderFunctionDescriptor decodeProviderFunctionDescriptor(byte[
441441
);
442442
}
443443

444-
private static void writeMap(CodedOutputStream out, int fieldNumber, Map<String, String> value) throws IOException {
445-
if (value == null || value.isEmpty()) {
446-
return;
447-
}
448-
for (Map.Entry<String, String> entry : value.entrySet()) {
449-
writeMessage(out, fieldNumber, encodeMapEntry(entry.getKey(), entry.getValue()));
450-
}
451-
}
452-
453444
private static void readMapEntry(byte[] data, Map<String, String> target) {
454445
String key = "";
455446
String value = "";

‎sdks/java/src/test/java/io/github/cuihairu/croupier/sdk/CroupierClientInboundTcpTest.java‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
1+
/**
2+
* 真实 TCP 链路的 Agent→Provider 入站调用测试（覆盖 attachInboundListener /
3+
* handleLocalRequest / handleInvokeRequest——此前全部测试用 FakeTransportClient，
4+
* instanceof TCPTransport 分支与入站 lambda 从未执行）。
5+
*/
6+
package io.github.cuihairu.croupier.sdk;
7+
8+
import io.github.cuihairu.croupier.sdk.transport.Protocol;
9+
import io.github.cuihairu.croupier.sdk.wire.SdkWireMessages;
10+
import org.junit.jupiter.api.AfterEach;
11+
import org.junit.jupiter.api.BeforeEach;
12+
import org.junit.jupiter.api.DisplayName;
13+
import org.junit.jupiter.api.Test;
14+
import org.junit.jupiter.api.Timeout;
15+
16+
import java.io.DataInputStream;
17+
import java.io.DataOutputStream;
18+
import java.io.IOException;
19+
import java.net.ServerSocket;
20+
import java.net.Socket;
21+
import java.util.List;
22+
import java.util.Map;
23+
import java.util.concurrent.TimeUnit;
24+
import java.util.concurrent.atomic.AtomicInteger;
25+
26+
import static org.junit.jupiter.api.Assertions.*;
27+
28+
@DisplayName("Agent→Provider inbound over real TCP")
29+
class CroupierClientInboundTcpTest {
30+
31+
private ServerSocket server;
32+
private Socket serverSide;
33+
private DataInputStream in;
34+
private DataOutputStream out;
35+
private CroupierClientImpl client;
36+
37+
@BeforeEach
38+
void setUp() throws IOException {
39+
server = new ServerSocket(0);
40+
}
41+
42+
@AfterEach
43+
void tearDown() {
44+
if (client != null) {
45+
client.stop();
46+
}
47+
closeQuietly(serverSide);
48+
closeQuietly(server);
49+
}
50+
51+
private volatile String handshakeLog = "";
52+
53+
private void acceptAndHandshake() throws IOException {
54+
server.setSoTimeout(10000);
55+
serverSide = server.accept();
56+
serverSide.setSoTimeout(8000); // 读超时：断言失败时线程可退出
57+
in = new DataInputStream(serverSide.getInputStream());
58+
out = new DataOutputStream(serverSide.getOutputStream());
59+
// 应答 ProviderConnectRequest。
60+
Protocol.ParsedMessage first = readFrame();
61+
handshakeLog = "first-frame=0x" + Integer.toHexString(first.msgId) + " reqId=" + first.reqId;
62+
System.err.println("[agent] " + handshakeLog);
63+
writeFrame(Protocol.MSG_PROVIDER_CONNECT_RESPONSE, first.reqId,
64+
SdkWireMessages.encodeProviderConnectResponse(
65+
new SdkWireMessages.ProviderConnectResponse("inbound-session")));
66+
}
67+
68+
private void writeFrame(int msgId, int reqId, byte[] body) throws IOException {
69+
byte[] frame = Protocol.newMessage(msgId, reqId, body);
70+
out.writeInt(frame.length);
71+
out.write(frame);
72+
out.flush();
73+
}
74+
75+
private Protocol.ParsedMessage readFrame() throws IOException {
76+
int len = in.readInt();
77+
byte[] payload = new byte[len];
78+
in.readFully(payload);
79+
return Protocol.parseMessage(payload);
80+
}
81+
82+
private ClientConfig config(int port) {
83+
ClientConfig config = new ClientConfig();
84+
config.setAgentAddr("127.0.0.1:" + port);
85+
config.setGameId("game-test");
86+
config.setEnv("development");
87+
config.setServiceId("java-inbound-tests");
88+
config.setHeartbeatInterval(60); // 拉长心跳，避免与断言交错
89+
config.setTimeoutSeconds(5);
90+
return config;
91+
}
92+
93+
@Test
94+
@Timeout(20)
95+
@DisplayName("agent 推送 invoke：经 TCPTransport 派发到本地 handler 并回写响应")
96+
void agentInvokeDispatchesToLocalHandler() throws Exception {
97+
int port = server.getLocalPort();
98+
client = new CroupierClientImpl(config(port));
99+
AtomicInteger calls = new AtomicInteger();
100+
client.registerFunction(new FunctionDescriptor("test.echo", "1.0.0"),
101+
(ctx, payload) -> {
102+
calls.incrementAndGet();
103+
return "echo:" + payload;
104+
});
105+
// connect() 会同步等 ProviderConnectResponse——agent 线程先行。
106+
Thread agent = new Thread(() -> {
107+
try {
108+
acceptAndHandshake();
109+
} catch (Exception e) {
110+
// 由主线程断言失败兜底
111+
}
112+
});
113+
agent.setDaemon(true);
114+
agent.start();
115+
client.connect().get(5, TimeUnit.SECONDS);
116+
agent.join(5000);
117+
118+
byte[] body = SdkWireMessages.encodeInvokeRequest(
119+
new SdkWireMessages.InvokeRequest("test.echo", "", new byte[] {'h', 'i'}, Map.of()));
120+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 6001, body);
121+
122+
Protocol.ParsedMessage resp = readFrame();
123+
assertEquals(6001, resp.reqId);
124+
SdkWireMessages.InvokeResponse parsed = SdkWireMessages.decodeInvokeResponse(resp.body);
125+
assertEquals("echo:hi", parsed.payloadUtf8());
126+
assertTrue(calls.get() >= 1, "handler should have been invoked");
127+
}
128+
129+
@Test
130+
@Timeout(20)
131+
@DisplayName("未知函数：空响应（handleInvokeRequest 的 not found 分支）")
132+
void unknownFunctionReturnsEmpty() throws Exception {
133+
int port = server.getLocalPort();
134+
client = new CroupierClientImpl(config(port));
135+
client.registerFunction(new FunctionDescriptor("known.fn", "1.0.0"),
136+
(ctx, payload) -> "{}");
137+
Thread agent = new Thread(() -> {
138+
try {
139+
acceptAndHandshake();
140+
} catch (Exception ignored) {
141+
}
142+
});
143+
agent.setDaemon(true);
144+
agent.start();
145+
client.connect().get(5, TimeUnit.SECONDS);
146+
agent.join(5000);
147+
148+
byte[] body = SdkWireMessages.encodeInvokeRequest(
149+
new SdkWireMessages.InvokeRequest("missing.fn", "", new byte[0], Map.of()));
150+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 6002, body);
151+
152+
Protocol.ParsedMessage resp = readFrame();
153+
assertEquals(6002, resp.reqId);
154+
// 未注册函数回空体或错误 JSON——帧必须到达（连接存活）。
155+
assertTrue(resp.body.length == 0 || new String(resp.body).contains("error"));
156+
}
157+
158+
private static void closeQuietly(java.io.Closeable c) {
159+
if (c != null) {
160+
try {
161+
c.close();
162+
} catch (IOException ignored) {
163+
}
164+
}
165+
}
166+
}

‎sdks/java/src/test/java/io/github/cuihairu/croupier/sdk/transport/TCPTransportInboundTest.java‎

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
/**
2+
* 入站派发路径测试（覆盖 dispatchInbound / inboundPool / writeResponseSilently /
3+
* readLoop 入站分支 / setInboundListener——与 C++/C# 同款语义验证）。
4+
*/
5+
package io.github.cuihairu.croupier.sdk.transport;
6+
7+
import org.junit.jupiter.api.AfterEach;
8+
import org.junit.jupiter.api.BeforeEach;
9+
import org.junit.jupiter.api.DisplayName;
10+
import org.junit.jupiter.api.Test;
11+
12+
import java.io.DataInputStream;
13+
import java.io.DataOutputStream;
14+
import java.io.IOException;
15+
import java.net.ServerSocket;
16+
import java.net.Socket;
17+
import java.util.concurrent.atomic.AtomicInteger;
18+
19+
import static org.junit.jupiter.api.Assertions.*;
20+
21+
@DisplayName("TCPTransport inbound dispatch")
22+
class TCPTransportInboundTest {
23+
24+
private ServerSocket server;
25+
private Socket serverSide;
26+
private DataInputStream in;
27+
private DataOutputStream out;
28+
private TCPTransport transport;
29+
30+
@BeforeEach
31+
void setUp() throws IOException {
32+
server = new ServerSocket(0);
33+
}
34+
35+
@AfterEach
36+
void tearDown() {
37+
if (transport != null) {
38+
try {
39+
transport.close();
40+
} catch (Exception ignored) {
41+
}
42+
}
43+
closeQuietly(serverSide);
44+
closeQuietly(server);
45+
}
46+
47+
/** 接受一条连接，供测试端直接读写帧（无 mock 循环）。 */
48+
private void acceptOne() throws IOException {
49+
server.setSoTimeout(5000);
50+
serverSide = server.accept();
51+
in = new DataInputStream(serverSide.getInputStream());
52+
out = new DataOutputStream(serverSide.getOutputStream());
53+
}
54+
55+
private void writeFrame(int msgId, int reqId, byte[] body) throws IOException {
56+
byte[] frame = Protocol.newMessage(msgId, reqId, body);
57+
out.writeInt(frame.length);
58+
out.write(frame);
59+
out.flush();
60+
}
61+
62+
private Protocol.ParsedMessage readFrame() throws IOException {
63+
int len = in.readInt();
64+
byte[] payload = new byte[len];
65+
in.readFully(payload);
66+
return Protocol.parseMessage(payload);
67+
}
68+
69+
@Test
70+
@DisplayName("agent 推送 invoke：listener 执行并回写 payload 响应")
71+
void invokeReachesListenerAndResponds() throws Exception {
72+
int port = server.getLocalPort();
73+
AtomicInteger calls = new AtomicInteger();
74+
transport = new TCPTransport("127.0.0.1", port, 3000);
75+
transport.setInboundListener((msgId, reqId, body) -> {
76+
calls.incrementAndGet();
77+
return body;
78+
});
79+
transport.connect();
80+
acceptOne();
81+
82+
byte[] body = {1, 2, 3, 4};
83+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 7001, body);
84+
85+
Protocol.ParsedMessage resp = readFrame();
86+
assertEquals(7001, resp.reqId);
87+
assertArrayEquals(body, resp.body);
88+
// 轮询等待 listener 计数（worker 异步执行）。
89+
for (int i = 0; i < 40 && calls.get() < 1; i++) {
90+
Thread.sleep(50);
91+
}
92+
assertTrue(calls.get() >= 1);
93+
}
94+
95+
@Test
96+
@DisplayName("listener 抛异常：回空响应且连接仍可用")
97+
void handlerExceptionYieldsEmptyResponseAndConnectionSurvives() throws Exception {
98+
int port = server.getLocalPort();
99+
transport = new TCPTransport("127.0.0.1", port, 3000);
100+
transport.setInboundListener((msgId, reqId, body) -> {
101+
if (body.length == 0) {
102+
throw new IllegalStateException("boom");
103+
}
104+
return body;
105+
});
106+
transport.connect();
107+
acceptOne();
108+
109+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 7002, new byte[0]);
110+
Protocol.ParsedMessage resp = readFrame();
111+
assertEquals(7002, resp.reqId);
112+
assertEquals(0, resp.body.length);
113+
114+
// 连接仍可用：合法请求正常回。
115+
byte[] ok = {9};
116+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 7003, ok);
117+
Protocol.ParsedMessage resp2 = readFrame();
118+
assertArrayEquals(ok, resp2.body);
119+
}
120+
121+
@Test
122+
@DisplayName("未设置 listener：帧被丢弃、连接保持")
123+
void noListenerDropsFrame() throws Exception {
124+
int port = server.getLocalPort();
125+
transport = new TCPTransport("127.0.0.1", port, 3000);
126+
transport.connect();
127+
acceptOne();
128+
129+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 7004, new byte[] {5});
130+
// 短等待后 transport 仍应处于连接态（无响应、无崩溃）。
131+
Thread.sleep(200);
132+
assertTrue(transport.isConnected());
133+
}
134+
135+
@Test
136+
@DisplayName("并发推送：全部请求收到响应（有界池 + 队列语义）")
137+
void concurrentInvocationsAllAnswered() throws Exception {
138+
int port = server.getLocalPort();
139+
transport = new TCPTransport("127.0.0.1", port, 5000);
140+
transport.setInboundListener((msgId, reqId, body) -> {
141+
try {
142+
Thread.sleep(30);
143+
} catch (InterruptedException ignored) {
144+
}
145+
return body;
146+
});
147+
transport.connect();
148+
acceptOne();
149+
150+
final int total = 16;
151+
for (int i = 0; i < total; i++) {
152+
writeFrame(Protocol.MSG_INVOKE_REQUEST, 8000 + i, new byte[] {(byte) i});
153+
}
154+
int answered = 0;
155+
for (int i = 0; i < total; i++) {
156+
Protocol.ParsedMessage resp = readFrame();
157+
// 空体（饱和快速失败）也计为应答——语义允许。
158+
if (resp.body.length == 1 || resp.body.length == 0) {
159+
answered++;
160+
}
161+
}
162+
assertEquals(total, answered);
163+
}
164+
165+
private static void closeQuietly(java.io.Closeable c) {
166+
if (c != null) {
167+
try {
168+
c.close();
169+
} catch (IOException ignored) {
170+
}
171+
}
172+
}
173+
}

0 commit comments

Comments
 (0)