class DeviceMessageSubscriptionProvider implements SubscriptionProvider {

    private final EventBus eventBus;

    @Override
    public String id() {
        return "device-message-subscriber";
    }

    @Override
    public String name() {
        return "订阅设备消息";
    }

    @Override
    public String[] getTopicPattern() {
        return new String[]{
            "/device/*/*/**"
        };
    }

    @Override
    public Flux<Message> subscribe(SubscribeRequest request) {<FILL_FUNCTION_BODY>}
}

        return eventBus
            .subscribe(
                org.jetlinks.core.event.Subscription.of(
                    "DeviceMessageSubscriptionProvider:" + request.getAuthentication().getUser().getId(),
                    new String[]{request.getTopic()},
                    org.jetlinks.core.event.Subscription.Feature.local,
                    Subscription.Feature.broker
                ))
            .map(topicMessage -> Message.success(request.getId(), topicMessage.getTopic(), topicMessage.decode()));
    
<no_super_class>