class DelayTaskExecutor extends AbstractTaskExecutor {

        private DelayTaskExecutorConfig config;

        private final Scheduler scheduler;

        public DelayTaskExecutor(ExecutionContext context, Scheduler scheduler) {
            super(context);
            this.scheduler = scheduler;
            init();
        }

        @Override
        protected Disposable doStart() {<FILL_FUNCTION_BODY>}

        void init() {
            config = DelayTaskExecutorConfig.of(context.getJob().getConfiguration());
        }

        @Override
        public void reload() {
            init();
            disposable = doStart();
        }

        @Override
        public String getName() {
            return "延迟";
        }

    }

            if (this.disposable != null) {
                this.disposable.dispose();
            }
            return config
                .create(context.getInput().accept(), context, scheduler)
                .map(context::newRuleData)
                .flatMap(ruleData ->
                             context
                                 .fireEvent(RuleConstants.Event.result, ruleData)
                                 .then(context.getOutput().write(Mono.just(ruleData)))
                )
                .onErrorResume(err -> context.onError(err, null))
                .subscribe();
        
<no_super_class>