Java 类org.apache.camel.processor.DelegateAsyncProcessor 实例源码

@Override
    public Processor wrapProcessorInInterceptors(CamelContext context,
            final ProcessorDefinition<?> definition, final Processor target,
            final Processor nextTarget) throws Exception {
        return new DelegateAsyncProcessor(new Processor() {
            @Override
            public void process(Exchange exchange) throws Exception {
//              if(!camelConfig.isRunning()){
//                  System.err.println("系统将关闭，不在处理任务");
//                  return ;
//              }
                System.out.println("defainition :"+definition);
                System.out.println("nextTarget :"+nextTarget);
                target.process(exchange);
            }
        });
    }

public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
                                             final Processor target, final Processor nextTarget) throws Exception {
    return new DelegateAsyncProcessor(target) {
        @Override
        public boolean process(final Exchange exchange, final AsyncCallback callback) {
            debugger.beforeProcess(exchange, target, definition);
            final StopWatch watch = new StopWatch();

            return processor.process(exchange, new AsyncCallback() {
                public void done(boolean doneSync) {
                    long diff = watch.stop();
                    debugger.afterProcess(exchange, processor, definition, diff);

                    // must notify original callback
                    callback.done(doneSync);
                }
            });
        }

        @Override
        public String toString() {
            return "Debug[" + target + "]";
        }
    };
}

public Processor createErrorHandler(RouteContext routeContext, Processor processor) {
    return new DelegateAsyncProcessor(processor) {
        @Override
        public boolean process(final Exchange exchange, final AsyncCallback callback) {
            return super.process(exchange, new AsyncCallback() {
                @Override
                public void done(boolean doneSync) {
                    exchange.removeProperty(Exchange.REDELIVERY_EXHAUSTED);
                    callback.done(doneSync);
                }
            });
        }

        @Override
        public String toString() {
            if (processor == null) {
                // if no output then dont do any description
                return "";
            }
            return "NoErrorHandler[" + processor + "]";
        }
    };
}

public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
                                             final Processor target, final Processor nextTarget) throws Exception {

    return new DelegateAsyncProcessor(new Processor() {

        public void process(Exchange exchange) throws Exception {
            // we just count number of interceptions
            count++;
            LOG.info("I am the container wide interceptor. Intercepted total count: " + count);
            target.process(exchange);
        }

        @Override
        public String toString() {
            return "ContainerWideInterceptor[" + target + "]";
        }
    });
}

public Processor wrapProcessorInInterceptors(final CamelContext context, final ProcessorDefinition<?> definition,
                                             final Processor target, final Processor nextTarget) throws Exception {

    // use DelegateAsyncProcessor to ensure the interceptor works well with the asynchronous routing
    // engine in Camel.
    // The target is the processor to continue routing to, which we must provide
    // in the constructor of the DelegateAsyncProcessor
    return new DelegateAsyncProcessor(target) {
        @Override
        public boolean process(Exchange exchange, AsyncCallback callback) {
            // we just want to count number of interceptions
            counter.incrementAndGet();

            // invoke processor to continue routing the message
            return processor.process(exchange, callback);
        }
    };
}

public Processor wrapProcessorInInterceptors(CamelContext context, ProcessorDefinition<?> definition, Processor target, Processor nextTarget) throws Exception {
    return new DelegateAsyncProcessor(target) {
        public boolean process(Exchange exchange, AsyncCallback callback) {
            invoked = true;
            return processor.process(exchange, callback);
        }
    };
}

@Override
public Processor wrapProcessorInInterceptors(CamelContext context, ProcessorDefinition<?> definition, Processor target, Processor nextTarget) throws Exception {
    if ("bean".equals(definition.getShortName())) {
        return new DelegateAsyncProcessor(exchange -> processorProxy.doAsSystem(target, exchange));
    } else {
        return target;
    }
}

public Processor wrapProcessorInInterceptors(CamelContext context,
        ProcessorDefinition<?> definition, final Processor target,
        Processor nextTarget) throws Exception {

    return new DelegateAsyncProcessor(new Processor() {

        public void process(Exchange exchange) throws Exception {
            LOG.info("Before the processor...");
            target.process(exchange);
            LOG.info("After the processor...");
        }
    });
}

public Processor wrapProcessorInInterceptors(CamelContext context,
        ProcessorDefinition<?> definition, final Processor target,
        Processor nextTarget) throws Exception {

    return new DelegateAsyncProcessor(new Processor() {

        public void process(Exchange exchange) throws Exception {
            LOG.info("Before the processor...");
            target.process(exchange);
            LOG.info("After the processor...");
        }
    });
}

public void initStatisticsInterceptor() throws NullPointerException {
      theContext.addInterceptStrategy(new InterceptStrategy() {

    public Processor wrapProcessorInInterceptors(CamelContext context,
            final ProcessorDefinition<?> node, final Processor target,
            Processor nextTarget) throws Exception {
        return new DelegateAsyncProcessor(target) {
            public boolean process(Exchange exchange, AsyncCallback callback) {
                CamelStat.this.addRowByTargetNode(node.getId());
                return super.process(exchange, callback);
            }
        };
    }
});
  }

public Processor wrapProcessorInInterceptors(CamelContext context,
        ProcessorDefinition<?> definition, final Processor target,
        Processor nextTarget) throws Exception {
    // to make the Default channel happy
    return new DelegateAsyncProcessor(new Processor() {

        public void process(Exchange exchange) throws Exception {
            log.info("Before the processor...");
            target.process(exchange);
            log.info("After the processor...");
        }
    });
}