class DeviceMessageSendTaskExecutor extends FunctionTaskExecutor {

        private DeviceMessageSendConfig config;

        private Function<Map<String, Object>, Flux<DeviceOperator>> selector;

        public DeviceMessageSendTaskExecutor(ExecutionContext context) {
            super("发送设备消息", context);
            reload();
        }

        protected Flux<DeviceOperator> selectDevice(Map<String, Object> ctx) {
            return selector.apply(ctx);
        }

        @Override
        protected Publisher<RuleData> apply(RuleData input) {<FILL_FUNCTION_BODY>}

        @Override
        public void validate() {
            if (CollectionUtils.isEmpty(context.getJob().getConfiguration())) {
                throw new IllegalArgumentException("配置不能为空");
            }
            FastBeanCopier.copy(context.getJob().getConfiguration(), new DeviceMessageSendConfig()).validate();
        }

        @Override
        public void reload() {
            config = FastBeanCopier.copy(context.getJob().getConfiguration(), new DeviceMessageSendConfig());
            config.validate();
            if (config.getSelectorSpec() != null) {
                selector = selectorBuilder.createSelector(config.getSelectorSpec())::select;
            }  else if (StringUtils.hasText(config.deviceId)) {
                selector = ctx -> registry.getDevice(config.getDeviceId()).flux();
            } else if (StringUtils.hasText(config.productId)) {
                selector = selectorBuilder.createSelector(DeviceSelectorProviders.product(config.productId))::select;
            } else {
                if (config.isFixed() && MapUtils.isNotEmpty(config.getMessage())) {
                    selector = ctx -> registry.getDevice(config.getDeviceIdInMessage(ctx)).flux();
                } else {
                    selector = ctx -> registry
                        .getDevice((String) ctx
                            .getOrDefault("deviceId",
                                          config.getMessage() == null
                                              ? null
                                              : config.getMessage().get("deviceId")))
                        .flux();
                }
            }
        }

    }

            Map<String, Object> ctx = RuleDataHelper.toContextMap(input);

            Flux<DeviceOperator> readySendDevice =
                "ignoreOffline".equals(config.getStateOperator())
                    ? selectDevice(ctx).filterWhen(DeviceOperator::isOnline)
                    : selectDevice(ctx);

            return readySendDevice
                .switchIfEmpty(context.onError(() -> new DeviceOperationException(ErrorCode.SYSTEM_ERROR, "无可用设备"), input))
                .flatMap(device -> config
                    .doSend(ctx, context, device, input)
                    .onErrorResume(error -> context.onError(error, input))
                    .subscribeOn(Schedulers.parallel())
                )
                .map(reply -> {
                    RuleData data = context.newRuleData(input.newData(reply.toJson()));
                    if (config.getResponseHeaders() != null) {
                        config.getResponseHeaders().forEach(data::setHeader);
                    }
                    return data;
                })
                ;
        
<no_super_class>