Skip to content

Commit db894b8

Browse files
zhou-haozhou-hao
committed
feat(设备网关): 增加连接与消息链路监控能力
扩展 DeviceGatewayMonitor 的连接、解码、上行和下行钩子。 接入 TCP、MQTT、HTTP 网络组件并补齐指标查询和回归测试。
1 parent f5010a0 commit db894b8

28 files changed

Lines changed: 1558 additions & 107 deletions

File tree

‎jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java‎

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,14 @@
1616
package org.jetlinks.community.gateway.monitor;
1717

1818

19+
import org.jetlinks.core.message.DeviceMessage;
20+
import org.jetlinks.core.message.codec.EncodedMessage;
21+
import org.jetlinks.core.server.ClientConnection;
22+
import org.jetlinks.core.server.session.DeviceSession;
23+
import reactor.core.publisher.Flux;
24+
import reactor.core.publisher.Mono;
25+
26+
import javax.annotation.Nullable;
1927
import java.util.ArrayList;
2028
import java.util.Arrays;
2129
import java.util.Collection;
@@ -24,7 +32,7 @@
2432

2533
class CompositeDeviceGatewayMonitor implements DeviceGatewayMonitor {
2634

27-
private List<DeviceGatewayMonitor> monitors = new ArrayList<>();
35+
private final List<DeviceGatewayMonitor> monitors = new ArrayList<>();
2836

2937
public CompositeDeviceGatewayMonitor add(DeviceGatewayMonitor... monitors) {
3038
return add(Arrays.asList(monitors));
@@ -69,4 +77,69 @@ public void receivedMessage() {
6977
public void sentMessage() {
7078
doWith(DeviceGatewayMonitor::sentMessage);
7179
}
80+
81+
@Override
82+
public boolean connected(ClientConnection connection) {
83+
boolean accepted = true;
84+
for (DeviceGatewayMonitor monitor : monitors) {
85+
if (!monitor.connected(connection)) {
86+
accepted = false;
87+
}
88+
}
89+
return accepted;
90+
}
91+
92+
@Override
93+
public void disconnected(ClientConnection connection) {
94+
doWith(monitor -> monitor.disconnected(connection));
95+
}
96+
97+
@Override
98+
public void rejected(ClientConnection connection, @Nullable Throwable error) {
99+
doWith(monitor -> monitor.rejected(connection, error));
100+
}
101+
102+
@Override
103+
public boolean beforeDecode(@Nullable ClientConnection connection, EncodedMessage message) {
104+
boolean accepted = true;
105+
for (DeviceGatewayMonitor monitor : monitors) {
106+
if (!monitor.beforeDecode(connection, message)) {
107+
accepted = false;
108+
}
109+
}
110+
return accepted;
111+
}
112+
113+
@Override
114+
public Flux<DeviceMessage> decode(@Nullable ClientConnection connection,
115+
DeviceSession session,
116+
EncodedMessage origin,
117+
Flux<DeviceMessage> decoder) {
118+
for (DeviceGatewayMonitor monitor : monitors) {
119+
decoder = monitor.decode(connection, session, origin, decoder);
120+
}
121+
return decoder;
122+
}
123+
124+
@Override
125+
public Flux<DeviceMessage> beforeSendToPlatform(@Nullable ClientConnection connection,
126+
DeviceSession session,
127+
EncodedMessage origin,
128+
Flux<DeviceMessage> handler) {
129+
for (DeviceGatewayMonitor monitor : monitors) {
130+
handler = monitor.beforeSendToPlatform(connection, session, origin, handler);
131+
}
132+
return handler;
133+
}
134+
135+
@Override
136+
public Mono<Void> downstream(ClientConnection connection,
137+
DeviceSession session,
138+
EncodedMessage origin,
139+
Mono<Void> sender) {
140+
for (DeviceGatewayMonitor monitor : monitors) {
141+
sender = monitor.downstream(connection, session, origin, sender);
142+
}
143+
return sender;
144+
}
72145
}

‎jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java‎

Lines changed: 150 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,41 +15,182 @@
1515
*/
1616
package org.jetlinks.community.gateway.monitor;
1717

18+
import org.jetlinks.core.message.DeviceMessage;
19+
import org.jetlinks.core.message.codec.EncodedMessage;
20+
import org.jetlinks.core.server.ClientConnection;
21+
import org.jetlinks.core.server.session.DeviceSession;
22+
import reactor.core.publisher.Flux;
23+
import reactor.core.publisher.Mono;
24+
25+
import javax.annotation.Nullable;
26+
1827
/**
19-
* 设备网关监控
28+
* 设备网关监控扩展点。
29+
*
30+
* <p>由设备网关在连接生命周期、报文解码和消息上下行阶段调用。实现可记录指标、拒绝连接或报文，
31+
* 也可按注册顺序包装响应式处理链；实现不得在调用线程中阻塞或主动订阅传入的发布者。</p>
32+
*
33+
* @see DeviceGatewayMonitorSupplier
34+
* @see GatewayMonitors
35+
* @since 1.0
2036
*/
2137
public interface DeviceGatewayMonitor {
2238

2339
/**
2440
* 上报总连接数
2541
*
2642
* @param total 总连接数
43+
* @deprecated 使用 {@link #connected(ClientConnection)} 和 {@link #disconnected(ClientConnection)} 监听连接生命周期
2744
*/
28-
void totalConnection(long total);
45+
@Deprecated
46+
default void totalConnection(long total) {
47+
48+
}
2949

3050
/**
3151
* 创建新连接
3252
*/
33-
void connected();
53+
default void connected() {
54+
55+
}
3456

3557
/**
3658
* 拒绝连接
3759
*/
38-
void rejected();
60+
default void rejected() {
61+
62+
}
3963

4064
/**
4165
* 断开连接
4266
*/
43-
void disconnected();
67+
default void disconnected() {
68+
69+
}
70+
71+
/**
72+
* 网关接收消息
73+
*/
74+
default void receivedMessage() {
75+
76+
}
77+
78+
/**
79+
* 网关发送消息
80+
*/
81+
default void sentMessage() {
82+
83+
}
4484

4585
/**
46-
* 接收消息
86+
* 客户端长连接建立时执行。
87+
*
88+
* <p>默认兼容调用 {@link #connected()}。返回 {@code false} 时网关将拒绝本次连接；
89+
* 多个监控实现会全部执行，并合并各自的允许结果。</p>
90+
*
91+
* @param connection 新建立的客户端连接
92+
* @return {@code true} 允许连接，{@code false} 拒绝连接
93+
* @since 2.12
94+
*/
95+
default boolean connected(ClientConnection connection) {
96+
connected();
97+
return true;
98+
}
99+
100+
/**
101+
* 客户端长连接断开时执行。
102+
*
103+
* @param connection 已断开的客户端连接
104+
* @since 2.12
47105
*/
48-
void receivedMessage();
106+
default void disconnected(ClientConnection connection) {
107+
disconnected();
108+
}
49109

50110
/**
51-
* 发送消息
111+
* 客户端连接被拒绝时执行，例如 MQTT 认证失败或 TCP 长时间未解析出设备。
112+
*
113+
* @param connection 被拒绝的客户端连接
114+
* @param error 拒绝原因；无关联异常时为 {@code null}
115+
* @since 2.12
116+
*/
117+
default void rejected(ClientConnection connection, @Nullable Throwable error) {
118+
rejected();
119+
}
120+
121+
/**
122+
* 原始报文进入协议解码前执行。
123+
*
124+
* <p>短连接传输可能不提供连接对象。返回 {@code false} 时本次报文会被丢弃，且不会进入协议解码。</p>
125+
*
126+
* @param connection 客户端连接，短连接场景可能为 {@code null}
127+
* @param message 待解码的原始报文
128+
* @return {@code true} 继续解码，{@code false} 丢弃报文
129+
* @since 2.12
130+
*/
131+
default boolean beforeDecode(@Nullable ClientConnection connection,
132+
EncodedMessage message) {
133+
return true;
134+
}
135+
136+
/**
137+
* 包装设备上行报文的协议解码任务。
138+
*
139+
* <p>实现应返回基于 {@code decoder} 组合出的发布者，保留原链路的背压、取消和错误信号，
140+
* 不得在方法内主动订阅。</p>
141+
*
142+
* @param connection 客户端连接，短连接场景可能为 {@code null}
143+
* @param session 当前设备会话
144+
* @param origin 原始报文
145+
* @param decoder 协议解码任务
146+
* @return 包装后的解码任务
147+
* @since 2.12
148+
*/
149+
default Flux<DeviceMessage> decode(@Nullable ClientConnection connection,
150+
DeviceSession session,
151+
EncodedMessage origin,
152+
Flux<DeviceMessage> decoder) {
153+
return decoder;
154+
}
155+
156+
/**
157+
* 包装解码完成后、发送到平台前的上行处理任务。
158+
*
159+
* <p>{@code handler} 已包含平台消息处理逻辑。实现应保留其背压、取消和错误信号，
160+
* 不得在方法内主动订阅。</p>
161+
*
162+
* @param connection 客户端连接，短连接场景可能为 {@code null}
163+
* @param session 当前设备会话
164+
* @param origin 原始报文
165+
* @param handler 上行平台处理任务
166+
* @return 包装后的上行处理任务
167+
* @since 2.12
168+
*/
169+
default Flux<DeviceMessage> beforeSendToPlatform(@Nullable ClientConnection connection,
170+
DeviceSession session,
171+
EncodedMessage origin,
172+
Flux<DeviceMessage> handler) {
173+
return handler;
174+
}
175+
176+
/**
177+
* 包装平台消息下发到设备的发送任务。
178+
*
179+
* <p>实现应返回基于 {@code sender} 组合出的发布者，并保留原任务的取消和错误信号，
180+
* 不得在方法内主动订阅。</p>
181+
*
182+
* @param connection 当前客户端连接
183+
* @param session 当前设备会话
184+
* @param origin 待发送的原始报文
185+
* @param sender 原始发送任务
186+
* @return 包装后的发送任务
187+
* @since 2.12
52188
*/
53-
void sentMessage();
189+
default Mono<Void> downstream(ClientConnection connection,
190+
DeviceSession session,
191+
EncodedMessage origin,
192+
Mono<Void> sender) {
193+
return sender;
194+
}
54195

55196
}

‎jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,28 @@
1515
*/
1616
package org.jetlinks.community.gateway.monitor;
1717

18+
/**
19+
* 设备网关监控供应商。
20+
*
21+
* <p>通过 {@link GatewayMonitors#register(DeviceGatewayMonitorSupplier)} 注册后，
22+
* 在设备网关首次使用监控能力时按网关标识创建监控实例。</p>
23+
*
24+
* @see DeviceGatewayMonitor
25+
* @see GatewayMonitors
26+
* @since 1.0
27+
*/
1828
public interface DeviceGatewayMonitorSupplier {
19-
DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags);
29+
30+
/**
31+
* 为指定设备网关创建监控实例。
32+
*
33+
* <p>该方法可能由多个网关并发调用。返回 {@code null} 表示当前供应商不监控该网关；
34+
* 返回的监控实例应遵循 {@link DeviceGatewayMonitor} 的非阻塞与响应式包装约束。</p>
35+
*
36+
* @param id 设备网关标识
37+
* @param tags 网关附加标签
38+
* @return 监控实例，或 {@code null} 跳过当前供应商
39+
*/
40+
DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags);
2041

2142
}

‎jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java‎

Lines changed: 31 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,19 +20,32 @@
2020
import java.util.concurrent.CopyOnWriteArrayList;
2121
import java.util.stream.Collectors;
2222

23+
/**
24+
* 设备网关监控注册与获取入口。
25+
*
26+
* <p>供应商使用写时复制集合保存，支持在网关运行期间注册。返回的监控会延迟解析供应商，
27+
* 以便网关可以早于监控组件创建。</p>
28+
*
29+
* @see DeviceGatewayMonitor
30+
* @see DeviceGatewayMonitorSupplier
31+
* @since 1.0
32+
*/
2333
public class GatewayMonitors {
2434

25-
2635
private static final List<DeviceGatewayMonitorSupplier> deviceGatewayMonitorSuppliers = new CopyOnWriteArrayList<>();
2736

28-
static final NoneDeviceGatewayMonitor nonDevice = new NoneDeviceGatewayMonitor();
29-
30-
31-
static {
32-
33-
}
34-
37+
/**
38+
* 未注册有效供应商时使用的空监控。
39+
*
40+
* @since 2.12
41+
*/
42+
public static final DeviceGatewayMonitor nonDevice = new NoneDeviceGatewayMonitor();
3543

44+
/**
45+
* 注册设备网关监控供应商。
46+
*
47+
* @param supplier 监控供应商
48+
*/
3649
public static void register(DeviceGatewayMonitorSupplier supplier) {
3750
deviceGatewayMonitorSuppliers.add(supplier);
3851
}
@@ -54,6 +67,16 @@ private static DeviceGatewayMonitor doGetDeviceGatewayMonitor(String id, String.
5467
return monitor;
5568
}
5669

70+
/**
71+
* 获取指定设备网关的延迟监控实例。
72+
*
73+
* <p>首次调用监控 API 时才解析已注册供应商。多个供应商返回监控时，
74+
* 将按注册顺序组合执行。</p>
75+
*
76+
* @param id 设备网关标识
77+
* @param tags 网关附加标签
78+
* @return 延迟解析的设备网关监控
79+
*/
5780
public static DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags) {
5881
return new LazyDeviceGatewayMonitor(() -> doGetDeviceGatewayMonitor(id, tags));
5982
}

‎jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayTimeSeriesMetric.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,23 @@
1717

1818
import org.jetlinks.community.timeseries.TimeSeriesMetric;
1919

20+
/**
21+
* 网关监控使用的时序指标定义。
22+
*
23+
* <p>统一设备网关监控数据的时序存储名称，不负责指标采集和查询。</p>
24+
*
25+
* @see DeviceGatewayMonitor
26+
* @since 1.0
27+
*/
2028
public interface GatewayTimeSeriesMetric {
2129

2230
String deviceGatewayMetric = "device_gateway_monitor";
2331

24-
static TimeSeriesMetric deviceGatewayMetric(){
32+
/**
33+
* @return 网关设备监控指标
34+
* @see DeviceGatewayMonitor
35+
*/
36+
static TimeSeriesMetric deviceGatewayMetric() {
2537
return TimeSeriesMetric.of(deviceGatewayMetric);
2638
}
2739
}

0 commit comments

Comments
 (0)