Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-987 Channels, dispatchers, and pollers now accept any Executor rather than being restricted to only TaskExecutor implementations.

Browse Source
This commit is contained in:
Mark Fisher
2010-02-20 19:23:21 +00:00
parent f26df90077
commit a0871d40f2
10 changed files with 98 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
org.springframework.integration/src/main/java/org/springframework/integration/channel/ExecutorChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,9 +16,10 @@
package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
@@ -29,11 +30,11 @@ import org.springframework.util.ErrorHandler;
/**
* An implementation of {@link MessageChannel} that delegates to an instance of
* {@link UnicastingDispatcher} which in turn delegates all dispatching
* invocations to a {@link TaskExecutor}.
* invocations to an {@link Executor}.
* <p>
* <emphasis>NOTE: unlike DirectChannel, the ExecutorChannel does not support a
* shared transactional context between sender and handler, because the
* {@link TaskExecutor} typically does not block the sender's Thread since it
* {@link Executor} typically does not block the sender's Thread since it
* uses another Thread for the dispatch.</emphasis> (SyncTaskExecutor is an
* exception but would provide no value for this channel. If synchronous
* dispatching is required, a DirectChannel should be used instead).
@@ -45,7 +46,7 @@ public class ExecutorChannel extends AbstractSubscribableChannel implements Bean
private volatile UnicastingDispatcher dispatcher;
private volatile TaskExecutor taskExecutor;
private volatile Executor executor;
private volatile boolean failover = true;
@@ -54,24 +55,24 @@ public class ExecutorChannel extends AbstractSubscribableChannel implements Bean
/**
* Create an ExecutorChannel that delegates to the provided
* {@link TaskExecutor} when dispatching Messages.
* {@link Executor} when dispatching Messages.
* <p>
* The TaskExecutor must not be null.
* The Executor must not be null.
*/
public ExecutorChannel(TaskExecutor taskExecutor) {
this(taskExecutor, null);
public ExecutorChannel(Executor executor) {
this(executor, null);
}
/**
* Create an ExecutorChannel with a {@link LoadBalancingStrategy} that
* delegates to the provided {@link TaskExecutor} when dispatching Messages.
* delegates to the provided {@link Executor} when dispatching Messages.
* <p>
* The TaskExecutor must not be null.
* The Executor must not be null.
*/
public ExecutorChannel(TaskExecutor taskExecutor, LoadBalancingStrategy loadBalancingStrategy) {
Assert.notNull(taskExecutor, "taskExecutor must not be null");
this.taskExecutor = taskExecutor;
this.dispatcher = new UnicastingDispatcher(taskExecutor);
public ExecutorChannel(Executor executor, LoadBalancingStrategy loadBalancingStrategy) {
Assert.notNull(executor, "executor must not be null");
this.executor = executor;
this.dispatcher = new UnicastingDispatcher(executor);
if (loadBalancingStrategy != null) {
this.loadBalancingStrategy = loadBalancingStrategy;
this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
@@ -94,11 +95,11 @@ public class ExecutorChannel extends AbstractSubscribableChannel implements Bean
}
public void setBeanFactory(BeanFactory beanFactory) {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
this.executor = new ErrorHandlingTaskExecutor(this.executor, errorHandler);
}
this.dispatcher = new UnicastingDispatcher(this.taskExecutor);
this.dispatcher = new UnicastingDispatcher(this.executor);
this.dispatcher.setFailover(this.failover);
if (this.loadBalancingStrategy != null) {
this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);

29
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,9 +16,10 @@
package org.springframework.integration.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.ErrorHandler;
@@ -32,7 +33,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel impleme
private volatile BroadcastingDispatcher dispatcher;
private volatile TaskExecutor taskExecutor;
private volatile Executor executor;
private volatile ErrorHandler errorHandler;
@@ -42,13 +43,13 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel impleme
/**
* Create a PublishSubscribeChannel that will use a {@link TaskExecutor}
* Create a PublishSubscribeChannel that will use an {@link Executor}
* to invoke the handlers. If this is null, each invocation will occur in
* the message sender's thread.
*/
public PublishSubscribeChannel(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
this.dispatcher = new BroadcastingDispatcher(taskExecutor);
public PublishSubscribeChannel(Executor executor) {
this.executor = executor;
this.dispatcher = new BroadcastingDispatcher(executor);
}
/**
@@ -63,14 +64,14 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel impleme
/**
* Provide an {@link ErrorHandler} strategy for handling Exceptions that
* occur downstream from this channel. This will <i>only</i> be applied if
* a TaskExecutor has been configured to dispatch the Messages for this
* an Executor has been configured to dispatch the Messages for this
* channel. Otherwise, Exceptions will be thrown directly within the
* sending Thread. If no ErrorHandler is provided, and this channel does
* delegate its dispatching to a TaskExecutor, the default strategy is
* delegate its dispatching to an Executor, the default strategy is
* a {@link MessagePublishingErrorHandler} that sends error messages to
* the failed request Message's error channel header if available or to
* the default 'errorChannel' otherwise.
* @see #PublishSubscribeChannel(TaskExecutor)
* @see #PublishSubscribeChannel(Executor)
*/
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
@@ -104,14 +105,14 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel impleme
* Callback method for the {@link BeanFactoryAware} interface.
*/
public void setBeanFactory(BeanFactory beanFactory) {
if (this.taskExecutor != null) {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.executor != null) {
if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(beanFactory));
}
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);
this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler);
}
this.dispatcher = new BroadcastingDispatcher(this.taskExecutor);
this.dispatcher = new BroadcastingDispatcher(this.executor);
this.dispatcher.setIgnoreFailures(this.ignoreFailures);
this.dispatcher.setApplySequence(this.applySequence);
}

26
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,8 +18,8 @@ package org.springframework.integration.dispatcher;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.Executor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.message.MessageBuilder;
@@ -29,7 +29,7 @@ import org.springframework.integration.message.MessageHandler;
* A broadcasting dispatcher implementation. If the 'ignoreFailures' property
* is set to <code>false</code> (the default), it will fail fast such that any
* Exception thrown by a MessageHandler may prevent subsequent handlers from
* receiving the Message. However, when a TaskExecutor is provided, the Messages
* receiving the Message. However, when an Executor is provided, the Messages
* may be dispatched in separate Threads so that other handlers are invoked even
* when the 'ignoreFailures' flag is <code>false</code>.
* <p>
@@ -48,15 +48,15 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
private volatile boolean applySequence;
private final TaskExecutor taskExecutor;
private final Executor executor;
public BroadcastingDispatcher() {
this.taskExecutor = null;
this.executor = null;
}
public BroadcastingDispatcher(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
public BroadcastingDispatcher(Executor executor) {
this.executor = executor;
}
@@ -66,11 +66,11 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
* Exception will be thrown when a handler fails. To override this and
* suppress Exceptions, set the value to <code>true</code>.
* <p>
* Keep in mind that when using a TaskExecutor, even without ignoring the
* Keep in mind that when using an Executor, even without ignoring the
* failures, other handlers may be invoked after one throws an Exception.
* Since the TaskExecutor is using a different thread, this flag will only
* affect whether an error Message is sent to the error channel or not in
* the case that a TaskExecutor has been configured.
* Since the Executor is most likely using a different thread, this flag would
* only affect whether an error Message is sent to the error channel or not in
* the case that such an Executor has been configured.
*/
public void setIgnoreFailures(boolean ignoreFailures) {
this.ignoreFailures = ignoreFailures;
@@ -98,8 +98,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
.setCorrelationId(message.getHeaders().getId())
.setHeader(MessageHeaders.ID, UUID.randomUUID())
.build();
if (this.taskExecutor != null) {
this.taskExecutor.execute(new Runnable() {
if (this.executor != null) {
this.executor.execute(new Runnable() {
public void run() {
invokeHandler(handler, messageToSend);
}

16
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java
View File

@@ -1,4 +1,4 @@
/* Copyright 2002-2009 the original author or authors.
/* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,8 +18,8 @@ package org.springframework.integration.dispatcher;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandler;
@@ -49,15 +49,15 @@ public class UnicastingDispatcher extends AbstractDispatcher {
private volatile LoadBalancingStrategy loadBalancingStrategy;
private final TaskExecutor taskExecutor;
private final Executor executor;
public UnicastingDispatcher() {
this.taskExecutor = null;
this.executor = null;
}
public UnicastingDispatcher(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
public UnicastingDispatcher(Executor executor) {
this.executor = executor;
}
@@ -79,8 +79,8 @@ public class UnicastingDispatcher extends AbstractDispatcher {
}
public final boolean dispatch(final Message<?> message) {
if (this.taskExecutor != null) {
this.taskExecutor.execute(new Runnable() {
if (this.executor != null) {
this.executor.execute(new Runnable() {
public void run() {
doDispatch(message);
}

8
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,13 +18,13 @@ package org.springframework.integration.endpoint;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import org.aopalliance.aop.Advice;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -51,7 +51,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
protected volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private volatile TaskExecutor taskExecutor;
private volatile Executor taskExecutor;
private volatile ErrorHandler errorHandler;
@@ -96,7 +96,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}

10
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,9 +17,9 @@
package org.springframework.integration.scheduling;
import java.util.List;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
@@ -37,7 +37,7 @@ public class PollerMetadata {
private List<Advice> adviceChain;
private volatile TaskExecutor taskExecutor;
private volatile Executor taskExecutor;
private volatile PlatformTransactionManager transactionManager;
@@ -76,11 +76,11 @@ public class PollerMetadata {
return this.adviceChain;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public TaskExecutor getTaskExecutor() {
public Executor getTaskExecutor() {
return this.taskExecutor;
}

16
org.springframework.integration/src/main/java/org/springframework/integration/util/ErrorHandlingTaskExecutor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,12 +16,14 @@
package org.springframework.integration.util;
import java.util.concurrent.Executor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
/**
* A {@link TaskExecutor} implementation that wraps an existing TaskExecutor
* A {@link TaskExecutor} implementation that wraps an existing Executor
* instance in order to catch any exceptions. If an exception is thrown, it
* will be handled by the provided {@link ErrorHandler}.
*
@@ -30,21 +32,21 @@ import org.springframework.util.ErrorHandler;
*/
public class ErrorHandlingTaskExecutor implements TaskExecutor {
private final TaskExecutor taskExecutor;
private final Executor executor;
private final ErrorHandler errorHandler;
public ErrorHandlingTaskExecutor(TaskExecutor taskExecutor, ErrorHandler errorHandler) {
Assert.notNull(taskExecutor, "taskExecutor must not be null");
public ErrorHandlingTaskExecutor(Executor executor, ErrorHandler errorHandler) {
Assert.notNull(executor, "executor must not be null");
Assert.notNull(errorHandler, "errorHandler must not be null");
this.taskExecutor = taskExecutor;
this.executor = executor;
this.errorHandler = errorHandler;
}
public void execute(final Runnable task) {
this.taskExecutor.execute(new Runnable() {
this.executor.execute(new Runnable() {
public void run() {
try {
task.run();

17
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelParserTests.java
View File

@@ -25,13 +25,14 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.Executor;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.FatalBeanException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -114,13 +115,13 @@ public class ChannelParserTests {
assertEquals(PublishSubscribeChannel.class, channel.getClass());
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
accessor = new DirectFieldAccessor(accessor.getPropertyValue("dispatcher"));
Object taskExecutorProperty = accessor.getPropertyValue("taskExecutor");
assertNotNull(taskExecutorProperty);
assertEquals(ErrorHandlingTaskExecutor.class, taskExecutorProperty.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(taskExecutorProperty);
TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
Object taskExecutorBean = context.getBean("taskExecutor");
assertEquals(taskExecutorBean, innerExecutor);
Object executorProperty = accessor.getPropertyValue("executor");
assertNotNull(executorProperty);
assertEquals(ErrorHandlingTaskExecutor.class, executorProperty.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executorProperty);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
Object executorBean = context.getBean("taskExecutor");
assertEquals(executorBean, innerExecutor);
}
@Test

12
org.springframework.integration/src/test/java/org/springframework/integration/channel/config/DispatchingChannelParserTests.java
View File

@@ -72,10 +72,10 @@ public class DispatchingChannelParserTests {
public void taskExecutorOnly() {
MessageChannel channel = channels.get("taskExecutorOnly");
assertEquals(ExecutorChannel.class, channel.getClass());
Object executor = getDispatcherProperty("taskExecutor", channel);
Object executor = getDispatcherProperty("executor", channel);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
assertSame(context.getBean("taskExecutor"),
new DirectFieldAccessor(executor).getPropertyValue("taskExecutor"));
new DirectFieldAccessor(executor).getPropertyValue("executor"));
assertTrue((Boolean) getDispatcherProperty("failover", channel));
assertEquals(RoundRobinLoadBalancingStrategy.class,
getDispatcherProperty("loadBalancingStrategy", channel).getClass());
@@ -113,10 +113,10 @@ public class DispatchingChannelParserTests {
assertEquals(ExecutorChannel.class, channel.getClass());
assertTrue((Boolean) getDispatcherProperty("failover", channel));
assertNull(getDispatcherProperty("loadBalancingStrategy", channel));
Object executor = getDispatcherProperty("taskExecutor", channel);
Object executor = getDispatcherProperty("executor", channel);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
assertSame(context.getBean("taskExecutor"),
new DirectFieldAccessor(executor).getPropertyValue("taskExecutor"));
new DirectFieldAccessor(executor).getPropertyValue("executor"));
}
@Test
@@ -126,10 +126,10 @@ public class DispatchingChannelParserTests {
assertTrue((Boolean) getDispatcherProperty("failover", channel));
assertEquals(RoundRobinLoadBalancingStrategy.class,
getDispatcherProperty("loadBalancingStrategy", channel).getClass());
Object executor = getDispatcherProperty("taskExecutor", channel);
Object executor = getDispatcherProperty("executor", channel);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
assertSame(context.getBean("taskExecutor"),
new DirectFieldAccessor(executor).getPropertyValue("taskExecutor"));
new DirectFieldAccessor(executor).getPropertyValue("executor"));
}

19
org.springframework.integration/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2010 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,11 +22,12 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.Executor;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
@@ -47,7 +48,7 @@ public class PublishSubscribeChannelParserTests {
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertNull(dispatcherAccessor.getPropertyValue("taskExecutor"));
assertNull(dispatcherAccessor.getPropertyValue("executor"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
}
@@ -86,11 +87,11 @@ public class PublishSubscribeChannelParserTests {
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
TaskExecutor executor = (TaskExecutor) dispatcherAccessor.getPropertyValue("taskExecutor");
Executor executor = (Executor) dispatcherAccessor.getPropertyValue("executor");
assertNotNull(executor);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
}
@@ -105,11 +106,11 @@ public class PublishSubscribeChannelParserTests {
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertTrue((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
TaskExecutor executor = (TaskExecutor) dispatcherAccessor.getPropertyValue("taskExecutor");
Executor executor = (Executor) dispatcherAccessor.getPropertyValue("executor");
assertNotNull(executor);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
}
@@ -124,11 +125,11 @@ public class PublishSubscribeChannelParserTests {
accessor.getPropertyValue("dispatcher");
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertTrue((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
TaskExecutor executor = (TaskExecutor) dispatcherAccessor.getPropertyValue("taskExecutor");
Executor executor = (Executor) dispatcherAccessor.getPropertyValue("executor");
assertNotNull(executor);
assertEquals(ErrorHandlingTaskExecutor.class, executor.getClass());
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
TaskExecutor innerExecutor = (TaskExecutor) executorAccessor.getPropertyValue("taskExecutor");
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
}