diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java index 890eb9f2b..72c19d9c5 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/CompositeDeviceGatewayMonitor.java @@ -16,6 +16,14 @@ package org.jetlinks.community.gateway.monitor; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import javax.annotation.Nullable; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -24,7 +32,7 @@ class CompositeDeviceGatewayMonitor implements DeviceGatewayMonitor { - private List monitors = new ArrayList<>(); + private final List monitors = new ArrayList<>(); public CompositeDeviceGatewayMonitor add(DeviceGatewayMonitor... monitors) { return add(Arrays.asList(monitors)); @@ -69,4 +77,69 @@ public void receivedMessage() { public void sentMessage() { doWith(DeviceGatewayMonitor::sentMessage); } + + @Override + public boolean connected(ClientConnection connection) { + boolean accepted = true; + for (DeviceGatewayMonitor monitor : monitors) { + if (!monitor.connected(connection)) { + accepted = false; + } + } + return accepted; + } + + @Override + public void disconnected(ClientConnection connection) { + doWith(monitor -> monitor.disconnected(connection)); + } + + @Override + public void rejected(ClientConnection connection, @Nullable Throwable error) { + doWith(monitor -> monitor.rejected(connection, error)); + } + + @Override + public boolean beforeDecode(@Nullable ClientConnection connection, EncodedMessage message) { + boolean accepted = true; + for (DeviceGatewayMonitor monitor : monitors) { + if (!monitor.beforeDecode(connection, message)) { + accepted = false; + } + } + return accepted; + } + + @Override + public Flux decode(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + for (DeviceGatewayMonitor monitor : monitors) { + decoder = monitor.decode(connection, session, origin, decoder); + } + return decoder; + } + + @Override + public Flux beforeSendToPlatform(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + for (DeviceGatewayMonitor monitor : monitors) { + handler = monitor.beforeSendToPlatform(connection, session, origin, handler); + } + return handler; + } + + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + for (DeviceGatewayMonitor monitor : monitors) { + sender = monitor.downstream(connection, session, origin, sender); + } + return sender; + } } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java index e14bfd1df..3051b66b0 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitor.java @@ -15,8 +15,26 @@ */ package org.jetlinks.community.gateway.monitor; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.message.codec.FromDeviceMessageContext; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import javax.annotation.Nullable; +import java.util.function.UnaryOperator; + /** - * 设备网关监控 + * 设备网关监控扩展点。 + * + *

由设备网关在连接生命周期、报文解码和消息上下行阶段调用。实现可记录指标、拒绝连接或报文， + * 也可按注册顺序包装响应式处理链；实现不得在调用线程中阻塞或主动订阅传入的发布者。

+ * + * @see DeviceGatewayMonitorSupplier + * @see GatewayMonitors + * @since 1.0 */ public interface DeviceGatewayMonitor { @@ -24,32 +42,188 @@ public interface DeviceGatewayMonitor { * 上报总连接数 * * @param total 总连接数 + * @deprecated 使用 {@link #connected(ClientConnection)} 和 {@link #disconnected(ClientConnection)} 监听连接生命周期 */ - void totalConnection(long total); + @Deprecated + default void totalConnection(long total) { + + } /** * 创建新连接 */ - void connected(); + default void connected() { + + } /** * 拒绝连接 */ - void rejected(); + default void rejected() { + + } /** * 断开连接 */ - void disconnected(); + default void disconnected() { + + } + + /** + * 网关接收消息 + */ + default void receivedMessage() { + + } /** - * 接收消息 + * 网关发送消息 */ - void receivedMessage(); + default void sentMessage() { + + } /** - * 发送消息 + * 客户端长连接建立时执行。 + * + *

默认兼容调用 {@link #connected()}。返回 {@code false} 时网关将拒绝本次连接； + * 多个监控实现会全部执行，并合并各自的允许结果。

+ * + * @param connection 新建立的客户端连接 + * @return {@code true} 允许连接，{@code false} 拒绝连接 + * @since 2.12 + */ + default boolean connected(ClientConnection connection) { + connected(); + return true; + } + + /** + * 客户端长连接断开时执行。 + * + * @param connection 已断开的客户端连接 + * @since 2.12 + */ + default void disconnected(ClientConnection connection) { + disconnected(); + } + + /** + * 客户端连接被拒绝时执行，例如 MQTT 认证失败或 TCP 长时间未解析出设备。 + * + * @param connection 被拒绝的客户端连接 + * @param error 拒绝原因；无关联异常时为 {@code null} + * @since 2.12 + */ + default void rejected(ClientConnection connection, @Nullable Throwable error) { + rejected(); + } + + /** + * 原始报文进入协议解码前执行。 + * + *

短连接传输可能不提供连接对象。返回 {@code false} 时本次报文会被丢弃，且不会进入协议解码。

+ * + * @param connection 客户端连接，短连接场景可能为 {@code null} + * @param message 待解码的原始报文 + * @return {@code true} 继续解码，{@code false} 丢弃报文 + * @since 2.12 + */ + default boolean beforeDecode(@Nullable ClientConnection connection, + EncodedMessage message) { + return true; + } + + /** + * 包装设备上行报文的协议解码任务。 + * + *

实现应返回基于 {@code decoder} 组合出的发布者，保留原链路的背压、取消和错误信号， + * 不得在方法内主动订阅。

+ * + * @param connection 客户端连接，短连接场景可能为 {@code null} + * @param session 当前设备会话 + * @param origin 原始报文 + * @param decoder 协议解码任务 + * @return 包装后的解码任务 + * @since 2.12 + */ + default Flux decode(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + return decoder; + } + + /** + * 组装平台上行处理任务。 + * + *

本方法固定按 {@code platformHandler}、 + * {@link #beforeSendToPlatform(ClientConnection, DeviceSession, EncodedMessage, Flux)} 的顺序组装处理链。 + * 调用方应再使用 {@link #decode(ClientConnection, DeviceSession, EncodedMessage, Flux)} 包装返回任务， + * 使解码监控覆盖协议解码、平台处理和发送前处理的完整链路。 + * {@code platformHandler} 只能组合传入的解码任务，不得主动订阅。

+ * + *

{@link FromDeviceMessageContext#handleMessage(DeviceMessage)} 手动输出的消息不会进入协议返回的 + * {@code decoder}，调用方应使用本方法单独包装该消息的平台处理任务。协议随后返回空流表示没有额外的 + * 返回值消息，不能再次处理已经手动输出的消息。

+ * + * @param connection 客户端连接，短连接场景可能为 {@code null} + * @param session 当前设备会话 + * @param origin 原始报文 + * @param decoder 协议解码任务 + * @param platformHandler 将解码任务转换为包含平台消息处理逻辑的任务 + * @return 待解码监控包装的上行处理任务 + * @since 2.12 + * @see FromDeviceMessageContext + */ + default Flux handleUpstream(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder, + UnaryOperator> platformHandler) { + Flux handler = platformHandler.apply(decoder); + return beforeSendToPlatform(connection, session, origin, handler); + } + + /** + * 包装解码完成后、发送到平台前的上行处理任务。 + * + *

{@code handler} 已包含平台消息处理逻辑。实现应保留其背压、取消和错误信号， + * 不得在方法内主动订阅。

+ * + * @param connection 客户端连接，短连接场景可能为 {@code null} + * @param session 当前设备会话 + * @param origin 原始报文 + * @param handler 上行平台处理任务 + * @return 包装后的上行处理任务 + * @since 2.12 + */ + default Flux beforeSendToPlatform(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + return handler; + } + + /** + * 包装平台消息下发到设备的发送任务。 + * + *

实现应返回基于 {@code sender} 组合出的发布者，并保留原任务的取消和错误信号， + * 不得在方法内主动订阅。

+ * + * @param connection 当前客户端连接 + * @param session 当前设备会话 + * @param origin 待发送的原始报文 + * @param sender 原始发送任务 + * @return 包装后的发送任务 + * @since 2.12 */ - void sentMessage(); + default Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + return sender; + } } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java index f312b856e..40e9f8473 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorSupplier.java @@ -15,7 +15,28 @@ */ package org.jetlinks.community.gateway.monitor; +/** + * 设备网关监控供应商。 + * + *

通过 {@link GatewayMonitors#register(DeviceGatewayMonitorSupplier)} 注册后， + * 在设备网关首次使用监控能力时按网关标识创建监控实例。

+ * + * @see DeviceGatewayMonitor + * @see GatewayMonitors + * @since 1.0 + */ public interface DeviceGatewayMonitorSupplier { - DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags); + + /** + * 为指定设备网关创建监控实例。 + * + *

该方法可能由多个网关并发调用。返回 {@code null} 表示当前供应商不监控该网关； + * 返回的监控实例应遵循 {@link DeviceGatewayMonitor} 的非阻塞与响应式包装约束。

+ * + * @param id 设备网关标识 + * @param tags 网关附加标签 + * @return 监控实例，或 {@code null} 跳过当前供应商 + */ + DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags); } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java index 7511c08de..f322615c1 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayMonitors.java @@ -20,19 +20,32 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.stream.Collectors; +/** + * 设备网关监控注册与获取入口。 + * + *

供应商使用写时复制集合保存，支持在网关运行期间注册。返回的监控会延迟解析供应商， + * 以便网关可以早于监控组件创建。

+ * + * @see DeviceGatewayMonitor + * @see DeviceGatewayMonitorSupplier + * @since 1.0 + */ public class GatewayMonitors { - private static final List deviceGatewayMonitorSuppliers = new CopyOnWriteArrayList<>(); - static final NoneDeviceGatewayMonitor nonDevice = new NoneDeviceGatewayMonitor(); - - - static { - - } - + /** + * 未注册有效供应商时使用的空监控。 + * + * @since 2.12 + */ + public static final DeviceGatewayMonitor nonDevice = new NoneDeviceGatewayMonitor(); + /** + * 注册设备网关监控供应商。 + * + * @param supplier 监控供应商 + */ public static void register(DeviceGatewayMonitorSupplier supplier) { deviceGatewayMonitorSuppliers.add(supplier); } @@ -54,6 +67,16 @@ private static DeviceGatewayMonitor doGetDeviceGatewayMonitor(String id, String. return monitor; } + /** + * 获取指定设备网关的延迟监控实例。 + * + *

首次调用监控 API 时才解析已注册供应商。多个供应商返回监控时， + * 将按注册顺序组合执行。

+ * + * @param id 设备网关标识 + * @param tags 网关附加标签 + * @return 延迟解析的设备网关监控 + */ public static DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags) { return new LazyDeviceGatewayMonitor(() -> doGetDeviceGatewayMonitor(id, tags)); } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayTimeSeriesMetric.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayTimeSeriesMetric.java index b51b66a64..ba9e32899 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayTimeSeriesMetric.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/GatewayTimeSeriesMetric.java @@ -17,11 +17,23 @@ import org.jetlinks.community.timeseries.TimeSeriesMetric; +/** + * 网关监控使用的时序指标定义。 + * + *

统一设备网关监控数据的时序存储名称，不负责指标采集和查询。

+ * + * @see DeviceGatewayMonitor + * @since 1.0 + */ public interface GatewayTimeSeriesMetric { String deviceGatewayMetric = "device_gateway_monitor"; - static TimeSeriesMetric deviceGatewayMetric(){ + /** + * @return 网关设备监控指标 + * @see DeviceGatewayMonitor + */ + static TimeSeriesMetric deviceGatewayMetric() { return TimeSeriesMetric.of(deviceGatewayMetric); } } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/LazyDeviceGatewayMonitor.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/LazyDeviceGatewayMonitor.java index 70936c61c..ee023ba82 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/LazyDeviceGatewayMonitor.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/LazyDeviceGatewayMonitor.java @@ -15,13 +15,21 @@ */ package org.jetlinks.community.gateway.monitor; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import javax.annotation.Nullable; import java.util.function.Supplier; class LazyDeviceGatewayMonitor implements DeviceGatewayMonitor { private volatile DeviceGatewayMonitor target; - private Supplier monitorSupplier; + private final Supplier monitorSupplier; public LazyDeviceGatewayMonitor(Supplier monitorSupplier) { this.monitorSupplier = monitorSupplier; @@ -64,4 +72,48 @@ public void receivedMessage() { public void sentMessage() { getTarget().sentMessage(); } + + @Override + public boolean connected(ClientConnection connection) { + return getTarget().connected(connection); + } + + @Override + public void disconnected(ClientConnection connection) { + getTarget().disconnected(connection); + } + + @Override + public void rejected(ClientConnection connection, @Nullable Throwable error) { + getTarget().rejected(connection, error); + } + + @Override + public boolean beforeDecode(@Nullable ClientConnection connection, EncodedMessage message) { + return getTarget().beforeDecode(connection, message); + } + + @Override + public Flux decode(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + return getTarget().decode(connection, session, origin, decoder); + } + + @Override + public Flux beforeSendToPlatform(@Nullable ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + return getTarget().beforeSendToPlatform(connection, session, origin, handler); + } + + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + return getTarget().downstream(connection, session, origin, sender); + } } diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerDeviceGatewayMonitor.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerDeviceGatewayMonitor.java index dee601ff3..ab0ec16ce 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerDeviceGatewayMonitor.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerDeviceGatewayMonitor.java @@ -26,7 +26,7 @@ class MicrometerDeviceGatewayMonitor implements DeviceGatewayMonitor { String id; String[] tags; - private AtomicReference totalRef = new AtomicReference<>(0L); + private final AtomicReference totalRef = new AtomicReference<>(0L); public MicrometerDeviceGatewayMonitor(MeterRegistry registry, String id, String[] tags) { this.registry = registry; diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerGatewayMonitorSupplier.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerGatewayMonitorSupplier.java index ac2cfc086..4ab6809cf 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerGatewayMonitorSupplier.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/MicrometerGatewayMonitorSupplier.java @@ -18,6 +18,14 @@ import org.jetlinks.community.micrometer.MeterRegistryManager; import org.springframework.stereotype.Component; +/** + * 基于 Micrometer 的设备网关监控供应商。 + * + *

由 Spring 创建后注册到 {@link GatewayMonitors}，并为每个网关创建独立监控实例。

+ * + * @see MicrometerDeviceGatewayMonitor + * @since 1.0 + */ @Component public class MicrometerGatewayMonitorSupplier implements DeviceGatewayMonitorSupplier { @@ -26,10 +34,8 @@ public class MicrometerGatewayMonitorSupplier implements DeviceGatewayMonitorSup public MicrometerGatewayMonitorSupplier(MeterRegistryManager meterRegistryManager) { this.meterRegistryManager = meterRegistryManager; GatewayMonitors.register(this); - } - @Override public DeviceGatewayMonitor getDeviceGatewayMonitor(String id, String... tags) { return new MicrometerDeviceGatewayMonitor(meterRegistryManager.getMeterRegister(GatewayTimeSeriesMetric.deviceGatewayMetric, "target"), id, tags); diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurement.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurement.java index af1abc9b5..7506b1d0d 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurement.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurement.java @@ -38,13 +38,13 @@ class DeviceGatewayMeasurement extends StaticMeasurement { - private TimeSeriesManager timeSeriesManager; + private final TimeSeriesManager timeSeriesManager; - private String type; + private final String type; - private Aggregation defaultAgg; + private final Aggregation defaultAgg; - private String property; + private final String property; public DeviceGatewayMeasurement(MeasurementDefinition definition, String property, diff --git a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/GatewayObjectDefinition.java b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/GatewayObjectDefinition.java index de3f24740..2aa7780e6 100644 --- a/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/GatewayObjectDefinition.java +++ b/jetlinks-components/gateway-component/src/main/java/org/jetlinks/community/gateway/monitor/measurements/GatewayObjectDefinition.java @@ -19,6 +19,12 @@ import lombok.Getter; import org.jetlinks.community.dashboard.ObjectDefinition; +/** + * 设备网关监控的 Dashboard 对象类型定义。 + * + * @see DeviceGatewayMeasurementProvider + * @since 1.0 + */ @AllArgsConstructor @Getter public enum GatewayObjectDefinition implements ObjectDefinition { @@ -26,7 +32,7 @@ public enum GatewayObjectDefinition implements ObjectDefinition { session("会话") ; - private String name; + private final String name; @Override public String getId() { diff --git a/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorTest.java b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorTest.java new file mode 100644 index 000000000..b3a572f0d --- /dev/null +++ b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/DeviceGatewayMonitorTest.java @@ -0,0 +1,423 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.gateway.monitor; + +import org.jetlinks.core.device.DeviceRegistry; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.message.codec.FromDeviceMessageContext; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.UnaryOperator; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; + +class DeviceGatewayMonitorTest { + + @Test + void shouldWrapWholeUpstreamWithDecodeAndKeepPlatformHandlerLazy() { + List signals = new ArrayList<>(); + AtomicInteger platformHandled = new AtomicInteger(); + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Flux decode(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + signals.add("decode"); + return decoder.doOnNext(current -> { + signals.add("decodeOnNext"); + assertEquals(1, platformHandled.get()); + }); + } + + @Override + public Flux beforeSendToPlatform(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + signals.add("beforeSend"); + return handler; + } + }; + + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + DeviceMessage message = mock(DeviceMessage.class); + + Flux upstream = monitor.decode( + connection, + session, + origin, + monitor.handleUpstream( + connection, + session, + origin, + Flux.defer(() -> { + signals.add("decoder"); + return Flux.just(message); + }), + decoded -> { + signals.add("platformHandler"); + return decoded.concatMap(current -> Mono + .fromRunnable(() -> { + signals.add("platform"); + platformHandled.incrementAndGet(); + }) + .thenReturn(current)); + } + ) + ); + + assertEquals(Arrays.asList("platformHandler", "beforeSend", "decode"), signals); + assertEquals(0, platformHandled.get()); + + StepVerifier + .create(upstream) + .expectNext(message) + .verifyComplete(); + + assertEquals(1, platformHandled.get()); + assertEquals( + Arrays.asList( + "platformHandler", + "beforeSend", + "decode", + "decoder", + "platform", + "decodeOnNext"), + signals + ); + } + + @Test + void shouldWrapManualProtocolOutputWithExistingMonitorMethods() { + String monitorContextKey = DeviceGatewayMonitorTest.class.getName(); + AtomicInteger decode = new AtomicInteger(); + AtomicInteger beforeSend = new AtomicInteger(); + AtomicInteger monitored = new AtomicInteger(); + AtomicInteger manualHandled = new AtomicInteger(); + AtomicInteger returnedHandled = new AtomicInteger(); + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Flux decode(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + decode.incrementAndGet(); + return decoder; + } + + @Override + public Flux beforeSendToPlatform(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + beforeSend.incrementAndGet(); + return handler + .doOnNext(ignore -> monitored.incrementAndGet()) + .contextWrite(context -> context.put(monitorContextKey, true)); + } + }; + + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + DeviceRegistry registry = mock(DeviceRegistry.class); + DeviceMessage message = mock(DeviceMessage.class); + UnaryOperator> platformHandler = decoded -> decoded + .concatMap(current -> Mono.deferContextual(ctx -> { + assertTrue(ctx.getOrDefault(monitorContextKey, false)); + assertSame(message, current); + manualHandled.incrementAndGet(); + return Mono.just(current); + })); + FromDeviceMessageContext context = FromDeviceMessageContext.of( + session, + origin, + registry, + connection, + current -> monitor + .handleUpstream( + connection, + session, + origin, + Flux.just(current), + platformHandler) + .then() + ); + + Flux upstream = monitor.decode( + connection, + session, + origin, + monitor.handleUpstream( + connection, + session, + origin, + Flux.defer(() -> context.handleMessage(message).thenMany(Flux.empty())), + decoded -> decoded.concatMap(current -> Mono + .fromRunnable(returnedHandled::incrementAndGet) + .thenReturn(current)) + ) + ); + + StepVerifier + .create(upstream) + .verifyComplete(); + + assertEquals(1, decode.get()); + assertEquals(2, beforeSend.get()); + assertEquals(1, monitored.get()); + assertEquals(1, manualHandled.get()); + assertEquals(0, returnedHandled.get()); + } + + @Test + void shouldKeepDefaultMonitorCompatibleAndTransparent() { + AtomicInteger connected = new AtomicInteger(); + AtomicInteger disconnected = new AtomicInteger(); + AtomicInteger rejected = new AtomicInteger(); + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public void connected() { + connected.incrementAndGet(); + } + + @Override + public void disconnected() { + disconnected.incrementAndGet(); + } + + @Override + public void rejected() { + rejected.incrementAndGet(); + } + }; + + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + Flux decoder = Flux.empty(); + Mono sender = Mono.empty(); + + assertTrue(monitor.connected(connection)); + monitor.disconnected(connection); + monitor.rejected(connection, null); + assertTrue(monitor.beforeDecode(connection, origin)); + assertSame(decoder, monitor.decode(connection, session, origin, decoder)); + assertSame(decoder, monitor.beforeSendToPlatform(connection, session, origin, decoder)); + assertSame(sender, monitor.downstream(connection, session, origin, sender)); + assertEquals(1, connected.get()); + assertEquals(1, disconnected.get()); + assertEquals(1, rejected.get()); + } + + @Test + void shouldComposeAllMonitorDecisionsAndReactiveWrappersInOrder() { + List signals = new ArrayList<>(); + RecordingMonitor first = new RecordingMonitor("first", true, true, signals); + RecordingMonitor second = new RecordingMonitor("second", false, false, signals); + CompositeDeviceGatewayMonitor monitor = new CompositeDeviceGatewayMonitor() + .add(first, second); + + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + DeviceMessage message = mock(DeviceMessage.class); + + assertFalse(monitor.connected(connection)); + assertFalse(monitor.beforeDecode(connection, origin)); + monitor.rejected(connection, new IllegalStateException("rejected")); + monitor.disconnected(connection); + + StepVerifier + .create(monitor.beforeSendToPlatform( + connection, + session, + origin, + monitor.decode(connection, session, origin, Flux.just(message)))) + .expectNext(message) + .verifyComplete(); + + StepVerifier + .create(monitor.downstream( + connection, + session, + origin, + Mono.fromRunnable(() -> signals.add("sender")))) + .verifyComplete(); + + assertEquals( + Arrays.asList( + "first:connected", "second:connected", + "first:beforeDecode", "second:beforeDecode", + "first:rejected", "second:rejected", + "first:disconnected", "second:disconnected", + "first:decode", "second:decode", + "first:beforeSend", "second:beforeSend", + "sender", "first:downstream", "second:downstream" + ), + signals + ); + } + + @Test + void shouldLazilyResolveAndDelegateExtendedApi() { + List signals = new ArrayList<>(); + AtomicInteger resolved = new AtomicInteger(); + RecordingMonitor target = new RecordingMonitor("target", false, false, signals); + LazyDeviceGatewayMonitor monitor = new LazyDeviceGatewayMonitor(() -> { + resolved.incrementAndGet(); + return target; + }); + + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + DeviceMessage message = mock(DeviceMessage.class); + + assertFalse(monitor.connected(connection)); + assertFalse(monitor.beforeDecode(connection, origin)); + StepVerifier + .create(monitor.decode(connection, session, origin, Flux.just(message))) + .expectNext(message) + .verifyComplete(); + StepVerifier + .create(monitor.beforeSendToPlatform(connection, session, origin, Flux.just(message))) + .expectNext(message) + .verifyComplete(); + StepVerifier + .create(monitor.downstream(connection, session, origin, Mono.empty())) + .verifyComplete(); + + assertEquals(1, resolved.get()); + assertEquals( + Arrays.asList( + "target:connected", "target:beforeDecode", + "target:decode", "target:beforeSend", "target:downstream" + ), + signals + ); + } + + @Test + void shouldPropagateReactiveErrorsThroughCompositeMonitor() { + CompositeDeviceGatewayMonitor monitor = new CompositeDeviceGatewayMonitor() + .add( + new RecordingMonitor("first", true, true, new ArrayList<>()), + new RecordingMonitor("second", true, true, new ArrayList<>()) + ); + ClientConnection connection = mock(ClientConnection.class); + DeviceSession session = mock(DeviceSession.class); + EncodedMessage origin = mock(EncodedMessage.class); + IllegalStateException error = new IllegalStateException("failed"); + + Flux decoder = monitor.decode( + connection, + session, + origin, + Flux.error(error) + ); + StepVerifier + .create(monitor.beforeSendToPlatform(connection, session, origin, decoder)) + .expectErrorMatches(actual -> actual == error) + .verify(); + + StepVerifier + .create(monitor.downstream(connection, session, origin, Mono.error(error))) + .expectErrorMatches(actual -> actual == error) + .verify(); + } + + private static class RecordingMonitor implements DeviceGatewayMonitor { + + private final String id; + private final boolean allowConnection; + private final boolean allowDecode; + private final List signals; + + private RecordingMonitor(String id, + boolean allowConnection, + boolean allowDecode, + List signals) { + this.id = id; + this.allowConnection = allowConnection; + this.allowDecode = allowDecode; + this.signals = signals; + } + + @Override + public boolean connected(ClientConnection connection) { + signals.add(id + ":connected"); + return allowConnection; + } + + @Override + public void disconnected(ClientConnection connection) { + signals.add(id + ":disconnected"); + } + + @Override + public void rejected(ClientConnection connection, Throwable error) { + signals.add(id + ":rejected"); + } + + @Override + public boolean beforeDecode(ClientConnection connection, EncodedMessage message) { + signals.add(id + ":beforeDecode"); + return allowDecode; + } + + @Override + public Flux decode(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + return decoder.doOnNext(ignore -> signals.add(id + ":decode")); + } + + @Override + public Flux beforeSendToPlatform(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + return handler.doOnNext(ignore -> signals.add(id + ":beforeSend")); + } + + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + return sender.doOnSuccess(ignore -> signals.add(id + ":downstream")); + } + } +} diff --git a/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/GatewayMonitorsTest.java b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/GatewayMonitorsTest.java new file mode 100644 index 000000000..4c10850bb --- /dev/null +++ b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/GatewayMonitorsTest.java @@ -0,0 +1,112 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.gateway.monitor; + +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import org.jetlinks.community.micrometer.MeterRegistryManager; +import org.jetlinks.community.micrometer.MeterRegistrySupplier; +import org.jetlinks.core.metadata.DataType; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; + +class GatewayMonitorsTest { + + @Test + void shouldUseNoopMonitorUntilSupplierIsRegistered() { + DeviceGatewayMonitor monitor = GatewayMonitors.getDeviceGatewayMonitor("test"); + + assertSame(GatewayMonitors.nonDevice, ((LazyDeviceGatewayMonitor) monitor).getTarget()); + execute(monitor); + + SimpleMeterRegistry registry = new SimpleMeterRegistry(); + MeterRegistryManager manager = new MeterRegistryManager( + Arrays.asList(new TestMeterRegistrySupplier(registry)), + null + ); + + new MicrometerGatewayMonitorSupplier(manager); + + monitor = GatewayMonitors.getDeviceGatewayMonitor("test"); + assertNotSame(GatewayMonitors.nonDevice, ((LazyDeviceGatewayMonitor) monitor).getTarget()); + + monitor.totalConnection(2); + monitor.connected(); + monitor.rejected(); + monitor.disconnected(); + monitor.receivedMessage(); + monitor.sentMessage(); + + assertEquals(2, registry.get("test") + .tag("target", "connection") + .gauge() + .value()); + assertEquals(1, registry.get("test") + .tag("target", "connected") + .counter() + .count()); + assertEquals(1, registry.get("test") + .tag("target", "rejected") + .counter() + .count()); + assertEquals(1, registry.get("test") + .tag("target", "disconnected") + .counter() + .count()); + assertEquals(1, registry.get("test") + .tag("target", "received_message") + .counter() + .count()); + assertEquals(1, registry.get("test") + .tag("target", "sent_message") + .counter() + .count()); + } + + private void execute(DeviceGatewayMonitor monitor) { + monitor.connected(); + monitor.totalConnection(1); + monitor.rejected(); + monitor.disconnected(); + monitor.receivedMessage(); + monitor.sentMessage(); + } + + private static class TestMeterRegistrySupplier implements MeterRegistrySupplier { + + private final MeterRegistry registry; + + private TestMeterRegistrySupplier(MeterRegistry registry) { + this.registry = registry; + } + + @Override + public MeterRegistry getMeterRegistry(String metric, String... tagKeys) { + return registry; + } + + @Override + public MeterRegistry getMeterRegistry(String metric, Map tagDefine) { + return registry; + } + } +} diff --git a/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurementProviderTest.java b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurementProviderTest.java new file mode 100644 index 000000000..5c6e4f6f2 --- /dev/null +++ b/jetlinks-components/gateway-component/src/test/java/org/jetlinks/community/gateway/monitor/measurements/DeviceGatewayMeasurementProviderTest.java @@ -0,0 +1,72 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.gateway.monitor.measurements; + +import org.jetlinks.community.dashboard.CommonDimensionDefinition; +import org.jetlinks.community.dashboard.MeasurementParameter; +import org.jetlinks.community.timeseries.TimeSeriesData; +import org.jetlinks.community.timeseries.TimeSeriesManager; +import org.jetlinks.community.timeseries.TimeSeriesMetric; +import org.jetlinks.community.timeseries.TimeSeriesService; +import org.jetlinks.community.timeseries.query.AggregationData; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +class DeviceGatewayMeasurementProviderTest { + + @Test + void shouldQueryHistoryAndAggregationMeasurements() { + TimeSeriesManager timeSeriesManager = Mockito.mock(TimeSeriesManager.class); + TimeSeriesService timeSeriesService = Mockito.mock(TimeSeriesService.class); + + Mockito.when(timeSeriesManager.getService(Mockito.any(TimeSeriesMetric.class))) + .thenReturn(timeSeriesService); + + Map value = new HashMap<>(); + value.put("value", 100); + value.put("time", "2020-01-10"); + Mockito.when(timeSeriesService.query(Mockito.any())) + .thenReturn(Flux.just(TimeSeriesData.of(System.currentTimeMillis(), value))); + Mockito.when(timeSeriesService.aggregation(Mockito.any())) + .thenReturn(Flux.just(AggregationData.of(value))); + + DeviceGatewayMeasurementProvider provider = new DeviceGatewayMeasurementProvider(timeSeriesManager); + + provider + .getMeasurement("connection") + .flatMapMany(measurement -> measurement + .getDimension(CommonDimensionDefinition.history.getId()) + .flatMapMany(dimension -> dimension.getValue(MeasurementParameter.of(Collections.emptyMap())))) + .as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete(); + + provider + .getMeasurement("connection") + .flatMapMany(measurement -> measurement + .getDimension(CommonDimensionDefinition.agg.getId()) + .flatMapMany(dimension -> dimension.getValue(MeasurementParameter.of(Collections.emptyMap())))) + .as(StepVerifier::create) + .expectNextCount(1) + .verifyComplete(); + } +} diff --git a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java index 592fe0f68..4fc4b9741 100755 --- a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java +++ b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpDeviceSession.java @@ -16,6 +16,7 @@ package org.jetlinks.community.network.http.device; import lombok.Setter; +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; import org.jetlinks.core.device.DeviceOperator; import org.jetlinks.core.enums.ErrorCode; import org.jetlinks.core.exception.DeviceOperationException; @@ -49,12 +50,17 @@ class HttpDeviceSession implements DeviceSession { @Setter private WebSocketExchange websocket; + private final DeviceGatewayMonitor monitor; + private long lastPingTime = System.currentTimeMillis(); //默认永不超时 private long keepAliveTimeOutMs = -1; - public HttpDeviceSession(DeviceOperator deviceOperator, InetSocketAddress address) { + public HttpDeviceSession(DeviceGatewayMonitor monitor, + DeviceOperator deviceOperator, + InetSocketAddress address) { + this.monitor = monitor; this.operator = deviceOperator; this.address = address; } @@ -94,15 +100,17 @@ public Mono send(EncodedMessage encodedMessage) { if (!websocket.isAlive()) { return Mono.error(new DeviceOperationException.NoStackTrace(ErrorCode.CONNECTION_LOST)); } + Mono sender; if (encodedMessage instanceof WebSocketMessage) { - return websocket - .send(((WebSocketMessage) encodedMessage)) - .thenReturn(true); + sender = websocket.send(((WebSocketMessage) encodedMessage)); } else { - return websocket - .send(DefaultWebSocketMessage.of(WebSocketMessage.Type.TEXT, encodedMessage.getPayload())) - .thenReturn(true); + sender = websocket.send( + DefaultWebSocketMessage.of(WebSocketMessage.Type.TEXT, encodedMessage.getPayload())); } + sender = sender.doOnSuccess(ignore -> monitor.sentMessage()); + return monitor + .downstream(websocket, this, encodedMessage, sender) + .thenReturn(true); } @Override diff --git a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpServerDeviceGateway.java b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpServerDeviceGateway.java index a077cd3d6..671e9a616 100755 --- a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpServerDeviceGateway.java +++ b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/HttpServerDeviceGateway.java @@ -47,6 +47,7 @@ import org.springframework.http.HttpStatus; import org.springframework.util.StringUtils; import reactor.core.Disposable; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.net.InetSocketAddress; @@ -54,6 +55,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.UnaryOperator; /** * Http 服务设备网关，使用指定的协议包，将网络组件中Http服务的请求处理为设备消息 @@ -112,6 +114,9 @@ private Disposable handleWebsocketRequest(String url) { private Mono handleWebsocketRequest(WebSocketExchange exchange) { + if (!monitor.connected(exchange)) { + return exchange.close(); + } return protocol .flatMap(protocol -> protocol @@ -120,6 +125,7 @@ private Mono handleWebsocketRequest(WebSocketExchange exchange) { .flatMap(result -> { if (result.isSuccess()) { String deviceId = result.getDeviceId(); + exchange.closeHandler(() -> monitor.disconnected(exchange)); if (StringUtils.hasText(deviceId)) { DeviceOnlineMessage message = new DeviceOnlineMessage(); message.setDeviceId(deviceId); @@ -136,6 +142,7 @@ private Mono handleWebsocketRequest(WebSocketExchange exchange) { .then(); } } else { + monitor.rejected(exchange, null); log.warn("设备[{}] Websocket 认证失败:{}", exchange .getRemoteAddress() .orElse(null), result.getMessage()); @@ -154,10 +161,19 @@ private Mono handleWebsocketRequest(WebSocketExchange exchange, WebSocketM return Mono.empty(); }); } - WebSocketDeviceSession session = new WebSocketDeviceSession(device, exchange); + if (!monitor.beforeDecode(exchange, msg)) { + return Mono.empty(); + } - return protocol - .flatMap(protocol -> { + WebSocketDeviceSession session = new WebSocketDeviceSession(monitor, device, exchange); + + UnaryOperator> platformHandler = task -> task + .concatMap(deviceMessage -> handleWebsocketMessage(deviceMessage, exchange, session) + .doOnNext(session::setOperator) + .thenReturn(deviceMessage)); + + Flux decodeTask = protocol + .flatMapMany(protocol -> { if (log.isDebugEnabled()) { log.debug("收到HTTP请求\n{}", msg); } @@ -165,18 +181,38 @@ private Mono handleWebsocketRequest(WebSocketExchange exchange, WebSocketM return protocol .getMessageCodec(DefaultTransport.WebSocket) .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of( - session, msg, registry, deviceMessage -> handleWebsocketMessage(deviceMessage, exchange, session).then()))) - .cast(DeviceMessage.class) - .concatMap(deviceMessage -> handleWebsocketMessage(deviceMessage, exchange, session)) - .doOnNext(session::setOperator) - .onErrorResume(err -> { - log.error("处理http请求失败:\n{}", msg, err); - return exchange - .close(HttpStatus.BAD_REQUEST) - .then(Mono.empty()); - }) - .then(); + session, + msg, + registry, + // 手动输出不进入 codec 返回值，单独复用同一平台处理与发送前监控链。 + deviceMessage -> monitor + .handleUpstream( + exchange, + session, + msg, + Flux.just(deviceMessage), + platformHandler) + .then()))) + .cast(DeviceMessage.class); + }); + + decodeTask = monitor.handleUpstream( + exchange, + session, + msg, + decodeTask, + platformHandler + ); + decodeTask = monitor.decode(exchange, session, msg, decodeTask); + + return decodeTask + .onErrorResume(err -> { + log.error("处理http请求失败:\n{}", msg, err); + return exchange + .close(HttpStatus.BAD_REQUEST) + .then(Mono.empty()); }) + .then() .as(MonoTracer.create("http-device-gateway/" + getId() + exchange.getPath())) .onErrorResume((error) -> { log.error(error.getMessage(), error); @@ -198,7 +234,7 @@ private Mono handleWebsocketMessage(DeviceMessage message, return helper .handleDeviceMessage( message, - device -> new WebSocketDeviceSession(device, exchange), + device -> new WebSocketDeviceSession(monitor, device, exchange), deviceSession -> { if (deviceSession.isWrapFrom(WebSocketDeviceSession.class)) { deviceSession @@ -231,22 +267,41 @@ private Mono handleHttpRequest(HttpExchange exchange) { if (log.isDebugEnabled()) { log.debug("收到HTTP请求\n{}", httpMessage); } - InetSocketAddress address = exchange.request().getClientAddress(); UnknownHttpDeviceSession session = new UnknownHttpDeviceSession(exchange); + if (!monitor.beforeDecode(null, httpMessage)) { + return completeHttpRequest(exchange); + } + UnaryOperator> platformHandler = task -> task + .concatMap(deviceMessage -> + handleMessage(deviceMessage, exchange, httpMessage) + .thenReturn(deviceMessage)); //调用协议执行解码 - return protocol + Flux decodeTask = protocol .getMessageCodec(getTransport()) .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of( - session, httpMessage, registry, msg -> handleMessage(msg, exchange, httpMessage)))) - .cast(DeviceMessage.class) - .concatMap(deviceMessage -> handleMessage(deviceMessage, exchange, httpMessage)) - .then(Mono.defer(() -> { - //如果协议包里没有回复，那就响应200 - if (!exchange.isClosed()) { - return exchange.ok(); - } - return Mono.empty(); - })) + session, + httpMessage, + registry, + // 手动输出不进入 codec 返回值，单独复用同一平台处理与发送前监控链。 + deviceMessage -> monitor + .handleUpstream( + null, + session, + httpMessage, + Flux.just(deviceMessage), + platformHandler) + .then()))) + .cast(DeviceMessage.class); + decodeTask = monitor.handleUpstream( + null, + session, + httpMessage, + decodeTask, + platformHandler + ); + decodeTask = monitor.decode(null, session, httpMessage, decodeTask); + return decodeTask + .then(completeHttpRequest(exchange)) .onErrorResume(err -> { log.error("处理http请求失败:\n{}", httpMessage, err); return response500Error(exchange, err); @@ -260,6 +315,16 @@ private Mono handleHttpRequest(HttpExchange exchange) { }); } + private Mono completeHttpRequest(HttpExchange exchange) { + return Mono.defer(() -> { + // 如果协议包没有主动响应，则使用 200 结束本次短连接请求。 + if (!exchange.isClosed()) { + return exchange.ok(); + } + return Mono.empty(); + }); + } + private Mono handleMessage(DeviceMessage deviceMessage, HttpExchange exchange, HttpExchangeMessage message) { @@ -269,7 +334,7 @@ private Mono handleMessage(DeviceMessage deviceMessage, monitor.receivedMessage(); return helper .handleDeviceMessage(deviceMessage, - device -> new HttpDeviceSession(device, address), + device -> new HttpDeviceSession(monitor, device, address), ignore -> { }, () -> { diff --git a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/WebSocketDeviceSession.java b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/WebSocketDeviceSession.java index 9a0bd38b6..b1594581a 100755 --- a/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/WebSocketDeviceSession.java +++ b/jetlinks-components/network-component/http-component/src/main/java/org/jetlinks/community/network/http/device/WebSocketDeviceSession.java @@ -18,6 +18,7 @@ import lombok.AllArgsConstructor; import lombok.Getter; import lombok.Setter; +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; import org.jetlinks.core.device.DeviceOperator; import org.jetlinks.core.message.codec.DefaultTransport; import org.jetlinks.core.message.codec.EncodedMessage; @@ -47,11 +48,16 @@ class WebSocketDeviceSession implements DeviceSession { @Setter private WebSocketExchange exchange; + private final DeviceGatewayMonitor monitor; + private final long connectTime = System.currentTimeMillis(); private Duration keepAliveTimeout; - public WebSocketDeviceSession(DeviceOperator device, WebSocketExchange exchange) { + public WebSocketDeviceSession(DeviceGatewayMonitor monitor, + DeviceOperator device, + WebSocketExchange exchange) { + this.monitor = monitor; this.operator = device; this.exchange = exchange; } @@ -78,16 +84,17 @@ public long connectTime() { @Override public Mono send(EncodedMessage encodedMessage) { + Mono sender; if (encodedMessage instanceof WebSocketMessage) { - return exchange - .send(((WebSocketMessage) encodedMessage)) - .thenReturn(true); + sender = exchange.send(((WebSocketMessage) encodedMessage)); } else { - return exchange - .send(DefaultWebSocketMessage.of(WebSocketMessage.Type.TEXT, encodedMessage.getPayload())) - .thenReturn(true); + sender = exchange.send( + DefaultWebSocketMessage.of(WebSocketMessage.Type.TEXT, encodedMessage.getPayload())); } - + sender = sender.doOnSuccess(ignore -> monitor.sentMessage()); + return monitor + .downstream(exchange, this, encodedMessage, sender) + .thenReturn(true); } @Override @@ -131,7 +138,7 @@ public Duration getKeepAliveTimeout() { } public WebSocketDeviceSession copy() { - WebSocketDeviceSession session = new WebSocketDeviceSession(operator, exchange); + WebSocketDeviceSession session = new WebSocketDeviceSession(monitor, operator, exchange); session.setKeepAliveTimeout(keepAliveTimeout); return session; diff --git a/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/HttpServerDeviceGatewayTest.java b/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/HttpServerDeviceGatewayTest.java new file mode 100644 index 000000000..37ec74002 --- /dev/null +++ b/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/HttpServerDeviceGatewayTest.java @@ -0,0 +1,164 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.network.http.device; + +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; +import org.jetlinks.community.gateway.monitor.GatewayMonitors; +import org.jetlinks.community.network.http.server.HttpExchange; +import org.jetlinks.community.network.http.server.HttpRequest; +import org.jetlinks.community.network.http.server.HttpServer; +import org.jetlinks.core.ProtocolSupport; +import org.jetlinks.core.device.DeviceRegistry; +import org.jetlinks.core.device.session.DeviceSessionManager; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.DefaultTransport; +import org.jetlinks.core.message.codec.DeviceMessageCodec; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.message.codec.FromDeviceMessageContext; +import org.jetlinks.core.message.codec.http.HttpExchangeMessage; +import org.jetlinks.core.route.HttpRoute; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import org.jetlinks.supports.server.DecodedClientMessageHandler; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpMethod; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.test.StepVerifier; + +import java.time.Duration; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class HttpServerDeviceGatewayTest { + + @Test + void manualHttpUpstreamShouldUseMonitorChain() throws InterruptedException { + String gatewayId = "http-monitor-test"; + RecordingMonitor monitor = new RecordingMonitor(); + GatewayMonitors.register((id, tags) -> gatewayId.equals(id) ? monitor : null); + + HttpServer server = mock(HttpServer.class); + DeviceRegistry registry = mock(DeviceRegistry.class); + DeviceSessionManager sessionManager = mock(DeviceSessionManager.class); + DecodedClientMessageHandler messageHandler = mock(DecodedClientMessageHandler.class); + ProtocolSupport protocol = mock(ProtocolSupport.class); + DeviceMessageCodec codec = mock(DeviceMessageCodec.class); + HttpRoute route = mock(HttpRoute.class); + HttpExchange exchange = mock(HttpExchange.class); + HttpExchangeMessage message = mock(HttpExchangeMessage.class); + DeviceMessage deviceMessage = mock(DeviceMessage.class); + HttpRequest request = mock(HttpRequest.class); + Sinks.Many requests = Sinks.many().unicast().onBackpressureBuffer(); + + when(route.getMethod()).thenReturn(new HttpMethod[]{HttpMethod.POST}); + when(route.getAddress()).thenReturn("/test"); + when(protocol.getRoutes(DefaultTransport.HTTP)).thenReturn(Flux.just(route)); + when(protocol.getRoutes(DefaultTransport.WebSocket)).thenReturn(Flux.empty()); + when(protocol.getMessageCodec(DefaultTransport.HTTP)) + .thenAnswer(ignore -> Mono.just(codec)); + when(codec.decode(any())).thenAnswer(invocation -> { + FromDeviceMessageContext context = invocation.getArgument(0); + return context + .handleMessage(deviceMessage) + .thenMany(Flux.empty()); + }); + when(server.handleRequest(HttpMethod.POST, "/test")).thenReturn(requests.asFlux()); + when(exchange.toExchangeMessage()).thenReturn(Mono.just(message)); + when(exchange.request()).thenReturn(request); + when(request.getPath()).thenReturn("/test"); + when(exchange.ok()).thenReturn(Mono.empty()); + + HttpServerDeviceGateway gateway = new HttpServerDeviceGateway( + gatewayId, + server, + Mono.just(protocol), + sessionManager, + registry, + messageHandler + ); + StepVerifier.create(gateway.startup()).verifyComplete(); + + assertEquals(Sinks.EmitResult.OK, requests.tryEmitNext(exchange)); + assertTrue(monitor.completed.await(Duration.ofSeconds(5).toMillis(), java.util.concurrent.TimeUnit.MILLISECONDS)); + + assertEquals(1, monitor.beforeDecode.get()); + assertEquals(1, monitor.decode.get()); + assertEquals(2, monitor.beforeSend.get()); + assertEquals(1, monitor.received.get()); + assertEquals(List.of(deviceMessage), monitor.monitored); + assertSame(message, monitor.origin); + + StepVerifier.create(gateway.shutdown()).verifyComplete(); + } + + private static class RecordingMonitor implements DeviceGatewayMonitor { + private final AtomicInteger beforeDecode = new AtomicInteger(); + private final AtomicInteger decode = new AtomicInteger(); + private final AtomicInteger beforeSend = new AtomicInteger(); + private final AtomicInteger received = new AtomicInteger(); + private final CountDownLatch completed = new CountDownLatch(1); + private final List monitored = new CopyOnWriteArrayList<>(); + private EncodedMessage origin; + + @Override + public void receivedMessage() { + received.incrementAndGet(); + } + + @Override + public boolean beforeDecode(ClientConnection connection, EncodedMessage message) { + assertNull(connection); + beforeDecode.incrementAndGet(); + origin = message; + return true; + } + + @Override + public Flux decode(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + assertNull(connection); + decode.incrementAndGet(); + return decoder; + } + + @Override + public Flux beforeSendToPlatform(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + assertNull(connection); + beforeSend.incrementAndGet(); + return handler.doOnNext(message -> { + monitored.add(message); + completed.countDown(); + }); + } + } +} diff --git a/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/WebSocketDeviceSessionTest.java b/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/WebSocketDeviceSessionTest.java new file mode 100644 index 000000000..722aeed5f --- /dev/null +++ b/jetlinks-components/network-component/http-component/src/test/java/org/jetlinks/community/network/http/device/WebSocketDeviceSessionTest.java @@ -0,0 +1,117 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.network.http.device; + +import io.netty.buffer.Unpooled; +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; +import org.jetlinks.community.network.http.server.WebSocketExchange; +import org.jetlinks.core.device.DeviceOperator; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.message.codec.http.websocket.DefaultWebSocketMessage; +import org.jetlinks.core.message.codec.http.websocket.WebSocketMessage; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class WebSocketDeviceSessionTest { + + @Test + void downstreamMonitorShouldWrapSender() { + WebSocketExchange exchange = mock(WebSocketExchange.class); + WebSocketMessage message = DefaultWebSocketMessage.of( + WebSocketMessage.Type.TEXT, Unpooled.EMPTY_BUFFER); + RuntimeException error = new RuntimeException("rejected by monitor"); + AtomicInteger senderSubscriptions = new AtomicInteger(); + AtomicInteger monitorCalls = new AtomicInteger(); + + when(exchange.send(message)).thenReturn(Mono.defer(() -> { + senderSubscriptions.incrementAndGet(); + return Mono.empty(); + })); + + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + assertSame(exchange, connection); + assertSame(message, origin); + monitorCalls.incrementAndGet(); + return Mono.error(error); + } + }; + WebSocketDeviceSession session = new WebSocketDeviceSession(monitor, null, exchange); + + StepVerifier + .create(session.send(message)) + .expectErrorSatisfies(actual -> assertSame(error, actual)) + .verify(); + + assertEquals(1, monitorCalls.get()); + assertEquals(0, senderSubscriptions.get()); + } + + @Test + void httpSessionWebSocketShouldUseDownstreamMonitor() { + DeviceOperator operator = mock(DeviceOperator.class); + WebSocketExchange exchange = mock(WebSocketExchange.class); + WebSocketMessage message = DefaultWebSocketMessage.of( + WebSocketMessage.Type.TEXT, Unpooled.EMPTY_BUFFER); + RuntimeException error = new RuntimeException("rejected by monitor"); + AtomicInteger senderSubscriptions = new AtomicInteger(); + AtomicInteger monitorCalls = new AtomicInteger(); + + when(operator.getDeviceId()).thenReturn("device-1"); + when(exchange.isAlive()).thenReturn(true); + when(exchange.send(message)).thenReturn(Mono.defer(() -> { + senderSubscriptions.incrementAndGet(); + return Mono.empty(); + })); + + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + assertSame(exchange, connection); + assertSame(message, origin); + monitorCalls.incrementAndGet(); + return Mono.error(error); + } + }; + HttpDeviceSession session = new HttpDeviceSession(monitor, operator, null); + session.setWebsocket(exchange); + + StepVerifier + .create(session.send(message)) + .expectErrorSatisfies(actual -> assertSame(error, actual)) + .verify(); + + assertEquals(1, monitorCalls.get()); + assertEquals(0, senderSubscriptions.get()); + } +} diff --git a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/DeviceMqttConnection.java b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/DeviceMqttConnection.java index 9e3f7f1ef..9462be8ba 100644 --- a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/DeviceMqttConnection.java +++ b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/DeviceMqttConnection.java @@ -47,6 +47,7 @@ import reactor.core.CoreSubscriber; import reactor.core.Disposable; import reactor.core.Disposables; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.Operators; import reactor.core.scheduler.Schedulers; @@ -55,6 +56,7 @@ import java.net.InetSocketAddress; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.UnaryOperator; import static org.jetlinks.community.network.mqtt.gateway.device.MqttServerDeviceGateway.clientId; @@ -84,6 +86,7 @@ void doAuth() { MqttAuth auth = connection.getAuth().orElse(null); if (auth == null || !StringUtils.hasText(connection.getClientId())) { reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED); + monitor.rejected(connection, null); } else { doAuth(auth); } @@ -214,6 +217,14 @@ protected Mono decodeAndHandleMessage(MqttMessage message) { if (operator == null) { return Mono.empty(); } + + if (!monitor.beforeDecode(connection, message)) { + return Mono.empty(); + } + + UnaryOperator> platformHandler = task -> task + .concatMap(this::handleMessage, 0); + // 上下文 FromDeviceMessageContext context = FromDeviceMessageContext @@ -221,20 +232,39 @@ protected Mono decodeAndHandleMessage(MqttMessage message) { message, helper.getRegistry(), connection, - this); - - return operator + // 手动输出不进入 codec 返回值，单独复用同一平台处理与发送前监控链。 + deviceMessage -> monitor + .handleUpstream( + connection, + session, + message, + Flux.just(deviceMessage), + platformHandler) + .then()); + + Flux decodeTask = operator .getProtocol() .flatMap(protocol -> protocol.getMessageCodec(getTransport())) //解码 .flatMapMany(codec -> codec.decode(context)) - .cast(DeviceMessage.class) - .concatMap(this::handleMessage, 0) - .doOnComplete(() -> { - if (message instanceof MqttPublishing) { - ((MqttPublishing) message).acknowledge(); - } - }) + .cast(DeviceMessage.class); + + decodeTask = monitor.handleUpstream( + connection, + session, + message, + decodeTask, + task -> platformHandler + .apply(task) + .doOnComplete(() -> { + if (message instanceof MqttPublishing) { + ((MqttPublishing) message).acknowledge(); + } + }) + ); + decodeTask = monitor.decode(connection, session, message, decodeTask); + + return decodeTask .as(FluxTracer .create(DeviceTracer.SpanName.decode0(operator.getDeviceId()), (span) -> span @@ -257,6 +287,7 @@ protected Mono decodeAndHandleMessage(MqttMessage message) { } private Mono handleMessage(DeviceMessage message) { + monitor.receivedMessage(); DeviceOperator mainDevice = session.getOperator(); @@ -355,6 +386,7 @@ void doAuth(MqttAuth auth) { if (err instanceof AuthenticationException) { reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD); } else { + monitor.rejected(connection, err); log.warn("MQTT连接认证[{}]失败", connection.getClientId(), err); //应答SERVER_UNAVAILABLE reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); @@ -408,6 +440,7 @@ public void cancel() { if (actual != null) { actual.onComplete(); } + monitor.disconnected(connection); } } diff --git a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java index 4626c62df..ec3b1a250 100755 --- a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java +++ b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGateway.java @@ -36,6 +36,7 @@ import org.jetlinks.community.gateway.DeviceGatewayHelper; import org.jetlinks.supports.server.DecodedClientMessageHandler; import reactor.core.Disposable; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; import reactor.util.function.Tuple2; @@ -43,6 +44,7 @@ import java.util.*; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.UnaryOperator; /** * MQTT Client 设备网关，使用网络组件中的MQTT Client来处理设备数据 @@ -151,15 +153,7 @@ protected Disposable doSubscribe(String topic, int qos) { return mqttClient .subscribe(Collections.singletonList(topic), qos) .filter(msg -> isStarted()) - .flatMap(mqttMessage -> codecMono - .flatMapMany(codec -> codec - .decode(FromDeviceMessageContext.of( - new UnknownDeviceMqttClientSession(getId(), mqttClient, monitor), - mqttMessage, - registry, - msg -> handleMessage(mqttMessage, msg).then()))) - .cast(DeviceMessage.class) - .concatMap(message -> handleMessage(mqttMessage, message)) + .flatMap(mqttMessage -> decodeAndHandleMessage(mqttMessage) .subscribeOn(Schedulers.parallel()) .onErrorResume((err) -> { log.error("handle mqtt client message error:{}", mqttMessage, err); @@ -169,6 +163,40 @@ protected Disposable doSubscribe(String topic, int qos) { .subscribe(); } + private Mono decodeAndHandleMessage(MqttMessage mqttMessage) { + if (!monitor.beforeDecode(null, mqttMessage)) { + return Mono.empty(); + } + UnknownDeviceMqttClientSession session = + new UnknownDeviceMqttClientSession(getId(), mqttClient, monitor); + UnaryOperator> platformHandler = task -> task + .concatMap(message -> handleMessage(mqttMessage, message).thenReturn(message)); + Flux decodeTask = codecMono + .flatMapMany(codec -> codec.decode(FromDeviceMessageContext.of( + session, + mqttMessage, + registry, + // 手动输出不进入 codec 返回值，单独复用同一平台处理与发送前监控链。 + message -> monitor + .handleUpstream( + null, + session, + mqttMessage, + Flux.just(message), + platformHandler) + .then()))) + .cast(DeviceMessage.class); + decodeTask = monitor.handleUpstream( + null, + session, + mqttMessage, + decodeTask, + platformHandler + ); + decodeTask = monitor.decode(null, session, mqttMessage, decodeTask); + return decodeTask.then(); + } + private Mono handleMessage(MqttMessage mqttMessage, DeviceMessage message) { monitor.receivedMessage(); return helper diff --git a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java index e92d784e7..f0b898c6b 100755 --- a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java +++ b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/MqttServerDeviceGateway.java @@ -31,6 +31,7 @@ import org.jetlinks.community.gateway.AbstractDeviceGateway; import org.jetlinks.community.gateway.DeviceGateway; import org.jetlinks.community.gateway.DeviceGatewayHelper; +import org.jetlinks.community.network.mqtt.server.MqttConnection; import org.jetlinks.community.network.mqtt.server.MqttServer; import org.jetlinks.supports.server.DecodedClientMessageHandler; import reactor.core.Disposable; @@ -90,7 +91,7 @@ private synchronized void doStart() { if (!isStarted()) { //直接响应SERVER_UNAVAILABLE conn.reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); - monitor.rejected(); + monitor.rejected(conn, null); } return true; }) @@ -101,8 +102,12 @@ private synchronized void doStart() { } - protected Mono handleConnection0(org.jetlinks.community.network.mqtt.server.MqttConnection connection) { - return new DeviceMqttConnection(helper,monitor,connection); + protected Mono handleConnection0(MqttConnection connection) { + if (!monitor.connected(connection)) { + connection.reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); + return Mono.empty(); + } + return new DeviceMqttConnection(helper, monitor, connection); } @Override diff --git a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java index 3b112d016..83fe88f5c 100755 --- a/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java +++ b/jetlinks-components/network-component/mqtt-component/src/main/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSession.java @@ -169,9 +169,11 @@ public Mono send(EncodedMessage encodedMessage) { if (connection == null) { return Mono.error(new DeviceOperationException.NoStackTrace(ErrorCode.CONNECTION_LOST)); } - return Mono + Mono sender = Mono .defer(() -> connection.publish(((MqttMessage) encodedMessage))) - .doOnSuccess(nil -> monitor.sentMessage()) + .doOnSuccess(nil -> monitor.sentMessage()); + return monitor + .downstream(connection, this, encodedMessage, sender) .thenReturn(true); } diff --git a/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGatewayTest.java b/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGatewayTest.java new file mode 100644 index 000000000..3c5b1c4f9 --- /dev/null +++ b/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/MqttClientDeviceGatewayTest.java @@ -0,0 +1,164 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.network.mqtt.gateway.device; + +import io.netty.buffer.Unpooled; +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; +import org.jetlinks.community.gateway.monitor.GatewayMonitors; +import org.jetlinks.community.network.mqtt.client.MqttClient; +import org.jetlinks.core.ProtocolSupport; +import org.jetlinks.core.device.DeviceRegistry; +import org.jetlinks.core.device.session.DeviceSessionManager; +import org.jetlinks.core.message.DeviceMessage; +import org.jetlinks.core.message.codec.DefaultTransport; +import org.jetlinks.core.message.codec.DeviceMessageCodec; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.message.codec.FromDeviceMessageContext; +import org.jetlinks.core.message.codec.SimpleMqttMessage; +import org.jetlinks.core.route.MqttRoute; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import org.jetlinks.supports.server.DecodedClientMessageHandler; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.test.StepVerifier; + +import java.time.Duration; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class MqttClientDeviceGatewayTest { + + @Test + void manualAndReturnedMqttUpstreamShouldUseMonitorChainOnce() throws InterruptedException { + String gatewayId = "mqtt-client-monitor-test"; + RecordingMonitor monitor = new RecordingMonitor(); + GatewayMonitors.register((id, tags) -> gatewayId.equals(id) ? monitor : null); + + MqttClient client = mock(MqttClient.class); + DeviceRegistry registry = mock(DeviceRegistry.class); + DeviceSessionManager sessionManager = mock(DeviceSessionManager.class); + DecodedClientMessageHandler messageHandler = mock(DecodedClientMessageHandler.class); + ProtocolSupport protocol = mock(ProtocolSupport.class); + DeviceMessageCodec codec = mock(DeviceMessageCodec.class); + DeviceMessage manualMessage = mock(DeviceMessage.class); + DeviceMessage returnedMessage = mock(DeviceMessage.class); + MqttRoute route = mock(MqttRoute.class); + Sinks.Many messages = + Sinks.many().unicast().onBackpressureBuffer(); + + when(route.isUpstream()).thenReturn(true); + when(route.getTopic()).thenReturn("/test"); + when(route.getQos()).thenReturn(0); + when(protocol.getRoutes(DefaultTransport.MQTT)).thenReturn(Flux.just(route)); + when(protocol.getMessageCodec(DefaultTransport.MQTT)) + .thenAnswer(ignore -> Mono.just(codec)); + when(codec.decode(any())).thenAnswer(invocation -> { + FromDeviceMessageContext context = invocation.getArgument(0); + return context + .handleMessage(manualMessage) + .thenMany(Flux.just(returnedMessage)); + }); + when(client.subscribe(Collections.singletonList("/test"), 0)).thenReturn(messages.asFlux()); + + MqttClientDeviceGateway gateway = new MqttClientDeviceGateway( + gatewayId, + client, + registry, + Mono.just(protocol), + sessionManager, + messageHandler + ); + StepVerifier.create(gateway.startup()).verifyComplete(); + + SimpleMqttMessage message = SimpleMqttMessage + .builder() + .topic("/test") + .payload(Unpooled.EMPTY_BUFFER) + .build(); + assertEquals(Sinks.EmitResult.OK, messages.tryEmitNext(message)); + assertTrue(monitor.completed.await(Duration.ofSeconds(5).toMillis(), java.util.concurrent.TimeUnit.MILLISECONDS)); + + assertEquals(1, monitor.beforeDecode.get()); + assertEquals(1, monitor.decode.get()); + assertEquals(2, monitor.beforeSend.get()); + assertEquals(2, monitor.received.get()); + assertEquals(1, Collections.frequency(monitor.monitored, manualMessage)); + assertEquals(1, Collections.frequency(monitor.monitored, returnedMessage)); + assertSame(message, monitor.origin); + + StepVerifier.create(gateway.shutdown()).verifyComplete(); + } + + private static class RecordingMonitor implements DeviceGatewayMonitor { + private final AtomicInteger beforeDecode = new AtomicInteger(); + private final AtomicInteger decode = new AtomicInteger(); + private final AtomicInteger beforeSend = new AtomicInteger(); + private final AtomicInteger received = new AtomicInteger(); + private final CountDownLatch completed = new CountDownLatch(2); + private final List monitored = new CopyOnWriteArrayList<>(); + private EncodedMessage origin; + + @Override + public void receivedMessage() { + received.incrementAndGet(); + } + + @Override + public boolean beforeDecode(ClientConnection connection, EncodedMessage message) { + assertNull(connection); + beforeDecode.incrementAndGet(); + origin = message; + return true; + } + + @Override + public Flux decode(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux decoder) { + assertNull(connection); + decode.incrementAndGet(); + return decoder; + } + + @Override + public Flux beforeSendToPlatform(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Flux handler) { + assertNull(connection); + beforeSend.incrementAndGet(); + return handler.doOnNext(message -> { + monitored.add(message); + completed.countDown(); + }); + } + } +} diff --git a/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSessionTest.java b/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSessionTest.java index a9d4ebb61..9fe9ea933 100644 --- a/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSessionTest.java +++ b/jetlinks-components/network-component/mqtt-component/src/test/java/org/jetlinks/community/network/mqtt/gateway/device/session/MqttConnectionSessionTest.java @@ -17,6 +17,7 @@ import org.jetlinks.core.message.codec.EncodedMessage; import org.jetlinks.core.message.codec.MqttMessage; import org.jetlinks.core.message.codec.SimpleMqttMessage; +import org.jetlinks.core.server.ClientConnection; import org.jetlinks.core.server.session.DeviceSession; import org.junit.jupiter.api.Test; import reactor.core.Disposable; @@ -30,6 +31,7 @@ import java.util.List; import java.util.Optional; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Predicate; @@ -38,6 +40,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; /** @@ -51,6 +54,11 @@ public class MqttConnectionSessionTest { private final DeviceGatewayMonitor monitor = GatewayMonitors.getDeviceGatewayMonitor("unit-test"); private MqttConnectionSession newSession(MqttConnection connection) { + return newSession(connection, monitor); + } + + private MqttConnectionSession newSession(MqttConnection connection, + DeviceGatewayMonitor monitor) { return new MqttConnectionSession( "device-1", null, DefaultTransport.MQTT, connection, monitor, new FakeSessionManager()); } @@ -147,6 +155,33 @@ public void registerAliveConnectionStillWorks() { assertEquals(1, connection.published.size()); } + @Test + public void downstreamMonitorShouldWrapSender() { + FakeMqttConnection connection = new FakeMqttConnection(true); + RuntimeException error = new RuntimeException("rejected by monitor"); + AtomicInteger calls = new AtomicInteger(); + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Mono downstream(ClientConnection actualConnection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + assertSame(connection, actualConnection); + calls.incrementAndGet(); + return Mono.error(error); + } + }; + MqttConnectionSession session = newSession(connection, monitor); + + StepVerifier + .create(session.send(mqttMessage())) + .expectErrorSatisfies(actual -> assertSame(error, actual)) + .verify(); + + assertEquals(1, calls.get()); + assertTrue(connection.published.isEmpty()); + } + /** * MQTT 连接测试替身:isAlive 可控,accept() 模拟 CONNACK 后 isConnected=true 的语义. */ diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSession.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSession.java index 6e9a88b55..cf70924ea 100755 --- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSession.java +++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSession.java @@ -24,7 +24,6 @@ import org.jetlinks.core.message.codec.EncodedMessage; import org.jetlinks.core.message.codec.Transport; import org.jetlinks.core.server.session.DeviceSession; -import org.jetlinks.community.network.tcp.TcpMessage; import org.jetlinks.community.network.tcp.client.TcpClient; import org.jetlinks.core.server.session.MultiConnectionDeviceSession; import reactor.core.publisher.Mono; @@ -70,7 +69,9 @@ public Mono send(EncodedMessage encodedMessage) { if (client == null) { return Mono.error(new DeviceOperationException.NoStackTrace(ErrorCode.CONNECTION_LOST)); } - return client.send(new TcpMessage(encodedMessage.getPayload())); + return monitor + .downstream(client, this, encodedMessage, client.sendMessage(encodedMessage)) + .thenReturn(true); }); } diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpServerDeviceGateway.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpServerDeviceGateway.java index c251de601..3058915a6 100755 --- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpServerDeviceGateway.java +++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/TcpServerDeviceGateway.java @@ -60,6 +60,7 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import java.util.concurrent.atomic.LongAdder; +import java.util.function.UnaryOperator; @Slf4j class TcpServerDeviceGateway extends AbstractDeviceGateway implements DeviceGateway, MonitorSupportDeviceGateway { @@ -131,8 +132,8 @@ static class TcpConnection extends Mono implements DeviceGatewayContext, R this.client = client; this.parent = parent; this.address = client.getRemoteAddress(); + parent.counter.increment(); parent.monitor.totalConnection(parent.counter.sum()); - parent.monitor.connected(); client.onDisconnect(this); legalityChecker = Schedulers @@ -146,6 +147,7 @@ public void checkLegality() { if (session == null) { log.info("tcp [{}] connection is illegal, close it.", address); try { + parent.monitor.rejected(client, null); client.disconnect(); } catch (Throwable ignore) { } @@ -197,25 +199,51 @@ Mono handleTcpMessage0(EncodedMessage message) { if (!parent.isStarted()) { return Mono.empty(); } - return parent + + if (!parent.monitor.beforeDecode(client, message)) { + return Mono.empty(); + } + + DeviceSession deviceSession = session(); + UnaryOperator> platformHandler = task -> task + .concatMap(this::handleDeviceMessage, 0); + Flux decodeTask = parent .getProtocol() .flatMap(pt -> pt.getMessageCodec(parent.getTransport())) .flatMapMany(codec -> codec .decode(FromDeviceMessageContext.of( - session(), + deviceSession, message, parent.registry, client, - msg -> handleDeviceMessage(msg).then()))) - .cast(DeviceMessage.class) - .concatMap(this::handleDeviceMessage, 0) + // 手动输出不进入 codec 返回值，单独复用同一平台处理与发送前监控链。 + deviceMessage -> parent.monitor + .handleUpstream( + client, + deviceSession, + message, + Flux.just(deviceMessage), + platformHandler) + .then()))) + .cast(DeviceMessage.class); + + decodeTask = parent.monitor.handleUpstream( + client, + deviceSession, + message, + decodeTask, + platformHandler + ); + decodeTask = parent.monitor.decode(client, deviceSession, message, decodeTask); + + return decodeTask .as(FluxTracer.create( DeviceTracer.SpanName.decode0(session == null ? "unknown" : session.getDeviceId()), builder -> builder .setAttributeLazy( DeviceTracer.SpanKey.message, message, - (m) -> message.toString()) + Object::toString) )) .onErrorResume((err) -> { log.error("{} Handle TCP[{}] message failed:\n{}", @@ -312,7 +340,7 @@ public void cancel() { disposable.dispose(); } parent.counter.decrement(); - parent.monitor.disconnected(); + parent.monitor.disconnected(client); parent.monitor.totalConnection(parent.counter.sum()); if (this.subscriber != null) { this.subscriber.onComplete(); @@ -343,7 +371,11 @@ private void doStart() { .publishOn(Schedulers.parallel()) .flatMap(client -> { try { - return new TcpConnection(this, client); + if (monitor.connected(client)) { + return new TcpConnection(this, client); + } + client.disconnect(); + return Mono.empty(); } catch (Throwable e) { try { client.disconnect(); diff --git a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/UnknownTcpDeviceSession.java b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/UnknownTcpDeviceSession.java index 4294b5f0e..47e57b73e 100755 --- a/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/UnknownTcpDeviceSession.java +++ b/jetlinks-components/network-component/tcp-component/src/main/java/org/jetlinks/community/network/tcp/gateway/device/UnknownTcpDeviceSession.java @@ -21,8 +21,6 @@ import org.jetlinks.core.message.codec.EncodedMessage; import org.jetlinks.core.message.codec.Transport; import org.jetlinks.core.server.session.DeviceSession; -import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; -import org.jetlinks.community.network.tcp.TcpMessage; import org.jetlinks.community.network.tcp.client.TcpClient; import reactor.core.publisher.Mono; @@ -76,8 +74,12 @@ public long connectTime() { @Override public Mono send(EncodedMessage encodedMessage) { - return client.send(new TcpMessage(encodedMessage.getPayload())) - .doOnSuccess(ignore -> monitor.sentMessage()); + Mono sender = client + .sendMessage(encodedMessage) + .doOnSuccess(ignore -> monitor.sentMessage()); + return monitor + .downstream(client, this, encodedMessage, sender) + .thenReturn(true); } @Override diff --git a/jetlinks-components/network-component/tcp-component/src/test/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSessionTest.java b/jetlinks-components/network-component/tcp-component/src/test/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSessionTest.java new file mode 100644 index 000000000..6109d5fe5 --- /dev/null +++ b/jetlinks-components/network-component/tcp-component/src/test/java/org/jetlinks/community/network/tcp/gateway/device/TcpDeviceSessionTest.java @@ -0,0 +1,119 @@ +/* + * Copyright 2026 JetLinks https://www.jetlinks.cn + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.jetlinks.community.network.tcp.gateway.device; + +import io.netty.buffer.Unpooled; +import org.jetlinks.community.gateway.monitor.DeviceGatewayMonitor; +import org.jetlinks.community.network.tcp.TcpMessage; +import org.jetlinks.community.network.tcp.client.TcpClient; +import org.jetlinks.core.device.DeviceOperator; +import org.jetlinks.core.device.session.DeviceSessionManager; +import org.jetlinks.core.message.codec.DefaultTransport; +import org.jetlinks.core.message.codec.EncodedMessage; +import org.jetlinks.core.server.ClientConnection; +import org.jetlinks.core.server.session.DeviceSession; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +class TcpDeviceSessionTest { + + @Test + void downstreamMonitorShouldWrapSender() { + DeviceOperator operator = mock(DeviceOperator.class); + DeviceSessionManager sessionManager = mock(DeviceSessionManager.class); + TcpClient client = mock(TcpClient.class); + TcpMessage message = new TcpMessage(Unpooled.EMPTY_BUFFER); + RuntimeException error = new RuntimeException("rejected by monitor"); + AtomicInteger senderSubscriptions = new AtomicInteger(); + AtomicInteger monitorCalls = new AtomicInteger(); + + when(operator.getId()).thenReturn("device-1"); + when(client.isAlive()).thenReturn(true); + when(client.sendMessage(message)).thenReturn(Mono.defer(() -> { + senderSubscriptions.incrementAndGet(); + return Mono.empty(); + })); + + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + assertSame(client, connection); + assertSame(message, origin); + monitorCalls.incrementAndGet(); + return Mono.error(error); + } + }; + TcpDeviceSession session = new TcpDeviceSession( + operator, DefaultTransport.TCP, monitor, sessionManager); + session.registerConnection(client); + + StepVerifier + .create(session.send(message)) + .expectErrorSatisfies(actual -> assertSame(error, actual)) + .verify(); + + assertEquals(1, monitorCalls.get()); + assertEquals(0, senderSubscriptions.get()); + } + + @Test + void unknownSessionShouldAlsoUseDownstreamMonitor() { + TcpClient client = mock(TcpClient.class); + TcpMessage message = new TcpMessage(Unpooled.EMPTY_BUFFER); + RuntimeException error = new RuntimeException("rejected by monitor"); + AtomicInteger senderSubscriptions = new AtomicInteger(); + AtomicInteger monitorCalls = new AtomicInteger(); + + when(client.sendMessage(message)).thenReturn(Mono.defer(() -> { + senderSubscriptions.incrementAndGet(); + return Mono.empty(); + })); + + DeviceGatewayMonitor monitor = new DeviceGatewayMonitor() { + @Override + public Mono downstream(ClientConnection connection, + DeviceSession session, + EncodedMessage origin, + Mono sender) { + assertSame(client, connection); + assertSame(message, origin); + monitorCalls.incrementAndGet(); + return Mono.error(error); + } + }; + UnknownTcpDeviceSession session = new UnknownTcpDeviceSession( + "unknown", client, DefaultTransport.TCP, monitor); + + StepVerifier + .create(session.send(message)) + .expectErrorSatisfies(actual -> assertSame(error, actual)) + .verify(); + + assertEquals(1, monitorCalls.get()); + assertEquals(0, senderSubscriptions.get()); + } +}