Java 类org.springframework.core.task.AsyncListenableTaskExecutor 实例源码

@Test
public void postProcessAfterInitialization() throws Exception {
    assertThat(processor.postProcessAfterInitialization(mock(Executor.class), toBeExcluded).getClass(),
            not(equalTo(ContextAwareExecutor.class)));
    //concurrent
    assertThat(processor.postProcessAfterInitialization(mock(Executor.class), beanName).getClass(),
            equalTo(ContextAwareExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(ExecutorService.class), beanName).getClass(),
            equalTo(ContextAwareExecutorService.class));
    assertThat(processor.postProcessAfterInitialization(mock(ScheduledExecutorService.class), beanName).getClass(),
            equalTo(ContextAwareScheduledExecutorService.class));

    //spring
    assertThat(processor.postProcessAfterInitialization(mock(TaskScheduler.class), beanName).getClass(),
            equalTo(ContextAwareTaskScheduler.class));
    assertThat(processor.postProcessAfterInitialization(new ThreadPoolTaskExecutor(), beanName).getClass(),
            equalTo(ContextAwareThreadPoolTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(new ThreadPoolTaskScheduler(), beanName).getClass(),
            equalTo(ContextAwareThreadPoolTaskScheduler.class));
    assertThat(processor.postProcessAfterInitialization(mock(AsyncListenableTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareAsyncListenableTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(AsyncTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareAsyncTaskExecutor.class));
    assertThat(processor.postProcessAfterInitialization(mock(SchedulingTaskExecutor.class), beanName).getClass(),
            equalTo(ContextAwareSchedulingTaskExecutor.class));
}

/**
 * Determine the specific executor to use when executing the given method.
 * Should preferably return an {@link AsyncListenableTaskExecutor} implementation.
 * @return the executor to use (or {@code null}, but just if no default executor has been set)
 */
protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
    AsyncTaskExecutor executor = this.executors.get(method);
    if (executor == null) {
        Executor executorToUse = this.defaultExecutor;
        String qualifier = getExecutorQualifier(method);
        if (StringUtils.hasLength(qualifier)) {
            if (this.beanFactory == null) {
                throw new IllegalStateException("BeanFactory must be set on " + getClass().getSimpleName() +
                        " to access qualified executor '" + qualifier + "'");
            }
            executorToUse = BeanFactoryAnnotationUtils.qualifiedBeanOfType(
                    this.beanFactory, Executor.class, qualifier);
        }
        else if (executorToUse == null) {
            return null;
        }
        executor = (executorToUse instanceof AsyncListenableTaskExecutor ?
                (AsyncListenableTaskExecutor) executorToUse : new TaskExecutorAdapter(executorToUse));
        this.executors.put(method, executor);
    }
    return executor;
}

/**
 * Delegate for actually executing the given task with the chosen executor.
 * @param task the task to execute
 * @param executor the chosen executor
 * @param returnType the declared return type (potentially a {@link Future} variant)
 * @return the execution result (potentially a corresponding {@link Future} handle)
 */
protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {
    if (completableFuturePresent) {
        Future<Object> result = CompletableFutureDelegate.processCompletableFuture(returnType, task, executor);
        if (result != null) {
            return result;
        }
    }
    if (ListenableFuture.class.isAssignableFrom(returnType)) {
        return ((AsyncListenableTaskExecutor) executor).submitListenable(task);
    }
    else if (Future.class.isAssignableFrom(returnType)) {
        return executor.submit(task);
    }
    else {
        executor.submit(task);
        return null;
    }
}

/**
 * Delegate for actually executing the given task with the chosen executor.
 * @param task the task to execute
 * @param executor the chosen executor
 * @param returnType the declared return type (potentially a {@link Future} variant)
 * @return the execution result (potentially a corresponding {@link Future} handle)
 */
protected Object doSubmit(Callable<Object> task, AsyncTaskExecutor executor, Class<?> returnType) {
    if (completableFuturePresent) {
        Future<Object> result = CompletableFutureDelegate.processCompletableFuture(returnType, task, executor);
        if (result != null) {
            return result;
        }
    }
    if (ListenableFuture.class.isAssignableFrom(returnType)) {
        return ((AsyncListenableTaskExecutor) executor).submitListenable(task);
    }
    else if (Future.class.isAssignableFrom(returnType)) {
        return executor.submit(task);
    }
    else {
        executor.submit(task);
        return null;
    }
}

/**
 * Will use the WriteListener if the TaskExecutor is an instance of AsyncListenableTaskExecutor. The WriteListener
 * will then be used to listen for failures.
 *
 * @param runnable
 * @param items
 */
protected void executeRunnable(Runnable runnable, final List<? extends DocumentWriteOperation> items) {
    if (writeListener != null && taskExecutor instanceof AsyncListenableTaskExecutor) {
        AsyncListenableTaskExecutor asyncListenableTaskExecutor = (AsyncListenableTaskExecutor)taskExecutor;
        ListenableFuture<?> future = asyncListenableTaskExecutor.submitListenable(runnable);
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable ex) {
                writeListener.onWriteFailure(ex, items);
            }
            @Override
            public void onSuccess(Object result) {
            }
        });
    } else {
        taskExecutor.execute(runnable);
    }
}

/**
 * Create a new instance of the {@code AsyncRestTemplate} using the given
 * {@link AsyncTaskExecutor}.
 * <p>This constructor uses a {@link SimpleClientHttpRequestFactory} in combination
 * with the given {@code AsyncTaskExecutor} for asynchronous execution.
 */
public AsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor) {
    Assert.notNull(taskExecutor, "AsyncTaskExecutor must not be null");
    SimpleClientHttpRequestFactory requestFactory =
            new SimpleClientHttpRequestFactory();
    requestFactory.setTaskExecutor(taskExecutor);
    this.syncTemplate = new RestTemplate(requestFactory);
    setAsyncRequestFactory(requestFactory);
}

SimpleStreamingAsyncClientHttpRequest(HttpURLConnection connection, int chunkSize,
        boolean outputStreaming, AsyncListenableTaskExecutor taskExecutor) {

    this.connection = connection;
    this.chunkSize = chunkSize;
    this.outputStreaming = outputStreaming;
    this.taskExecutor = taskExecutor;
}

SimpleBufferingAsyncClientHttpRequest(HttpURLConnection connection,
        boolean outputStreaming, AsyncListenableTaskExecutor taskExecutor) {

    this.connection = connection;
    this.outputStreaming = outputStreaming;
    this.taskExecutor = taskExecutor;
}

/**
 * Create a new instance of the {@code AsyncRestTemplate} using the given
 * {@link AsyncTaskExecutor}.
 * <p>This constructor uses a {@link SimpleClientHttpRequestFactory} in combination
 * with the given {@code AsyncTaskExecutor} for asynchronous execution.
 */
public AsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor) {
    Assert.notNull(taskExecutor, "AsyncTaskExecutor must not be null");
    SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    requestFactory.setTaskExecutor(taskExecutor);
    this.syncTemplate = new RestTemplate(requestFactory);
    setAsyncRequestFactory(requestFactory);
}

SimpleStreamingAsyncClientHttpRequest(HttpURLConnection connection, int chunkSize,
        boolean outputStreaming, AsyncListenableTaskExecutor taskExecutor) {

    this.connection = connection;
    this.chunkSize = chunkSize;
    this.outputStreaming = outputStreaming;
    this.taskExecutor = taskExecutor;
}

SimpleBufferingAsyncClientHttpRequest(HttpURLConnection connection,
        boolean outputStreaming, AsyncListenableTaskExecutor taskExecutor) {

    this.connection = connection;
    this.outputStreaming = outputStreaming;
    this.taskExecutor = taskExecutor;
}

@Override
protected AsyncClientHttpRequestFactory createRequestFactory() {
    SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
    AsyncListenableTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    requestFactory.setTaskExecutor(taskExecutor);
    return requestFactory;
}

private String registerAsyncListenableTaskExecutor(final String id) {
    return registry.register(id, AsyncListenableTaskExecutor.class, () -> {
        LOG.debug("Client [{}]: Registering AsyncListenableTaskExecutor", id);

        return genericBeanDefinition(ConcurrentTaskExecutor.class)
                .addConstructorArgValue(BeanDefinitionBuilder.genericBeanDefinition(TracingExecutors.class)
                        .setFactoryMethod("preserve")
                        .addConstructorArgValue(genericBeanDefinition(Executors.class)
                                .setFactoryMethod("newCachedThreadPool")
                                .setDestroyMethodName("shutdown")
                                .getBeanDefinition())
                        .addConstructorArgReference("tracer")
                        .getBeanDefinition());
    });
}

/**
 * Determine the specific executor to use when executing the given method.
 * Should preferably return an {@link AsyncListenableTaskExecutor} implementation.
 * @return the executor to use (or {@code null}, but just if no default executor is available)
 */
protected AsyncTaskExecutor determineAsyncExecutor(Method method) {
    AsyncTaskExecutor executor = this.executors.get(method);
    if (executor == null) {
        Executor targetExecutor;
        String qualifier = getExecutorQualifier(method);
        if (StringUtils.hasLength(qualifier)) {
            targetExecutor = findQualifiedExecutor(this.beanFactory, qualifier);
        }
        else {
            targetExecutor = this.defaultExecutor;
            if (targetExecutor == null) {
                synchronized (this.executors) {
                    if (this.defaultExecutor == null) {
                        this.defaultExecutor = getDefaultExecutor(this.beanFactory);
                    }
                    targetExecutor = this.defaultExecutor;
                }
            }
        }
        if (targetExecutor == null) {
            return null;
        }
        executor = (targetExecutor instanceof AsyncListenableTaskExecutor ?
                (AsyncListenableTaskExecutor) targetExecutor : new TaskExecutorAdapter(targetExecutor));
        this.executors.put(method, executor);
    }
    return executor;
}

public RestAsyncClientHttpRequestFactory(final HttpClient client, final AsyncListenableTaskExecutor executor) {
    final RequestConfig config = Configurable.class.cast(client).getConfig();

    this.factory = new HttpComponentsClientHttpRequestFactory(client) {
        @Override
        protected void postProcessHttpRequest(final HttpUriRequest request) {
            // restore the client's request settings that are incorrectly handled by spring
            HttpRequestBase.class.cast(request).setConfig(config);
        }
    };
    this.executor = executor;
}

/**
 * @return an Authenticated task executor ready to run.
 */
protected AsyncListenableTaskExecutor getAsyncExecutor() {
    final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(10);
    executor.setWaitForTasksToCompleteOnShutdown(false);
    executor.initialize();
    return executor;
}

/**
 * {@inheritDoc}
 */
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {
    if (bean instanceof Executor || bean instanceof TaskScheduler) {
        if (properties.getExecutor().accept(beanName)) {
            if (bean instanceof AsyncListenableTaskExecutor && bean instanceof SchedulingTaskExecutor && bean instanceof TaskScheduler) {
                log.info("Context propagation enabled for ~ThreadPoolTaskScheduler [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareThreadPoolTaskScheduler((AsyncListenableTaskExecutor) bean, (SchedulingTaskExecutor) bean, (TaskScheduler) bean);
            } else if (bean instanceof AsyncListenableTaskExecutor && bean instanceof SchedulingTaskExecutor) {
                log.info("Context propagation enabled for ~ThreadPoolTaskExecutor [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareThreadPoolTaskExecutor((AsyncListenableTaskExecutor) bean, (SchedulingTaskExecutor) bean);
            } else if (bean instanceof TaskScheduler) {
                log.info("Context propagation enabled for TaskScheduler [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareTaskScheduler((TaskScheduler) bean);
            } else if (bean instanceof SchedulingTaskExecutor) {
                log.info("Context propagation enabled for SchedulingTaskExecutor [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareSchedulingTaskExecutor((SchedulingTaskExecutor) bean);
            } else if (bean instanceof AsyncListenableTaskExecutor) {
                log.info("Context propagation enabled for AsyncListenableTaskExecutor [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareAsyncListenableTaskExecutor((AsyncListenableTaskExecutor) bean);
            } else if (bean instanceof AsyncTaskExecutor) {
                log.info("Context propagation enabled for AsyncTaskExecutor [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareAsyncTaskExecutor((AsyncTaskExecutor) bean);
            } else if (bean instanceof ScheduledExecutorService) {
                log.info("Context propagation enabled for ScheduledExecutorService [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareScheduledExecutorService((ScheduledExecutorService) bean);
            } else if (bean instanceof ExecutorService) {
                log.info("Context propagation enabled for ExecutorService [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareExecutorService((ExecutorService) bean);
            } else {
                log.info("Context propagation enabled for Executor [{}]:[{}].", beanName, bean.getClass().getName());
                return new ContextAwareExecutor((Executor) bean);
            }
        } else {
            log.debug("Context propagation disabled for Executor [{}]", beanName);
        }
    }
    return bean;
}

/**
 * Return the configured {@link TaskExecutor}.
 */
public AsyncListenableTaskExecutor getTaskExecutor() {
    return this.taskExecutor;
}

/**
 * Return the configured {@link TaskExecutor}.
 */
public AsyncListenableTaskExecutor getTaskExecutor() {
    return this.taskExecutor;
}

@Conditional(RestifyAsyncConfiguration.RestifyAsyncTaskExecutorCondition.class)
@Bean
public AsyncListenableTaskExecutor restifyAsyncTaskExecutor() {
    return new SimpleAsyncTaskExecutor("RestifyAsyncTaskExecutor");
}

@ConditionalOnMissingBean
@Bean
public ListenableFutureEndpointCallExecutableFactory<Object, Object> listenableFutureEndpointCallExecutableFactory(
        @Qualifier("restifyAsyncTaskExecutor") AsyncListenableTaskExecutor executor) {
    return new ListenableFutureEndpointCallExecutableFactory<>(executor);
}

@ConditionalOnMissingBean
@Bean
public ListenableFutureTaskEndpointCallExecutableFactory<Object, Object> listenableFutureTaskEndpointCallExecutableFactory(
        @Qualifier("restifyAsyncTaskExecutor") AsyncListenableTaskExecutor executor) {
    return new ListenableFutureTaskEndpointCallExecutableFactory<>(executor);
}

public ListenableFutureTaskEndpointCallExecutableFactory(AsyncListenableTaskExecutor asyncTaskExecutor) {
    this.asyncListenableTaskExecutor = asyncTaskExecutor;
}

public ListenableFutureEndpointCallExecutableFactory(AsyncListenableTaskExecutor asyncTaskExecutor) {
    this.asyncListenableTaskExecutor = asyncTaskExecutor;
}

@Bean
@Qualifier("example")
public AsyncListenableTaskExecutor exampleAsyncListenableTaskExecutor() {
    return mock(AsyncListenableTaskExecutor.class);
}

TraceAsyncListenableTaskExecutor(AsyncListenableTaskExecutor delegate,
        Tracer tracer) {
    this.delegate = delegate;
    this.tracer = tracer;
}

private AsyncListenableTaskExecutor asyncListenableTaskExecutor(Tracer tracer) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.initialize();
    return new TraceAsyncListenableTaskExecutor(threadPoolTaskScheduler, tracer);
}

private AsyncListenableTaskExecutor asyncListenableTaskExecutor(Tracer tracer) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.initialize();
    return new TraceAsyncListenableTaskExecutor(threadPoolTaskScheduler, tracer);
}

public TraceAsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor, Tracer tracer) {
    super(taskExecutor);
    this.tracer = tracer;
}

public AsyncListenableTaskExecutor getExecutor() {
    return executor;
}

public void setExecutor(AsyncListenableTaskExecutor executor) {
    this.executor = executor;
}

public StackdriverSpanConsumer(
    TraceTranslator translator, TraceConsumer consumer, AsyncListenableTaskExecutor executor) {
  this.translator = translator;
  this.consumer = consumer;
  this.executor = executor;
}

public void setTaskExecutor( AsyncListenableTaskExecutor jobExecutor )
{
    this.jobExecutor = jobExecutor;
}

RestAsyncClientHttpRequest(final ClientHttpRequest request, final AsyncListenableTaskExecutor executor) {
    this.request = request;
    this.executor = executor;
}

public AsyncRest(AsyncListenableTaskExecutor taskExecutor) {
    super(taskExecutor);
}

TraceAsyncListenableTaskExecutor(AsyncListenableTaskExecutor delegate,
        Tracer tracer) {
    this.delegate = delegate;
    this.tracer = tracer;
}

private AsyncListenableTaskExecutor asyncListenableTaskExecutor(Tracer tracer) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.initialize();
    return new TraceAsyncListenableTaskExecutor(threadPoolTaskScheduler, tracer);
}

private AsyncListenableTaskExecutor asyncListenableTaskExecutor(Tracer tracer) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.initialize();
    return new TraceAsyncListenableTaskExecutor(threadPoolTaskScheduler, tracer);
}

public TraceAsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor,
        Tracer tracer, ErrorParser errorParser) {
    super(taskExecutor);
    this.tracer = tracer;
    this.errorParser = errorParser;
}

public void setTaskExecutor( AsyncListenableTaskExecutor taskExecutor )
{
    this.taskExecutor = taskExecutor;
}