Skip to content

Commit 523aee7

Browse files
zhou-haozhou-hao
authored
test(设备网关): 补充TCP上行监控回归测试 (#789)
1 parent 3c9dfc4 commit 523aee7

1 file changed

Lines changed: 188 additions & 0 deletions

File tree

‎jetlinks-components/network-component/tcp-component/src/test/java/org/jetlinks/community/network/tcp/gateway/device/TcpServerDeviceGatewayMonitorTest.java‎

Lines changed: 188 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,188 @@
1+
/*
2+
* Copyright 2026 JetLinks https://www.jetlinks.cn
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.jetlinks.community.network.tcp.gateway.device;
17+
18+
import io.netty.buffer.Unpooled;
19+
import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor;
20+
import org.jetlinks.community.gateway.monitor.GatewayMonitors;
21+
import org.jetlinks.community.network.tcp.TcpMessage;
22+
import org.jetlinks.community.network.tcp.client.TcpClient;
23+
import org.jetlinks.community.network.tcp.server.TcpServer;
24+
import org.jetlinks.core.ProtocolSupport;
25+
import org.jetlinks.core.device.DeviceInfo;
26+
import org.jetlinks.core.device.DeviceRegistry;
27+
import org.jetlinks.core.device.ProductInfo;
28+
import org.jetlinks.core.device.session.DeviceSessionManager;
29+
import org.jetlinks.core.message.DeviceMessage;
30+
import org.jetlinks.core.message.codec.DeviceMessageCodec;
31+
import org.jetlinks.core.message.codec.EncodedMessage;
32+
import org.jetlinks.core.message.codec.FromDeviceMessageContext;
33+
import org.jetlinks.core.message.codec.Transport;
34+
import org.jetlinks.core.message.property.ReportPropertyMessage;
35+
import org.jetlinks.core.server.ClientConnection;
36+
import org.jetlinks.core.server.session.DeviceSession;
37+
import org.jetlinks.supports.device.session.LocalDeviceSessionManager;
38+
import org.jetlinks.supports.server.DecodedClientMessageHandler;
39+
import org.jetlinks.supports.test.InMemoryDeviceRegistry;
40+
import org.jetlinks.supports.test.MockProtocolSupport;
41+
import org.junit.jupiter.api.Test;
42+
import reactor.core.publisher.Flux;
43+
import reactor.core.publisher.Mono;
44+
import reactor.core.publisher.Sinks;
45+
import reactor.test.StepVerifier;
46+
47+
import javax.annotation.Nonnull;
48+
import java.net.InetSocketAddress;
49+
import java.time.Duration;
50+
import java.util.Collections;
51+
import java.util.List;
52+
import java.util.concurrent.CopyOnWriteArrayList;
53+
import java.util.concurrent.CountDownLatch;
54+
import java.util.concurrent.TimeUnit;
55+
import java.util.concurrent.atomic.AtomicInteger;
56+
57+
import static org.junit.jupiter.api.Assertions.assertEquals;
58+
import static org.junit.jupiter.api.Assertions.assertSame;
59+
import static org.junit.jupiter.api.Assertions.assertTrue;
60+
import static org.mockito.ArgumentMatchers.any;
61+
import static org.mockito.Mockito.mock;
62+
import static org.mockito.Mockito.when;
63+
64+
class TcpServerDeviceGatewayMonitorTest {
65+
66+
@Test
67+
void manualAndReturnedTcpUpstreamShouldUseMonitorChainOnce() throws InterruptedException {
68+
String gatewayId = "tcp-monitor-test";
69+
RecordingMonitor monitor = new RecordingMonitor();
70+
GatewayMonitors.register((id, tags) -> gatewayId.equals(id) ? monitor : null);
71+
72+
DeviceMessageCodec codec = mock(DeviceMessageCodec.class);
73+
ProtocolSupport protocol = new MockProtocolSupport() {
74+
@Nonnull
75+
@Override
76+
public Mono<DeviceMessageCodec> getMessageCodec(Transport transport) {
77+
return Mono.just(codec);
78+
}
79+
};
80+
DeviceRegistry registry = new InMemoryDeviceRegistry();
81+
String productId = "monitor-test-product";
82+
Mono<Void> registerDevices = registry
83+
.register(ProductInfo.builder().id(productId).protocol("test").build())
84+
.then(registry.register(DeviceInfo
85+
.builder()
86+
.id("manual-device")
87+
.productId(productId)
88+
.build()))
89+
.then(registry.register(DeviceInfo
90+
.builder()
91+
.id("returned-device")
92+
.productId(productId)
93+
.build()))
94+
.then();
95+
StepVerifier.create(registerDevices).verifyComplete();
96+
DeviceSessionManager sessionManager = LocalDeviceSessionManager.create();
97+
DecodedClientMessageHandler messageHandler = (deviceOperator, message) -> Mono.just(true);
98+
TcpServer server = mock(TcpServer.class);
99+
TcpClient client = mock(TcpClient.class);
100+
ReportPropertyMessage manualMessage = new ReportPropertyMessage();
101+
manualMessage.setDeviceId("manual-device");
102+
ReportPropertyMessage returnedMessage = new ReportPropertyMessage();
103+
returnedMessage.setDeviceId("returned-device");
104+
TcpMessage origin = new TcpMessage(Unpooled.EMPTY_BUFFER);
105+
Sinks.Many<TcpClient> clients = Sinks.many().unicast().onBackpressureBuffer();
106+
Sinks.Many<TcpMessage> messages = Sinks.many().replay().all();
107+
108+
when(codec.decode(any())).thenAnswer(invocation -> {
109+
FromDeviceMessageContext context = invocation.getArgument(0);
110+
return context
111+
.handleMessage(manualMessage)
112+
.thenMany(Flux.just(returnedMessage));
113+
});
114+
when(server.handleConnection()).thenReturn(clients.asFlux());
115+
when(client.subscribe()).thenReturn(messages.asFlux());
116+
when(client.getRemoteAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 1883));
117+
when(client.getId()).thenReturn("tcp-client");
118+
119+
TcpServerDeviceGateway gateway = new TcpServerDeviceGateway(
120+
gatewayId,
121+
Mono.just(protocol),
122+
registry,
123+
messageHandler,
124+
sessionManager,
125+
server
126+
);
127+
StepVerifier.create(gateway.startup()).verifyComplete();
128+
129+
assertEquals(Sinks.EmitResult.OK, clients.tryEmitNext(client));
130+
assertEquals(Sinks.EmitResult.OK, messages.tryEmitNext(origin));
131+
assertTrue(monitor.completed.await(Duration.ofSeconds(5).toMillis(), TimeUnit.MILLISECONDS));
132+
133+
assertEquals(1, monitor.beforeDecode.get());
134+
assertEquals(1, monitor.decode.get());
135+
assertEquals(2, monitor.beforeSend.get());
136+
assertEquals(2, monitor.received.get());
137+
assertEquals(1, Collections.frequency(monitor.monitored, manualMessage));
138+
assertEquals(1, Collections.frequency(monitor.monitored, returnedMessage));
139+
assertSame(origin, monitor.origin);
140+
141+
messages.tryEmitComplete();
142+
clients.tryEmitComplete();
143+
StepVerifier.create(gateway.shutdown()).verifyComplete();
144+
}
145+
146+
private static class RecordingMonitor implements DeviceGatewayMonitor {
147+
private final AtomicInteger beforeDecode = new AtomicInteger();
148+
private final AtomicInteger decode = new AtomicInteger();
149+
private final AtomicInteger beforeSend = new AtomicInteger();
150+
private final AtomicInteger received = new AtomicInteger();
151+
private final CountDownLatch completed = new CountDownLatch(2);
152+
private final List<DeviceMessage> monitored = new CopyOnWriteArrayList<>();
153+
private EncodedMessage origin;
154+
155+
@Override
156+
public void receivedMessage() {
157+
received.incrementAndGet();
158+
}
159+
160+
@Override
161+
public boolean beforeDecode(ClientConnection connection, EncodedMessage message) {
162+
beforeDecode.incrementAndGet();
163+
origin = message;
164+
return true;
165+
}
166+
167+
@Override
168+
public Flux<DeviceMessage> decode(ClientConnection connection,
169+
DeviceSession session,
170+
EncodedMessage origin,
171+
Flux<DeviceMessage> decoder) {
172+
decode.incrementAndGet();
173+
return decoder;
174+
}
175+
176+
@Override
177+
public Flux<DeviceMessage> beforeSendToPlatform(ClientConnection connection,
178+
DeviceSession session,
179+
EncodedMessage origin,
180+
Flux<DeviceMessage> handler) {
181+
beforeSend.incrementAndGet();
182+
return handler.doOnNext(message -> {
183+
monitored.add(message);
184+
completed.countDown();
185+
});
186+
}
187+
}
188+
}

0 commit comments

Comments
 (0)