class DeviceCurrentStateSubscriptionProvider implements SubscriptionProvider {

    private final LocalDeviceInstanceService instanceService;

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

    @Override
    public String name() {
        return "设备当前状态消息";
    }

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

    @Override
    @SuppressWarnings("all")
    public Flux<Map<String, Object>> subscribe(SubscribeRequest request) {<FILL_FUNCTION_BODY>}
}

        List<String> deviceId = request.get("deviceId")
            .map(List.class::cast)
            .orElseThrow(() -> new IllegalArgumentException("deviceId不能为空"));

        return Flux
            .fromIterable(deviceId)
            .buffer(200)
            .concatMap(buf -> {
                return instanceService.createQuery()
                    .select(DeviceInstanceEntity::getId, DeviceInstanceEntity::getState)
                    .in(DeviceInstanceEntity::getId, buf)
                    .fetch();
            })
            .map(instance -> Collections.singletonMap(instance.getId(), instance.getState().name()));

    
<no_super_class>