Core Lambdas - Phase 1 src/main

Also package-protect private inner class ctors.

Core Lambdas - Phase 2 src/test
This commit is contained in:
Gary Russell
2016-10-26 13:37:03 -04:00
committed by Artem Bilan
parent c6026ce8fe
commit 9225c514fe
141 changed files with 1392 additions and 2128 deletions

View File

@@ -733,6 +733,10 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP
private class ForceReleaseMessageGroupProcessor implements MessageGroupProcessor {
ForceReleaseMessageGroupProcessor() {
super();
}
@Override
public Object processMessageGroup(MessageGroup group) {
forceComplete(group);

View File

@@ -122,7 +122,7 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
* interfaces for the annotation type as well (even if the annotation type
* is not marked as inherited itself)
*/
private MetaAnnotationMatchingPointcut(Class<? extends Annotation> classAnnotationType, boolean checkInherited) {
MetaAnnotationMatchingPointcut(Class<? extends Annotation> classAnnotationType, boolean checkInherited) {
this.classFilter = new AnnotationClassFilter(classAnnotationType, checkInherited);
this.methodMatcher = MethodMatcher.TRUE;
}
@@ -134,7 +134,7 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
* @param methodAnnotationType the annotation type to look for at the method level
* (can be <code>null</code>)
*/
private MetaAnnotationMatchingPointcut(
MetaAnnotationMatchingPointcut(
Class<? extends Annotation> classAnnotationType, Class<? extends Annotation> methodAnnotationType) {
Assert.isTrue((classAnnotationType != null || methodAnnotationType != null),
@@ -177,7 +177,7 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
* Create a new AnnotationClassFilter for the given annotation type.
* @param annotationType the annotation type to look for
*/
private MetaAnnotationMethodMatcher(Class<? extends Annotation> annotationType) {
MetaAnnotationMethodMatcher(Class<? extends Annotation> annotationType) {
super(annotationType);
this.annotationType = annotationType;
}

View File

@@ -235,7 +235,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
private final long expireAt;
private MessageChannelWrapper(MessageChannel channel, long expireAt) {
MessageChannelWrapper(MessageChannel channel, long expireAt) {
this.channel = channel;
this.expireAt = expireAt;
}

View File

@@ -20,13 +20,11 @@ import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.LoadBalancingStrategy;
import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageHandlingRunnable;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
@@ -120,18 +118,13 @@ public class ExecutorChannel extends AbstractExecutorChannel {
unicastingDispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);
}
unicastingDispatcher.setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() {
@Override
public Runnable decorate(MessageHandlingRunnable task) {
if (ExecutorChannel.this.executorInterceptorsSize > 0) {
return new MessageHandlingTask(task);
}
else {
return task;
}
unicastingDispatcher.setMessageHandlingTaskDecorator(task -> {
if (ExecutorChannel.this.executorInterceptorsSize > 0) {
return new MessageHandlingTask(task);
}
else {
return task;
}
});
this.dispatcher = unicastingDispatcher;

View File

@@ -112,7 +112,7 @@ public class PriorityChannel extends QueueChannel {
private final Comparator<Message<?>> targetComparator;
private SequenceFallbackComparator(Comparator<Message<?>> targetComparator) {
SequenceFallbackComparator(Comparator<Message<?>> targetComparator) {
this.targetComparator = targetComparator;
}
@@ -145,7 +145,7 @@ public class PriorityChannel extends QueueChannel {
private final Message<?> rootMessage;
private final long sequence;
private MessageWrapper(Message<?> rootMessage) {
MessageWrapper(Message<?> rootMessage) {
this.rootMessage = rootMessage;
this.sequence = PriorityChannel.this.sequenceCounter.incrementAndGet();
}

View File

@@ -20,10 +20,8 @@ import java.util.concurrent.Executor;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.support.MessageHandlingRunnable;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
@@ -158,18 +156,13 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel {
}
getDispatcher().setBeanFactory(this.getBeanFactory());
getDispatcher().setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() {
@Override
public Runnable decorate(MessageHandlingRunnable task) {
if (PublishSubscribeChannel.this.executorInterceptorsSize > 0) {
return new MessageHandlingTask(task);
}
else {
return task;
}
getDispatcher().setMessageHandlingTaskDecorator(task -> {
if (PublishSubscribeChannel.this.executorInterceptorsSize > 0) {
return new MessageHandlingTask(task);
}
else {
return task;
}
});
}

View File

@@ -99,7 +99,7 @@ public abstract class ThreadStatePropagationChannelInterceptor<S>
private final S state;
@SuppressWarnings("unchecked")
private MessageWithThreadState(Message<?> message, S state) {
MessageWithThreadState(Message<?> message, S state) {
this.message = (Message<Object>) message;
this.state = state;
}

View File

@@ -27,7 +27,6 @@ import org.springframework.util.Assert;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.pool.KryoCallback;
import com.esotericsoftware.kryo.pool.KryoFactory;
import com.esotericsoftware.kryo.pool.KryoPool;
@@ -43,13 +42,11 @@ public abstract class AbstractKryoCodec implements Codec {
protected final KryoPool pool;
protected AbstractKryoCodec() {
KryoFactory factory = new KryoFactory() {
public Kryo create() {
Kryo kryo = new Kryo();
// configure Kryo instance, customize settings
configureKryoInstance(kryo);
return kryo;
}
KryoFactory factory = () -> {
Kryo kryo = new Kryo();
// configure Kryo instance, customize settings
configureKryoInstance(kryo);
return kryo;
};
// Build pool with SoftReferences enabled (optional)
this.pool = new KryoPool.Builder(factory).softReferences().build();
@@ -60,13 +57,9 @@ public abstract class AbstractKryoCodec implements Codec {
Assert.notNull(object, "cannot encode a null object");
Assert.notNull(outputStream, "'outputSteam' cannot be null");
final Output output = (outputStream instanceof Output ? (Output) outputStream : new Output(outputStream));
this.pool.run(new KryoCallback<Object>() {
public Object execute(Kryo kryo) {
doEncode(kryo, object, output);
return Void.class;
}
this.pool.run(kryo -> {
doEncode(kryo, object, output);
return Void.class;
});
output.close();
}
@@ -90,13 +83,7 @@ public abstract class AbstractKryoCodec implements Codec {
final Input input = (inputStream instanceof Input ? (Input) inputStream : new Input(inputStream));
T result = null;
try {
result = this.pool.run(new KryoCallback<T>() {
public T execute(Kryo kryo) {
return doDecode(kryo, input, type);
}
});
result = this.pool.run(kryo -> doDecode(kryo, input, type));
}
finally {
input.close();

View File

@@ -89,7 +89,7 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
private final MessageHandler target;
private ReplyProducingMessageHandlerWrapper(MessageHandler target) {
ReplyProducingMessageHandlerWrapper(MessageHandler target) {
this.target = target;
}

View File

@@ -93,6 +93,10 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa
private class NamespaceHandlerDelegate extends NamespaceHandlerSupport {
NamespaceHandlerDelegate() {
super();
}
@Override
public void init() {
AbstractIntegrationNamespaceHandler.this.init();

View File

@@ -16,7 +16,6 @@
package org.springframework.integration.core;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
@@ -47,221 +46,119 @@ public class AsyncMessagingTemplate extends MessagingTemplate implements AsyncMe
@Override
public Future<?> asyncSend(final Message<?> message) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
send(message);
}
});
return this.executor.submit(() -> send(message));
}
@Override
public Future<?> asyncSend(final MessageChannel channel, final Message<?> message) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
send(channel, message);
}
});
return this.executor.submit(() -> send(channel, message));
}
@Override
public Future<?> asyncSend(final String channelName, final Message<?> message) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
send(channelName, message);
}
});
return this.executor.submit(() -> send(channelName, message));
}
@Override
public Future<?> asyncConvertAndSend(final Object object) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
convertAndSend(object);
}
});
return this.executor.submit(() -> convertAndSend(object));
}
@Override
public Future<?> asyncConvertAndSend(final MessageChannel channel, final Object object) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
convertAndSend(channel, object);
}
});
return this.executor.submit(() -> convertAndSend(channel, object));
}
@Override
public Future<?> asyncConvertAndSend(final String channelName, final Object object) {
return this.executor.submit(new Runnable() {
@Override
public void run() {
convertAndSend(channelName, object);
}
});
return this.executor.submit(() -> convertAndSend(channelName, object));
}
@Override
public Future<Message<?>> asyncReceive() {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return receive();
}
});
return this.executor.submit(() -> receive());
}
@Override
public Future<Message<?>> asyncReceive(final PollableChannel channel) {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return receive(channel);
}
});
return this.executor.submit(() -> receive(channel));
}
@Override
public Future<Message<?>> asyncReceive(final String channelName) {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return receive(channelName);
}
});
return this.executor.submit(() -> receive(channelName));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncReceiveAndConvert() {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) receiveAndConvert(null);
}
});
return this.executor.submit(() -> (R) receiveAndConvert(null));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncReceiveAndConvert(final PollableChannel channel) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) receiveAndConvert(channel, null);
}
});
return this.executor.submit(() -> (R) receiveAndConvert(channel, null));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncReceiveAndConvert(final String channelName) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) receiveAndConvert(channelName, null);
}
});
return this.executor.submit(() -> (R) receiveAndConvert(channelName, null));
}
@Override
public Future<Message<?>> asyncSendAndReceive(final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return sendAndReceive(requestMessage);
}
});
return this.executor.submit(() -> sendAndReceive(requestMessage));
}
@Override
public Future<Message<?>> asyncSendAndReceive(final MessageChannel channel, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return sendAndReceive(channel, requestMessage);
}
});
return this.executor.submit(() -> sendAndReceive(channel, requestMessage));
}
@Override
public Future<Message<?>> asyncSendAndReceive(final String channelName, final Message<?> requestMessage) {
return this.executor.submit(new Callable<Message<?>>() {
@Override
public Message<?> call() throws Exception {
return sendAndReceive(channelName, requestMessage);
}
});
return this.executor.submit(() -> sendAndReceive(channelName, requestMessage));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final Object request) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(request, null);
}
});
return this.executor.submit(() -> (R) convertSendAndReceive(request, null));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(channel, request, null);
}
});
return this.executor.submit(() -> (R) convertSendAndReceive(channel, request, null));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(channelName, request, null);
}
});
return this.executor.submit(() -> (R) convertSendAndReceive(channelName, request, null));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(request, null, requestPostProcessor);
}
});
public <R> Future<R> asyncConvertSendAndReceive(final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(() -> (R) convertSendAndReceive(request, null, requestPostProcessor));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(channel, request, null, requestPostProcessor);
}
});
public <R> Future<R> asyncConvertSendAndReceive(final MessageChannel channel, final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(() -> (R) convertSendAndReceive(channel, request, null, requestPostProcessor));
}
@Override
@SuppressWarnings("unchecked")
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request, final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(new Callable<R>() {
@Override
public R call() throws Exception {
return (R) convertSendAndReceive(channelName, request, null, requestPostProcessor);
}
});
public <R> Future<R> asyncConvertSendAndReceive(final String channelName, final Object request,
final MessagePostProcessor requestPostProcessor) {
return this.executor.submit(() -> (R) convertSendAndReceive(channelName, request, null, requestPostProcessor));
}
}

View File

@@ -66,15 +66,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
private volatile boolean messageBuilderFactorySet;
private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator =
new MessageHandlingTaskDecorator() {
@Override
public Runnable decorate(MessageHandlingRunnable task) {
return task;
}
};
private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator = task -> task;
private BeanFactory beanFactory;
@@ -202,14 +194,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
private Runnable createMessageHandlingTask(final MessageHandler handler, final Message<?> message) {
MessageHandlingRunnable task = new MessageHandlingRunnable() {
private final MessageHandler delegate = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
invokeHandler(handler, message);
}
};
private final MessageHandler delegate = message1 -> invokeHandler(handler, message1);
@Override
public void run() {

View File

@@ -25,7 +25,6 @@ import org.springframework.integration.MessageDispatchingException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageHandlingRunnable;
import org.springframework.util.Assert;
@@ -52,14 +51,7 @@ import org.springframework.util.Assert;
*/
public class UnicastingDispatcher extends AbstractDispatcher {
private final MessageHandler dispatchHandler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
doDispatch(message);
}
};
private final MessageHandler dispatchHandler = message -> doDispatch(message);
private final Executor executor;
@@ -67,15 +59,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
private volatile LoadBalancingStrategy loadBalancingStrategy;
private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator =
new MessageHandlingTaskDecorator() {
@Override
public Runnable decorate(MessageHandlingRunnable task) {
return task;
}
};
private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator = task -> task;
public UnicastingDispatcher() {
this.executor = null;

View File

@@ -183,13 +183,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
}
Callable<Boolean> pollingTask = new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
return doPoll();
}
};
Callable<Boolean> pollingTask = () -> doPoll();
List<Advice> adviceChain = this.adviceChain;
if (!CollectionUtils.isEmpty(adviceChain)) {
@@ -335,37 +329,32 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
private final Callable<Boolean> pollingTask;
private Poller(Callable<Boolean> pollingTask) {
Poller(Callable<Boolean> pollingTask) {
this.pollingTask = pollingTask;
}
@Override
public void run() {
AbstractPollingEndpoint.this.taskExecutor.execute(new Runnable() {
@Override
public void run() {
int count = 0;
while (AbstractPollingEndpoint.this.initialized
&& (AbstractPollingEndpoint.this.maxMessagesPerPoll <= 0
|| count < AbstractPollingEndpoint.this.maxMessagesPerPoll)) {
try {
if (!Poller.this.pollingTask.call()) {
break;
}
count++;
AbstractPollingEndpoint.this.taskExecutor.execute(() -> {
int count = 0;
while (AbstractPollingEndpoint.this.initialized
&& (AbstractPollingEndpoint.this.maxMessagesPerPoll <= 0
|| count < AbstractPollingEndpoint.this.maxMessagesPerPoll)) {
try {
if (!Poller.this.pollingTask.call()) {
break;
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessageHandlingException(new ErrorMessage(e), e);
}
count++;
}
catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException) e;
}
else {
throw new MessageHandlingException(new ErrorMessage(e), e);
}
}
}
});
}

View File

@@ -153,7 +153,7 @@ public class ReactiveConsumer extends AbstractEndpoint {
private Subscription actualSubscription;
private SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
SubscribableChannelPublisherAdapter(SubscribableChannel channel) {
this.channel = channel;
}

View File

@@ -64,14 +64,7 @@ import org.springframework.util.Assert;
*/
public final class ExpressionEvalMap extends AbstractMap<String, Object> {
public static final EvaluationCallback SIMPLE_CALLBACK = new EvaluationCallback() {
@Override
public Object evaluate(Expression expression) {
return expression.getValue();
}
};
public static final EvaluationCallback SIMPLE_CALLBACK = expression -> expression.getValue();
private final Map<String, ?> original;
@@ -187,6 +180,7 @@ public final class ExpressionEvalMap extends AbstractMap<String, Object> {
* Implementations of this interface can be provided to build 'on demand {@link #get(Object)} logic'
* for {@link ExpressionEvalMap}.
*/
@FunctionalInterface
public interface EvaluationCallback {
Object evaluate(Expression expression);
@@ -313,7 +307,7 @@ public final class ExpressionEvalMap extends AbstractMap<String, Object> {
}
@FunctionalInterface
public interface ExpressionEvalMapFinalBuilder {
ExpressionEvalMap build();

View File

@@ -537,7 +537,7 @@ public class ReloadableResourceBundleExpressionSource implements ExpressionSourc
* change detection, and the timestamp of the last refresh attempt
* (updated every time the cache entry gets re-validated).
*/
private final class PropertiesHolder {
private static final class PropertiesHolder {
private Properties properties;
@@ -545,13 +545,13 @@ public class ReloadableResourceBundleExpressionSource implements ExpressionSourc
private long refreshTimestamp = -1;
private PropertiesHolder(Properties properties, long fileTimestamp) {
this.properties = properties;
this.fileTimestamp = fileTimestamp;
PropertiesHolder() {
super();
}
private PropertiesHolder() {
PropertiesHolder(Properties properties, long fileTimestamp) {
this.properties = properties;
this.fileTimestamp = fileTimestamp;
}
public Properties getProperties() {

View File

@@ -69,7 +69,7 @@ public class GatewayCompletableFutureProxyFactoryBean extends GatewayProxyFactor
private final MethodInvocation invocation;
private Invoker(MethodInvocation methodInvocation) {
Invoker(MethodInvocation methodInvocation) {
this.invocation = methodInvocation;
}

View File

@@ -325,13 +325,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
}
this.serviceProxy = new ProxyFactory(proxyInterface, this).getProxy(this.beanClassLoader);
if (this.asyncExecutor != null) {
Callable<String> task = new Callable<String>() {
@Override
public String call() throws Exception {
return null;
}
};
Callable<String> task = () -> null;
Future<String> submitType = this.asyncExecutor.submit(task);
this.asyncSubmitType = submitType.getClass();
if (this.asyncExecutor instanceof AsyncListenableTaskExecutor) {
@@ -661,7 +655,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
private static final class MethodInvocationGateway extends MessagingGatewaySupport {
private MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) {
MethodInvocationGateway(GatewayMethodInboundMessageMapper messageMapper) {
this.setRequestMapper(messageMapper);
}
@@ -672,7 +666,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint
private final MethodInvocation invocation;
private AsyncInvocationTask(MethodInvocation invocation) {
AsyncInvocationTask(MethodInvocation invocation) {
this.invocation = invocation;
}

View File

@@ -563,6 +563,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
DefaultRequestMapper() {
super();
}
void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}

View File

@@ -173,6 +173,10 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
private class AdvisedRequestHandler implements RequestHandler {
AdvisedRequestHandler() {
super();
}
@Override
public Object handleRequestMessage(Message<?> requestMessage) {
return AbstractReplyProducingMessageHandler.this.handleRequestMessage(requestMessage);

View File

@@ -321,28 +321,16 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
if (this.messageStore instanceof SimpleMessageStore) {
final Message<?> messageToSchedule = delayedMessage;
releaseTask = new Runnable() {
@Override
public void run() {
releaseMessage(messageToSchedule);
}
};
releaseTask = () -> releaseMessage(messageToSchedule);
}
else {
final UUID messageId = delayedMessage.getHeaders().getId();
releaseTask = new Runnable() {
@Override
public void run() {
Message<?> messageToRelease = getMessageById(messageId);
if (messageToRelease != null) {
releaseMessage(messageToRelease);
}
releaseTask = () -> {
Message<?> messageToRelease = getMessageById(messageId);
if (messageToRelease != null) {
releaseMessage(messageToRelease);
}
};
}
@@ -417,21 +405,16 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
public synchronized void reschedulePersistedMessages() {
MessageGroup messageGroup = this.messageStore.getMessageGroup(this.messageGroupId);
for (final Message<?> message : messageGroup.getMessages()) {
getTaskScheduler().schedule(new Runnable() {
@Override
public void run() {
// This is fine to keep the reference to the message,
// because the scheduled task is performed immediately.
long delay = determineDelayForMessage(message);
if (delay > 0) {
releaseMessageAfterDelay(message, delay);
}
else {
releaseMessage(message);
}
getTaskScheduler().schedule((Runnable) () -> {
// This is fine to keep the reference to the message,
// because the scheduled task is performed immediately.
long delay = determineDelayForMessage(message);
if (delay > 0) {
releaseMessageAfterDelay(message, delay);
}
else {
releaseMessage(message);
}
}, new Date());
}
}
@@ -465,6 +448,10 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
*/
private class ReleaseMessageHandler implements MessageHandler {
ReleaseMessageHandler() {
super();
}
@Override
public void handleMessage(Message<?> message) throws MessagingException {
DelayHandler.this.doReleaseMessage(message);

View File

@@ -84,7 +84,7 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor<
private final MethodFilter methodFilter;
private ExpressionCommandMethodResolver(MethodFilter methodFilter) {
ExpressionCommandMethodResolver(MethodFilter methodFilter) {
this.methodFilter = methodFilter;
}

View File

@@ -232,7 +232,11 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler impleme
}
}
private class ReplyForwardingMessageChannel implements MessageChannel {
private final class ReplyForwardingMessageChannel implements MessageChannel {
ReplyForwardingMessageChannel() {
super();
}
@Override
public boolean send(Message<?> message) {

View File

@@ -182,9 +182,9 @@ public abstract class AbstractRequestHandlerAdvice extends IntegrationObjectSupp
}
@SuppressWarnings("serial")
private final class ThrowableHolderException extends RuntimeException {
private static final class ThrowableHolderException extends RuntimeException {
private ThrowableHolderException(Throwable cause) {
ThrowableHolderException(Throwable cause) {
super(cause);
}

View File

@@ -79,6 +79,10 @@ public class RequestHandlerCircuitBreakerAdvice extends AbstractRequestHandlerAd
private volatile long lastFailure;
AdvisedMetadata() {
super();
}
private long getLastFailure() {
return this.lastFailure;
}
@@ -99,7 +103,7 @@ public class RequestHandlerCircuitBreakerAdvice extends AbstractRequestHandlerAd
private static final long serialVersionUID = 1L;
private CircuitBreakerOpenException(String message) {
CircuitBreakerOpenException(String message) {
super(message);
}

View File

@@ -48,15 +48,7 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice
private static final ThreadLocal<Message<?>> messageHolder = new ThreadLocal<Message<?>>();
// Stateless unless a state generator is provided
private volatile RetryStateGenerator retryStateGenerator =
new RetryStateGenerator() {
@Override
public RetryState determineRetryState(Message<?> message) {
return null;
}
};
private volatile RetryStateGenerator retryStateGenerator = message -> null;
public void setRetryTemplate(RetryTemplate retryTemplate) {
Assert.notNull(retryTemplate, "'retryTemplate' cannot be null");
@@ -86,14 +78,7 @@ public class RequestHandlerRetryAdvice extends AbstractRequestHandlerAdvice
messageHolder.set(message);
try {
return this.retryTemplate.execute(new RetryCallback<Object, Exception>() {
@Override
public Object doWithRetry(RetryContext context) throws Exception {
return callback.cloneAndExecute();
}
}, this.recoveryCallback, retryState);
return this.retryTemplate.execute(context -> callback.cloneAndExecute(), this.recoveryCallback, retryState);
}
catch (MessagingException e) {
if (e.getFailedMessage() == null) {

View File

@@ -26,7 +26,9 @@ import org.springframework.retry.RetryState;
* @since 2.2
*
*/
@FunctionalInterface
public interface RetryStateGenerator {
RetryState determineRetryState(Message<?> message);
}

View File

@@ -107,26 +107,21 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
this.gatherEndpoint.afterPropertiesSet();
}
((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageHeaders headers = message.getHeaders();
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
if (gatherResultChannel instanceof MessageChannel) {
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
}
else if (gatherResultChannel instanceof String) {
messagingTemplate.send((String) gatherResultChannel, message);
}
((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(message -> {
MessageHeaders headers = message.getHeaders();
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
if (gatherResultChannel instanceof MessageChannel) {
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
}
else {
throw new MessageDeliveryException(message,
"The 'gatherResultChannel' header is required to delivery gather result.");
else if (gatherResultChannel instanceof String) {
messagingTemplate.send((String) gatherResultChannel, message);
}
}
else {
throw new MessageDeliveryException(message,
"The 'gatherResultChannel' header is required to delivery gather result.");
}
}));
this.replyChannelRegistry = getBeanFactory()

View File

@@ -62,7 +62,11 @@ public class PollSkipAdvice implements MethodInterceptor {
}
private static class DefaultPollSkipStrategy implements PollSkipStrategy {
private static final class DefaultPollSkipStrategy implements PollSkipStrategy {
DefaultPollSkipStrategy() {
super();
}
@Override
public boolean skipPoll() {

View File

@@ -335,7 +335,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
private final Iterator<?> idIterator;
private MessageGroupIterator(Iterator<?> idIterator) {
MessageGroupIterator(Iterator<?> idIterator) {
this.idIterator = idIterator;
}

View File

@@ -35,7 +35,7 @@ class PersistentMessageGroup implements MessageGroup {
private static final Log logger = LogFactory.getLog(PersistentMessageGroup.class);
private MessageGroupStore messageGroupStore;
private final MessageGroupStore messageGroupStore;
private final Collection<Message<?>> messages = new PersistentCollection();
@@ -168,10 +168,14 @@ class PersistentMessageGroup implements MessageGroup {
}
private class PersistentCollection extends AbstractCollection<Message<?>> {
private final class PersistentCollection extends AbstractCollection<Message<?>> {
private volatile Collection<Message<?>> collection;
PersistentCollection() {
super();
}
private void load() {
if (this.collection == null) {
synchronized (this) {

View File

@@ -18,7 +18,6 @@ package org.springframework.integration.support;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
@@ -133,15 +132,8 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
List<SmartLifecycle> lifecycles = this.lifecycles.get(role);
if (lifecycles != null) {
lifecycles = new ArrayList<SmartLifecycle>(lifecycles);
Collections.sort(lifecycles, new Comparator<SmartLifecycle>() {
@Override
public int compare(SmartLifecycle o1, SmartLifecycle o2) {
return o1.getPhase() < o2.getPhase() ? -1
: o1.getPhase() > o2.getPhase() ? 1 : 0;
}
});
Collections.sort(lifecycles, (o1, o2) ->
o1.getPhase() < o2.getPhase() ? -1 : o1.getPhase() > o2.getPhase() ? 1 : 0);
if (logger.isDebugEnabled()) {
logger.debug("Zookeeper leadership granted: Starting: " + lifecycles);
}
@@ -172,15 +164,8 @@ public class SmartLifecycleRoleController implements ApplicationListener<Abstrac
List<SmartLifecycle> lifecycles = this.lifecycles.get(role);
if (lifecycles != null) {
lifecycles = new ArrayList<SmartLifecycle>(lifecycles);
Collections.sort(lifecycles, new Comparator<SmartLifecycle>() {
@Override
public int compare(SmartLifecycle o1, SmartLifecycle o2) {
return o1.getPhase() < o2.getPhase() ? 1
: o1.getPhase() > o2.getPhase() ? -1 : 0;
}
});
Collections.sort(lifecycles, (o1, o2) ->
o1.getPhase() < o2.getPhase() ? 1 : o1.getPhase() > o2.getPhase() ? -1 : 0);
if (logger.isDebugEnabled()) {
logger.debug("Zookeeper leadership revoked: Stopping: " + lifecycles);
}

View File

@@ -123,6 +123,10 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar
private class DefaultInboundMessageMapper implements InboundMessageMapper<Object> {
DefaultInboundMessageMapper() {
super();
}
@Override
public Message<?> toMessage(Object object) throws Exception {
if (object == null) {
@@ -139,6 +143,10 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar
private class DefaultOutboundMessageMapper implements OutboundMessageMapper<Object> {
DefaultOutboundMessageMapper() {
super();
}
@Override
public Object fromMessage(Message<?> message) throws Exception {
return (message != null) ? message.getPayload() : null;

View File

@@ -91,13 +91,7 @@ public class BoonJsonObjectMapper extends JsonObjectMapperAdapter<Map<String, Ob
public Map<String, Object> toJsonNode(final Object value) throws Exception {
PipedReader in = new PipedReader();
final PipedWriter out = new PipedWriter(in);
Executors.newSingleThreadExecutor()
.execute(new Runnable() {
@Override
public void run() {
toJson(value, out);
}
});
Executors.newSingleThreadExecutor().execute(() -> toJson(value, out));
return (Map<String, Object>) this.slurper.parse(in);
}

View File

@@ -386,6 +386,10 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
*/
private class LockContext implements Context {
LockContext() {
super();
}
@Override
public boolean isLeader() {
return LockRegistryLeaderInitiator.this.leaderSelector.isLeader();
@@ -412,6 +416,10 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
private static final class NullContext implements Context {
NullContext() {
super();
}
@Override
public boolean isLeader() {
return false;

View File

@@ -267,6 +267,10 @@ public class IntegrationGraphServer implements ApplicationContextAware, Applicat
private final AtomicInteger nodeId = new AtomicInteger();
NodeFactory() {
super();
}
private MessageChannelNode channelNode(String name, MessageChannel channel) {
return new MessageChannelNode(this.nodeId.incrementAndGet(), name, channel);
}

View File

@@ -39,7 +39,7 @@ public class MessageChannelNode extends IntegrationNode {
private final MessageChannelMetrics channel;
private Stats(MessageChannelMetrics channel) {
Stats(MessageChannelMetrics channel) {
this.channel = channel;
}

View File

@@ -36,7 +36,7 @@ public class MessageGatewayNode extends ErrorCapableEndpointNode {
private final MessagingGatewaySupport gateway;
private Stats(MessagingGatewaySupport gateway) {
Stats(MessagingGatewaySupport gateway) {
this.gateway = gateway;
}

View File

@@ -45,7 +45,7 @@ public class MessageHandlerNode extends EndpointNode {
private final MessageHandlerMetrics handler;
private Stats(MessageHandlerMetrics handler) {
Stats(MessageHandlerMetrics handler) {
this.handler = handler;
}

View File

@@ -38,7 +38,7 @@ public class MessageSourceNode extends ErrorCapableEndpointNode {
private final MessageSourceMetrics source;
private Stats(MessageSourceMetrics source) {
Stats(MessageSourceMetrics source) {
this.source = source;
}

View File

@@ -55,7 +55,7 @@ public class DefaultTransactionSynchronizationFactory implements TransactionSync
*/
private final class DefaultTransactionalResourceSynchronization extends IntegrationResourceHolderSynchronization {
private DefaultTransactionalResourceSynchronization(Object resourceKey) {
DefaultTransactionalResourceSynchronization(Object resourceKey) {
super(new IntegrationResourceHolder(), resourceKey);
}

View File

@@ -458,6 +458,10 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler
*/
private static final class Gateway extends MessagingGatewaySupport {
Gateway() {
super();
}
@Override
protected Message<?> sendAndReceiveMessage(Object object) {
return super.sendAndReceiveMessage(object);

View File

@@ -48,15 +48,12 @@ public class ErrorHandlingTaskExecutor implements TaskExecutor {
@Override
public void execute(final Runnable task) {
this.executor.execute(new Runnable() {
@Override
public void run() {
try {
task.run();
}
catch (Throwable t) { //NOSONAR
ErrorHandlingTaskExecutor.this.errorHandler.handleError(t);
}
this.executor.execute(() -> {
try {
task.run();
}
catch (Throwable t) { //NOSONAR
ErrorHandlingTaskExecutor.this.errorHandler.handleError(t);
}
});
}

View File

@@ -78,20 +78,10 @@ public final class MessagingAnnotationUtils {
public static Method findAnnotatedMethod(Object target, final Class<? extends Annotation> annotationType) {
final AtomicReference<Method> reference = new AtomicReference<Method>();
ReflectionUtils.doWithMethods(getTargetClass(target), new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
reference.compareAndSet(null, method);
}
}, new ReflectionUtils.MethodFilter() {
@Override
public boolean matches(Method method) {
return ReflectionUtils.USER_DECLARED_METHODS.matches(method) &&
AnnotatedElementUtils.isAnnotated(method, annotationType.getName());
}
});
ReflectionUtils.doWithMethods(getTargetClass(target),
method -> reference.compareAndSet(null, method),
method -> ReflectionUtils.USER_DECLARED_METHODS.matches(method) &&
AnnotatedElementUtils.isAnnotated(method, annotationType.getName()));
return reference.get();
}

View File

@@ -914,7 +914,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
@SuppressWarnings("serial")
private static final class IneligibleMethodException extends RuntimeException {
private IneligibleMethodException(String message) {
IneligibleMethodException(String message) {
super(message);
}

View File

@@ -63,45 +63,34 @@ public class AbstractCorrelatingMessageHandlerTests {
final CountDownLatch waitForSendLatch = new CountDownLatch(1);
final CountDownLatch waitReapStartLatch = new CountDownLatch(1);
final CountDownLatch waitReapCompleteLatch = new CountDownLatch(1);
AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
}, groupStore) {
AbstractCorrelatingMessageHandler handler = new AbstractCorrelatingMessageHandler(group -> group, groupStore) {
@Override
protected void afterRelease(MessageGroup group, Collection<Message<?>> completedMessages) {
}
};
handler.setReleasePartialSequences(true);
/*
* Runs "reap" when group 'bar' is in completion
*/
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
waitReapStartLatch.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
waitForSendLatch.countDown();
try {
Thread.sleep(100);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
groupStore.expireMessageGroups(50);
waitReapCompleteLatch.countDown();
Executors.newSingleThreadExecutor().execute(() -> {
try {
waitReapStartLatch.await(10, TimeUnit.SECONDS);
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
waitForSendLatch.countDown();
try {
Thread.sleep(100);
}
catch (InterruptedException e2) {
Thread.currentThread().interrupt();
}
groupStore.expireMessageGroups(50);
waitReapCompleteLatch.countDown();
});
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
@@ -133,14 +122,7 @@ public class AbstractCorrelatingMessageHandlerTests {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 2;
}
});
handler.setReleaseStrategy(group -> group.size() == 2);
QueueChannel discards = new QueueChannel();
handler.setDiscardChannel(discards);
@@ -175,15 +157,7 @@ public class AbstractCorrelatingMessageHandlerTests {
@Test // INT-2833
public void testReaperReapsAnEmptyGroup() throws Exception {
final MessageGroupStore groupStore = new SimpleMessageStore();
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
}, groupStore) {
};
AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
@@ -202,13 +176,7 @@ public class AbstractCorrelatingMessageHandlerTests {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
});
handler.setReleaseStrategy(group -> group.size() == 1);
Message<String> message = MessageBuilder.withPayload("foo")
.setCorrelationId("bar")
@@ -225,15 +193,7 @@ public class AbstractCorrelatingMessageHandlerTests {
@Test // INT-2833
public void testReaperReapsAnEmptyGroupAfterConfiguredDelay() throws Exception {
final MessageGroupStore groupStore = new SimpleMessageStore();
AggregatingMessageHandler handler = new AggregatingMessageHandler(
new MessageGroupProcessor() {
@Override
public Object processMessageGroup(MessageGroup group) {
return group;
}
}, groupStore) {
};
AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore);
final List<Message<?>> outputMessages = new ArrayList<Message<?>>();
handler.setOutputChannel(new MessageChannel() {
@@ -252,13 +212,7 @@ public class AbstractCorrelatingMessageHandlerTests {
return this.send(message, 0);
}
});
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return group.size() == 1;
}
});
handler.setReleaseStrategy(group -> group.size() == 1);
handler.setMinimumTimeoutForEmptyGroups(1000);
@@ -281,12 +235,7 @@ public class AbstractCorrelatingMessageHandlerTests {
public void testReapWithChangeInSameMillisecond() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
handler.setReleaseStrategy(group -> true);
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -312,14 +261,7 @@ public class AbstractCorrelatingMessageHandlerTests {
public void testDontReapIfAlreadyComplete() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
handler.setReleaseStrategy(group -> true);
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -345,14 +287,7 @@ public class AbstractCorrelatingMessageHandlerTests {
public void testDontReapIfAlreadyCompleteAfterRefetch() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
handler.setReleaseStrategy(group -> true);
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -381,14 +316,7 @@ public class AbstractCorrelatingMessageHandlerTests {
public void testDontReapIfNewGroupFoundDuringRefetch() throws Exception {
MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor();
AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
handler.setReleaseStrategy(group -> true);
QueueChannel outputChannel = new QueueChannel();
handler.setOutputChannel(outputChannel);
MessageGroupStore mgs = TestUtils.getPropertyValue(handler, "messageStore", MessageGroupStore.class);
@@ -418,14 +346,7 @@ public class AbstractCorrelatingMessageHandlerTests {
handler.setOutputChannel(new QueueChannel());
QueueChannel discardChannel = new QueueChannel();
handler.setDiscardChannel(discardChannel);
handler.setReleaseStrategy(new ReleaseStrategy() {
@Override
public boolean canRelease(MessageGroup group) {
return true;
}
});
handler.setReleaseStrategy(group -> true);
handler.setExpireGroupsUponTimeout(false);
SimpleMessageStore messageStore = new SimpleMessageStore() {
@Override
@@ -447,16 +368,11 @@ public class AbstractCorrelatingMessageHandlerTests {
//suppress an intentional 'removeMessageGroup' exception
}
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.execute(new Runnable() {
@Override
public void run() {
handler.handleMessage(MessageBuilder.withPayload("foo")
.setCorrelationId(1)
.setSequenceNumber(2)
.setSequenceSize(2)
.build());
}
});
executorService.execute(() -> handler.handleMessage(MessageBuilder.withPayload("foo")
.setCorrelationId(1)
.setSequenceNumber(2)
.setSequenceSize(2)
.build()));
executorService.shutdown();
/* Previously lock for the groupId hasn't been unlocked from the 'forceComplete', because it wasn't
reachable in case of exception from the BasicMessageGroupStore.removeMessageGroup

View File

@@ -271,6 +271,10 @@ public class AggregatingMessageGroupProcessorHeaderTests {
private static class TestAggregatorBean {
TestAggregatorBean() {
super();
}
@SuppressWarnings("unused")
public Object aggregate(List<String> payloads) {
StringBuilder sb = new StringBuilder();

View File

@@ -40,8 +40,6 @@ import org.junit.Ignore;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.AbstractMessageHandler;
@@ -51,10 +49,8 @@ import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.StopWatch;
@@ -79,19 +75,7 @@ public class AggregatorTests {
public void configureAggregator() {
this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), store);
this.aggregator.setBeanFactory(mock(BeanFactory.class));
this.aggregator.setApplicationEventPublisher(new ApplicationEventPublisher() {
@Override
public void publishEvent(ApplicationEvent event) {
expiryEvents.add((MessageGroupExpiredEvent) event);
}
@Override
public void publishEvent(Object event) {
}
});
this.aggregator.setApplicationEventPublisher(event -> expiryEvents.add((MessageGroupExpiredEvent) event));
this.aggregator.setBeanName("testAggregator");
this.aggregator.afterPropertiesSet();
expiryEvents.clear();
@@ -100,14 +84,7 @@ public class AggregatorTests {
@Test
public void testAggPerf() throws InterruptedException, ExecutionException, TimeoutException {
AggregatingMessageHandler handler = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor());
handler.setCorrelationStrategy(new CorrelationStrategy() {
@Override
public Object getCorrelationKey(Message<?> message) {
return "foo";
}
});
handler.setCorrelationStrategy(message -> "foo");
handler.setReleaseStrategy(new MessageCountReleaseStrategy(60000));
handler.setExpireGroupsUponCompletion(true);
handler.setSendPartialResultOnExpiry(true);
@@ -115,15 +92,10 @@ public class AggregatorTests {
handler.setOutputChannel(outputChannel);
final CompletableFuture<Collection<?>> resultFuture = new CompletableFuture<>();
outputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Collection<?> payload = (Collection<?>) message.getPayload();
logger.warn("Received " + payload.size());
resultFuture.complete(payload);
}
outputChannel.subscribe(message -> {
Collection<?> payload = (Collection<?>) message.getPayload();
logger.warn("Received " + payload.size());
resultFuture.complete(payload);
});
SimpleMessageStore store = new SimpleMessageStore();
@@ -200,15 +172,10 @@ public class AggregatorTests {
CustomHandler handler = new CustomHandler(outputChannel);
final CompletableFuture<Collection<?>> resultFuture = new CompletableFuture<>();
outputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Collection<?> payload = (Collection<?>) message.getPayload();
logger.warn("Received " + payload.size());
resultFuture.complete(payload);
}
outputChannel.subscribe(message -> {
Collection<?> payload = (Collection<?>) message.getPayload();
logger.warn("Received " + payload.size());
resultFuture.complete(payload);
});
Message<?> message = new GenericMessage<String>("foo");
StopWatch stopwatch = new StopWatch();
@@ -441,6 +408,11 @@ public class AggregatorTests {
private class MultiplyingProcessor implements MessageGroupProcessor {
MultiplyingProcessor() {
super();
}
@Override
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;

View File

@@ -341,6 +341,10 @@ public class ConcurrentAggregatorTests {
private class MultiplyingProcessor implements MessageGroupProcessor {
MultiplyingProcessor() {
super();
}
@Override
public Object processMessageGroup(MessageGroup group) {
Integer product = 1;
@@ -356,6 +360,10 @@ public class ConcurrentAggregatorTests {
@SuppressWarnings("unused")
private class NullReturningMessageProcessor implements MessageGroupProcessor {
NullReturningMessageProcessor() {
super();
}
@Override
public Object processMessageGroup(MessageGroup group) {
return null;

View File

@@ -109,18 +109,15 @@ public class CorrelatingMessageBarrierTests {
}
private void sendAsynchronously(final MessageHandler handler, final Message<Object> message, final CountDownLatch start, final CountDownLatch sent) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
handler.handleMessage(message);
sent.countDown();
Executors.newSingleThreadExecutor().execute(() -> {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
handler.handleMessage(message);
sent.countDown();
});
}
@@ -134,8 +131,13 @@ public class CorrelatingMessageBarrierTests {
* ReleaseStrategy that emulates the use case described in INT-1068
*/
private static class OneMessagePerKeyReleaseStrategy implements ReleaseStrategy {
private final ConcurrentMap<Object, Semaphore> keyLocks = new ConcurrentHashMap<Object, Semaphore>();
OneMessagePerKeyReleaseStrategy() {
super();
}
@Override
public boolean canRelease(MessageGroup messageGroup) {
Object correlationKey = messageGroup.getGroupId();

View File

@@ -146,12 +146,10 @@ public class CorrelatingMessageHandlerTests {
handler.handleMessage(message1);
bothMessagesHandled.countDown();
storedMessages.add(message1);
Executors.newSingleThreadExecutor().submit(new Runnable() {
public void run() {
handler.handleMessage(message2);
storedMessages.add(message2);
bothMessagesHandled.countDown();
}
Executors.newSingleThreadExecutor().submit(() -> {
handler.handleMessage(message2);
storedMessages.add(message2);
bothMessagesHandled.countDown();
});
assertTrue(bothMessagesHandled.await(10, TimeUnit.SECONDS));

View File

@@ -75,6 +75,11 @@ public class CorrelationStrategyAdapterTests {
}
private static class MultiHeaderCorrelator {
MultiHeaderCorrelator() {
super();
}
@SuppressWarnings("unused")
public String getKey(@Header("a") String header, @Header("c") String other) {
return header + other;
@@ -83,6 +88,11 @@ public class CorrelationStrategyAdapterTests {
}
private static class SimpleHeaderCorrelator {
SimpleHeaderCorrelator() {
super();
}
@SuppressWarnings("unused")
public String getKey(@Header("a") String header) {
return header;
@@ -91,6 +101,11 @@ public class CorrelationStrategyAdapterTests {
}
private static class SimplePojoCorrelator {
SimplePojoCorrelator() {
super();
}
@SuppressWarnings("unused")
public String getKey(String message) {
return message;
@@ -99,6 +114,11 @@ public class CorrelationStrategyAdapterTests {
}
private static class SimpleMessageCorrelator {
SimpleMessageCorrelator() {
super();
}
@SuppressWarnings("unused")
public String getKey(Message<?> message) {
return (String) message.getHeaders().get("a");

View File

@@ -113,12 +113,7 @@ public class ResequencerTests {
releaseStrategy.setReleasePartialSequences(true);
this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy);
QueueChannel replyChannel = new QueueChannel();
this.resequencer.setCorrelationStrategy(new CorrelationStrategy() {
@Override
public Object getCorrelationKey(Message<?> message) {
return "A";
}
});
this.resequencer.setCorrelationStrategy(message -> "A");
this.resequencer.setBeanFactory(mock(BeanFactory.class));
this.resequencer.afterPropertiesSet();

View File

@@ -16,9 +16,14 @@
package org.springframework.integration.aggregator.integration;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.util.List;
import org.junit.Test;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.DefaultAggregatingMessageGroupProcessor;
import org.springframework.integration.aggregator.ReleaseStrategy;
@@ -28,10 +33,6 @@ import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
/**
* @author Oleg Zhurakousky
* @author Artem Bilan
@@ -39,11 +40,11 @@ import static org.junit.Assert.assertNull;
*/
public class AggregatorSupportedUseCasesTests {
private MessageGroupStore store = new SimpleMessageStore(100);
private final MessageGroupStore store = new SimpleMessageStore(100);
private DefaultAggregatingMessageGroupProcessor processor = new DefaultAggregatingMessageGroupProcessor();
private final DefaultAggregatingMessageGroupProcessor processor = new DefaultAggregatingMessageGroupProcessor();
private AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store);
private final AggregatingMessageHandler defaultHandler = new AggregatingMessageHandler(processor, store);
@Test
public void waitForAllDefaultReleaseStrategyWithLateArrivals() {
@@ -151,6 +152,11 @@ public class AggregatorSupportedUseCasesTests {
private class SampleSizeReleaseStrategy implements ReleaseStrategy {
SampleSizeReleaseStrategy() {
super();
}
@Override
public boolean canRelease(MessageGroup group) {
return group.getMessages().size() == 5;
}
@@ -159,6 +165,11 @@ public class AggregatorSupportedUseCasesTests {
private class FirstBestReleaseStrategy implements ReleaseStrategy {
FirstBestReleaseStrategy() {
super();
}
@Override
public boolean canRelease(MessageGroup group) {
return true;
}

View File

@@ -108,29 +108,20 @@ public class AggregatorWithCustomReleaseStrategyTests {
for (int i = 0; i < 600; i++) {
final int counter = i;
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload("foo").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("foo").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload("bar").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("bar").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload("baz").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload("baz").
setHeader("correlation", "foo" + counter).build());
latch.countDown();
});
}
@@ -155,26 +146,17 @@ public class AggregatorWithCustomReleaseStrategyTests {
final CountDownLatch latch = new CountDownLatch(1800);
for (int i = 0; i < 600; i++) {
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{1, 2, 3, 4, 5, 6, 7, 8}).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{1, 2, 3, 4, 5, 6, 7, 8}).build());
latch.countDown();
});
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{9, 10, 11, 12, 13, 14, 15, 16}).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{9, 10, 11, 12, 13, 14, 15, 16}).build());
latch.countDown();
});
executor.execute(new Runnable() {
@Override
public void run() {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{17, 18, 19, 20, 21, 22, 23, 24}).build());
latch.countDown();
}
executor.execute(() -> {
inputChannel.send(MessageBuilder.withPayload(new Integer[]{17, 18, 19, 20, 21, 22, 23, 24}).build());
latch.countDown();
});
}

View File

@@ -32,8 +32,6 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -57,12 +55,7 @@ public class PartialSequencesWithGapsTests {
@Before
public void collectOutput() {
out.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
received.add(message);
}
});
out.subscribe(message -> received.add(message));
}
@Test

View File

@@ -109,6 +109,7 @@ public class MessagePublishingInterceptorTests {
static class TestBeanImpl implements TestBean {
@Override
public String test() {
return "foo";
}
@@ -118,14 +119,21 @@ public class MessagePublishingInterceptorTests {
private static class TestPublisherMetadataSource implements PublisherMetadataSource {
TestPublisherMetadataSource() {
super();
}
@Override
public String getPayloadExpression(Method method) {
return "'test-' + #return";
}
@Override
public Map<String, String> getHeaderExpressions(Method method) {
return null;
}
@Override
public String getChannelName(Method method) {
return "c";
}

View File

@@ -25,7 +25,6 @@ import java.util.List;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.integration.core.MessageSelector;
import org.springframework.messaging.support.GenericMessage;
/**
@@ -51,11 +50,7 @@ public class ChannelPurgerTests {
channel.send(new GenericMessage<String>("test1"));
channel.send(new GenericMessage<String>("test2"));
channel.send(new GenericMessage<String>("test3"));
ChannelPurger purger = new ChannelPurger(new MessageSelector() {
public boolean accept(Message<?> message) {
return false;
}
}, channel);
ChannelPurger purger = new ChannelPurger(message -> false, channel);
List<Message<?>> purgedMessages = purger.purge();
assertEquals(3, purgedMessages.size());
assertNull(channel.receive(0));
@@ -67,11 +62,7 @@ public class ChannelPurgerTests {
channel.send(new GenericMessage<String>("test1"));
channel.send(new GenericMessage<String>("test2"));
channel.send(new GenericMessage<String>("test3"));
ChannelPurger purger = new ChannelPurger(new MessageSelector() {
public boolean accept(Message<?> message) {
return true;
}
}, channel);
ChannelPurger purger = new ChannelPurger(message -> true, channel);
List<Message<?>> purgedMessages = purger.purge();
assertEquals(0, purgedMessages.size());
assertNotNull(channel.receive(0));
@@ -85,11 +76,7 @@ public class ChannelPurgerTests {
channel.send(new GenericMessage<String>("test1"));
channel.send(new GenericMessage<String>("test2"));
channel.send(new GenericMessage<String>("test3"));
ChannelPurger purger = new ChannelPurger(new MessageSelector() {
public boolean accept(Message<?> message) {
return (message.getPayload().equals("test2"));
}
}, channel);
ChannelPurger purger = new ChannelPurger(message -> (message.getPayload().equals("test2")), channel);
List<Message<?>> purgedMessages = purger.purge();
assertEquals(2, purgedMessages.size());
Message<?> message = channel.receive(0);
@@ -123,11 +110,7 @@ public class ChannelPurgerTests {
channel2.send(new GenericMessage<String>("test1"));
channel2.send(new GenericMessage<String>("test2"));
channel2.send(new GenericMessage<String>("test3"));
ChannelPurger purger = new ChannelPurger(new MessageSelector() {
public boolean accept(Message<?> message) {
return (message.getPayload().equals("test2"));
}
}, channel1, channel2);
ChannelPurger purger = new ChannelPurger(message -> (message.getPayload().equals("test2")), channel1, channel2);
List<Message<?>> purgedMessages = purger.purge();
assertEquals(4, purgedMessages.size());
Message<?> message1 = channel1.receive(0);
@@ -148,11 +131,7 @@ public class ChannelPurgerTests {
channel1.send(new GenericMessage<String>("test2"));
channel2.send(new GenericMessage<String>("test1"));
channel2.send(new GenericMessage<String>("test2"));
ChannelPurger purger = new ChannelPurger(new MessageSelector() {
public boolean accept(Message<?> message) {
return true;
}
}, channel1, channel2);
ChannelPurger purger = new ChannelPurger(message -> true, channel1, channel2);
List<Message<?>> purgedMessages = purger.purge();
assertEquals(0, purgedMessages.size());
assertNotNull(channel1.receive(0));

View File

@@ -233,18 +233,34 @@ public class DatatypeChannelTests {
private static class Foo {
Foo() {
super();
}
}
private static class Bar extends Foo {
Bar() {
super();
}
}
private static class Baz extends Foo {
Baz() {
super();
}
}
private static class StringToBarConverter implements GenericConverter {
StringToBarConverter() {
super();
}
@Override
public Set<ConvertiblePair> getConvertibleTypes() {
Set<ConvertiblePair> pairs = new HashSet<ConvertiblePair>();
@@ -262,6 +278,10 @@ public class DatatypeChannelTests {
private static class IntegerToBazConverter implements GenericConverter {
IntegerToBazConverter() {
super();
}
@Override
public Set<ConvertiblePair> getConvertibleTypes() {
Set<ConvertiblePair> pairs = new HashSet<ConvertiblePair>();

View File

@@ -45,7 +45,6 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ReflectionUtils;
@@ -93,13 +92,7 @@ public class DirectChannelTests {
*/
DirectChannel channel = new DirectChannel();
final AtomicInteger count = new AtomicInteger();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count.incrementAndGet();
}
});
channel.subscribe(message -> count.incrementAndGet());
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 10000000; i++) {
@@ -119,20 +112,8 @@ public class DirectChannelTests {
DirectChannel channel = new DirectChannel();
final AtomicInteger count1 = new AtomicInteger();
final AtomicInteger count2 = new AtomicInteger();
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count1.incrementAndGet();
}
});
channel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count2.getAndIncrement();
}
});
channel.subscribe(message -> count1.incrementAndGet());
channel.subscribe(message -> count2.getAndIncrement());
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 10000000; i++) {
@@ -152,13 +133,7 @@ public class DirectChannelTests {
* Added the same code to the other tests for comparison.
*/
final AtomicInteger count = new AtomicInteger();
FixedSubscriberChannel channel = new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
count.incrementAndGet();
}
});
FixedSubscriberChannel channel = new FixedSubscriberChannel(message -> count.incrementAndGet());
GenericMessage<String> message = new GenericMessage<String>("test");
assertTrue(channel.send(message));
for (int i = 0; i < 100000000; i++) {
@@ -173,12 +148,7 @@ public class DirectChannelTests {
ThreadNameExtractingTestTarget target = new ThreadNameExtractingTestTarget(latch);
channel.subscribe(target);
final GenericMessage<String> message = new GenericMessage<String>("test");
new Thread(new Runnable() {
@Override
public void run() {
channel.send(message);
}
}, "test-thread").start();
new Thread((Runnable) () -> channel.send(message), "test-thread").start();
latch.await(1000, TimeUnit.MILLISECONDS);
assertEquals("test-thread", target.threadName);
}

View File

@@ -24,6 +24,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
@@ -46,10 +47,8 @@ public class DispatchingChannelErrorHandlingTests {
@Test(expected = MessageDeliveryException.class)
public void handlerThrowsExceptionPublishSubscribeWithoutExecutor() {
PublishSubscribeChannel channel = new PublishSubscribeChannel();
channel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) {
throw new UnsupportedOperationException("intentional test failure");
}
channel.subscribe(message -> {
throw new UnsupportedOperationException("intentional test failure");
});
Message<?> message = MessageBuilder.withPayload("test").build();
channel.send(message);
@@ -69,11 +68,9 @@ public class DispatchingChannelErrorHandlingTests {
channel.afterPropertiesSet();
ResultHandler resultHandler = new ResultHandler();
defaultErrorChannel.subscribe(resultHandler);
channel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) {
throw new MessagingException(message,
new UnsupportedOperationException("intentional test failure"));
}
channel.subscribe(message -> {
throw new MessagingException(message,
new UnsupportedOperationException("intentional test failure"));
});
Message<?> message = MessageBuilder.withPayload("test").build();
channel.send(message);
@@ -100,11 +97,9 @@ public class DispatchingChannelErrorHandlingTests {
channel.afterPropertiesSet();
ResultHandler resultHandler = new ResultHandler();
defaultErrorChannel.subscribe(resultHandler);
channel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) {
throw new MessagingException(message,
new UnsupportedOperationException("intentional test failure"));
}
channel.subscribe(message -> {
throw new MessagingException(message,
new UnsupportedOperationException("intentional test failure"));
});
Message<?> message = MessageBuilder.withPayload("test").build();
channel.send(message);
@@ -137,6 +132,7 @@ public class DispatchingChannelErrorHandlingTests {
private volatile Thread lastThread;
@Override
public void handleMessage(Message<?> message) {
this.lastMessage = message;
this.lastThread = Thread.currentThread();

View File

@@ -264,6 +264,10 @@ public class ExecutorChannelTests {
private Message<?> messageToReturn;
BeforeHandleInterceptor() {
super();
}
public void setMessageToReturn(Message<?> messageToReturn) {
this.messageToReturn = messageToReturn;
}

View File

@@ -38,9 +38,7 @@ import org.junit.runner.RunWith;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.runners.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -130,28 +128,24 @@ public class MixedDispatcherConfigurationScenarioTests {
dispatcher.addHandler(handlerA);
dispatcher.addHandler(handlerB);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e) {
exceptionRegistry.add(e);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e2) {
exceptionRegistry.add(e2);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);
@@ -176,39 +170,27 @@ public class MixedDispatcherConfigurationScenarioTests {
dispatcher.addHandler(handlerA);
dispatcher.addHandler(handlerB);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
RuntimeException e = new RuntimeException();
allDone.countDown();
failed.set(true);
exceptionRegistry.add(e);
throw e;
}
doAnswer(invocation -> {
RuntimeException e = new RuntimeException();
allDone.countDown();
failed.set(true);
exceptionRegistry.add(e);
throw e;
}).when(handlerA).handleMessage(message);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
allDone.countDown();
return null;
}
doAnswer(invocation -> {
allDone.countDown();
return null;
}).when(handlerB).handleMessage(message);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);
@@ -272,28 +254,24 @@ public class MixedDispatcherConfigurationScenarioTests {
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e) {
exceptionRegistry.add(e);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e2) {
exceptionRegistry.add(e2);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);
@@ -323,46 +301,30 @@ public class MixedDispatcherConfigurationScenarioTests {
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
failed.set(true);
RuntimeException e = new RuntimeException();
exceptionRegistry.add(e);
allDone.countDown();
throw e;
}
doAnswer(invocation -> {
failed.set(true);
RuntimeException e = new RuntimeException();
exceptionRegistry.add(e);
allDone.countDown();
throw e;
}).when(handlerA).handleMessage(message);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
allDone.countDown();
return null;
}
doAnswer(invocation -> {
allDone.countDown();
return null;
}).when(handlerB).handleMessage(message);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
allDone.countDown();
return null;
}
doAnswer(invocation -> {
allDone.countDown();
return null;
}).when(handlerC).handleMessage(message);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);
@@ -426,28 +388,24 @@ public class MixedDispatcherConfigurationScenarioTests {
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e) {
exceptionRegistry.add(e);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
boolean sent = false;
try {
sent = channel.send(message);
}
catch (Exception e2) {
exceptionRegistry.add(e2);
}
if (!sent) {
failed.set(true);
}
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);
@@ -473,43 +431,25 @@ public class MixedDispatcherConfigurationScenarioTests {
dispatcher.addHandler(handlerB);
dispatcher.addHandler(handlerC);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
RuntimeException e = new RuntimeException();
failed.set(true);
throw e;
}
doAnswer(invocation -> {
RuntimeException e = new RuntimeException();
failed.set(true);
throw e;
}).when(handlerA).handleMessage(message);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) {
allDone.countDown();
return null;
}
doAnswer(invocation -> {
allDone.countDown();
return null;
}).when(handlerB).handleMessage(message);
doAnswer(new Answer<Object>() {
doAnswer(invocation -> null).when(handlerC).handleMessage(message);
@Override
public Object answer(InvocationOnMock invocation) {
return null;
Runnable messageSenderTask = () -> {
try {
start.await();
}
}).when(handlerC).handleMessage(message);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.send(message);
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
executor.execute(messageSenderTask);

View File

@@ -23,7 +23,6 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Executor;
@@ -31,13 +30,10 @@ import java.util.concurrent.Executor;
import org.apache.commons.logging.Log;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.FieldCallback;
/**
* @author Oleg Zhurakousky
@@ -78,21 +74,14 @@ public class P2pChannelTests {
final Log logger = mock(Log.class);
when(logger.isInfoEnabled()).thenReturn(true);
final List<String> logs = new ArrayList<String>();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
logs.add((String) invocation.getArguments()[0]);
return null;
}
doAnswer(invocation -> {
logs.add((String) invocation.getArguments()[0]);
return null;
}).when(logger).info(Mockito.anyString());
ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() {
@Override
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
ReflectionUtils.doWithFields(AbstractMessageChannel.class, field -> {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
});
String log = "Channel '"
@@ -122,15 +111,10 @@ public class P2pChannelTests {
final Log logger = mock(Log.class);
when(logger.isInfoEnabled()).thenReturn(true);
ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() {
@Override
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
ReflectionUtils.doWithFields(AbstractMessageChannel.class, field -> {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
});
channel.subscribe(mock(MessageHandler.class));
@@ -145,15 +129,10 @@ public class P2pChannelTests {
final Log logger = mock(Log.class);
when(logger.isInfoEnabled()).thenReturn(true);
ReflectionUtils.doWithFields(AbstractMessageChannel.class, new FieldCallback() {
@Override
public void doWith(Field field) throws IllegalArgumentException,
IllegalAccessException {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
ReflectionUtils.doWithFields(AbstractMessageChannel.class, field -> {
if ("logger".equals(field.getName())) {
field.setAccessible(true);
field.set(channel, logger);
}
});
channel.subscribe(mock(MessageHandler.class));

View File

@@ -91,18 +91,8 @@ public class PriorityChannelTests {
final Message<String> message = new GenericMessage<String>("hello");
for (int i = 0; i < 1000; i++) {
channel.send(message);
new Thread(new Runnable() {
@Override
public void run() {
channel.receive();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
message.getHeaders().toString();
}
}).start();
new Thread(() -> channel.receive()).start();
new Thread(() -> message.getHeaders().toString()).start();
}
}
@@ -239,14 +229,7 @@ public class PriorityChannelTests {
final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(new Runnable() {
@Override
public void run() {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 10));
}
});
executor.execute(() -> sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 10)));
assertFalse(sentSecondMessage.get());
executor.shutdown();
@@ -265,12 +248,9 @@ public class PriorityChannelTests {
final CountDownLatch latch = new CountDownLatch(1);
Executor executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(new Runnable() {
@Override
public void run() {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 3000));
latch.countDown();
}
executor.execute(() -> {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), 3000));
latch.countDown();
});
assertFalse(sentSecondMessage.get());
Thread.sleep(500);
@@ -290,12 +270,7 @@ public class PriorityChannelTests {
final AtomicBoolean sentSecondMessage = new AtomicBoolean(false);
ExecutorService executor = Executors.newSingleThreadScheduledExecutor();
channel.send(new GenericMessage<String>("test-1"));
executor.execute(new Runnable() {
@Override
public void run() {
sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), -1));
}
});
executor.execute(() -> sentSecondMessage.set(channel.send(new GenericMessage<String>("test-2"), -1)));
assertFalse(sentSecondMessage.get());
Thread.sleep(500);
Message<?> message1 = channel.receive(1000);

View File

@@ -49,14 +49,11 @@ public class QueueChannelTests {
final AtomicBoolean messageReceived = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
final QueueChannel channel = new QueueChannel();
new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
if (message != null) {
messageReceived.set(true);
latch.countDown();
}
new Thread(() -> {
Message<?> message = channel.receive();
if (message != null) {
messageReceived.set(true);
latch.countDown();
}
}).start();
assertFalse(messageReceived.get());
@@ -72,35 +69,24 @@ public class QueueChannelTests {
final CountDownLatch latch1 = new CountDownLatch(1);
final CountDownLatch latch2 = new CountDownLatch(1);
Executor singleThreadExecutor = Executors.newSingleThreadExecutor();
Runnable receiveTask1 = new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
}
latch1.countDown();
}
};
Runnable sendTask = new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("testing"));
Runnable receiveTask1 = () -> {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
}
latch1.countDown();
};
Runnable sendTask = () -> channel.send(new GenericMessage<String>("testing"));
singleThreadExecutor.execute(receiveTask1);
latch1.await();
singleThreadExecutor.execute(sendTask);
assertFalse(messageReceived.get());
Runnable receiveTask2 = new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
}
latch2.countDown();
Runnable receiveTask2 = () -> {
Message<?> message = channel.receive(0);
if (message != null) {
messageReceived.set(true);
}
latch2.countDown();
};
singleThreadExecutor.execute(receiveTask2);
latch2.await();
@@ -112,14 +98,11 @@ public class QueueChannelTests {
final QueueChannel channel = new QueueChannel();
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive();
receiveInterrupted.set(true);
assertTrue(message == null);
latch.countDown();
}
Thread t = new Thread(() -> {
Message<?> message = channel.receive();
receiveInterrupted.set(true);
assertTrue(message == null);
latch.countDown();
});
t.start();
assertFalse(receiveInterrupted.get());
@@ -133,14 +116,11 @@ public class QueueChannelTests {
final QueueChannel channel = new QueueChannel();
final AtomicBoolean receiveInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
Message<?> message = channel.receive(10000);
receiveInterrupted.set(true);
assertTrue(message == null);
latch.countDown();
}
Thread t = new Thread(() -> {
Message<?> message = channel.receive(10000);
receiveInterrupted.set(true);
assertTrue(message == null);
latch.countDown();
});
t.start();
assertFalse(receiveInterrupted.get());
@@ -169,13 +149,10 @@ public class QueueChannelTests {
assertTrue(result1);
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("test-2"));
sendInterrupted.set(true);
latch.countDown();
}
Thread t = new Thread(() -> {
channel.send(new GenericMessage<String>("test-2"));
sendInterrupted.set(true);
latch.countDown();
});
t.start();
assertFalse(sendInterrupted.get());
@@ -191,13 +168,10 @@ public class QueueChannelTests {
assertTrue(result1);
final AtomicBoolean sendInterrupted = new AtomicBoolean(false);
final CountDownLatch latch = new CountDownLatch(1);
Thread t = new Thread(new Runnable() {
@Override
public void run() {
channel.send(new GenericMessage<String>("test-2"), 10000);
sendInterrupted.set(true);
latch.countDown();
}
Thread t = new Thread(() -> {
channel.send(new GenericMessage<String>("test-2"), 10000);
sendInterrupted.set(true);
latch.countDown();
});
t.start();
assertFalse(sendInterrupted.get());

View File

@@ -59,11 +59,9 @@ public class ThreadLocalChannelParserTests {
simpleChannel.send(new GenericMessage<String>("test"));
Executor otherThreadExecutor = Executors.newSingleThreadExecutor();
final CountDownLatch latch = new CountDownLatch(1);
otherThreadExecutor.execute(new Runnable() {
public void run() {
simpleChannel.send(new GenericMessage<String>("crap"));
latch.countDown();
}
otherThreadExecutor.execute(() -> {
simpleChannel.send(new GenericMessage<String>("crap"));
latch.countDown();
});
latch.await(1, TimeUnit.SECONDS);
assertEquals("test", simpleChannel.receive(10).getPayload());
@@ -81,17 +79,13 @@ public class ThreadLocalChannelParserTests {
Executor otherThreadExecutor = Executors.newSingleThreadExecutor();
final List<Object> otherThreadResults = new ArrayList<Object>();
final CountDownLatch latch = new CountDownLatch(2);
otherThreadExecutor.execute(new Runnable() {
public void run() {
otherThreadResults.add(simpleChannel.receive(0));
latch.countDown();
}
otherThreadExecutor.execute(() -> {
otherThreadResults.add(simpleChannel.receive(0));
latch.countDown();
});
otherThreadExecutor.execute(new Runnable() {
public void run() {
otherThreadResults.add(channelWithInterceptor.receive(0));
latch.countDown();
}
otherThreadExecutor.execute(() -> {
otherThreadResults.add(channelWithInterceptor.receive(0));
latch.countDown();
});
latch.await(1, TimeUnit.SECONDS);
assertEquals(2, otherThreadResults.size());

View File

@@ -45,7 +45,6 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.ExecutorChannelInterceptor;
@@ -271,15 +270,10 @@ public class ChannelInterceptorTests {
final CountDownLatch latch2 = new CountDownLatch(2);
final List<Message<?>> messages = new ArrayList<>();
PollingConsumer consumer = new PollingConsumer(channel, new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
messages.add(message);
latch1.countDown();
latch2.countDown();
}
PollingConsumer consumer = new PollingConsumer(channel, message -> {
messages.add(message);
latch1.countDown();
latch2.countDown();
});
testApplicationContext.registerBean("consumer", consumer);
@@ -339,6 +333,10 @@ public class ChannelInterceptorTests {
private static AtomicInteger counter = new AtomicInteger();
PreSendReturnsNullInterceptor() {
super();
}
protected int getCount() {
return counter.get();
}
@@ -359,6 +357,10 @@ public class ChannelInterceptorTests {
private RuntimeException exceptionToRaise;
AfterCompletionTestInterceptor() {
super();
}
public void setExceptionToRaise(RuntimeException exception) {
this.exceptionToRaise = exception;
}
@@ -397,6 +399,10 @@ public class ChannelInterceptorTests {
private RuntimeException exceptionToRaise;
PreReceiveReturnsTrueInterceptor() {
super();
}
public void setExceptionToRaise(RuntimeException exception) {
this.exceptionToRaise = exception;
}
@@ -430,6 +436,10 @@ public class ChannelInterceptorTests {
private static AtomicInteger counter = new AtomicInteger();
PreReceiveReturnsFalseInterceptor() {
super();
}
@Override
public boolean preReceive(MessageChannel channel) {
counter.incrementAndGet();
@@ -441,6 +451,10 @@ public class ChannelInterceptorTests {
private static class TestExecutorInterceptor extends ChannelInterceptorAdapter
implements ExecutorChannelInterceptor {
TestExecutorInterceptor() {
super();
}
@Override
public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())

View File

@@ -36,8 +36,6 @@ import java.util.Map;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
@@ -226,12 +224,8 @@ public class HeaderChannelRegistryTests {
when(beanFactory.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
HeaderChannelRegistry.class))
.thenReturn(mock(HeaderChannelRegistry.class));
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
throw new NoSuchBeanDefinitionException("bar");
}
doAnswer(invocation -> {
throw new NoSuchBeanDefinitionException("bar");
}).when(beanFactory).getBean("foo", MessageChannel.class);
resolver.setBeanFactory(beanFactory);
try {
@@ -248,12 +242,8 @@ public class HeaderChannelRegistryTests {
public void testBFCRNoRegistry() {
BeanFactoryChannelResolver resolver = new BeanFactoryChannelResolver();
BeanFactory beanFactory = mock(BeanFactory.class);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
throw new NoSuchBeanDefinitionException("bar");
}
doAnswer(invocation -> {
throw new NoSuchBeanDefinitionException("bar");
}).when(beanFactory).getBean("foo", MessageChannel.class);
resolver.setBeanFactory(beanFactory);
try {

View File

@@ -44,7 +44,6 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy;
@@ -61,9 +60,6 @@ import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
@@ -142,13 +138,7 @@ public class AggregatorParserTests {
MessageChannel input = (MessageChannel) context.getBean("aggregatorWithExpressionsInput");
SubscribableChannel outputChannel = (SubscribableChannel) context.getBean("aggregatorWithExpressionsOutput");
final AtomicReference<Message<?>> aggregatedMessage = new AtomicReference<Message<?>>();
outputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
aggregatedMessage.set(message);
}
});
outputChannel.subscribe(message -> aggregatedMessage.set(message));
List<Message<?>> outboundMessages = new ArrayList<Message<?>>();
outboundMessages.add(MessageBuilder.withPayload("123").setHeader("foo", "1").build());
outboundMessages.add(MessageBuilder.withPayload("456").setHeader("foo", "1").build());

View File

@@ -37,8 +37,6 @@ import org.hamcrest.Factory;
import org.hamcrest.Matcher;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.BeansException;
import org.springframework.beans.DirectFieldAccessor;
@@ -310,12 +308,9 @@ public class ChainParserTests {
Log logger = mock(Log.class);
final AtomicReference<String> log = new AtomicReference<String>();
when(logger.isWarnEnabled()).thenReturn(true);
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
log.set((String) invocation.getArguments()[0]);
return null;
}
doAnswer(invocation -> {
log.set((String) invocation.getArguments()[0]);
return null;
}).when(logger).warn(any());
@SuppressWarnings("unchecked")

View File

@@ -33,9 +33,6 @@ import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ErrorHandler;
@@ -56,14 +53,7 @@ public class PublishSubscribeChannelParserTests {
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
dispatcher.setApplySequence(true);
dispatcher.addHandler(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
}
});
dispatcher.addHandler(message -> { });
dispatcher.dispatch(new GenericMessage<String>("foo"));
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertNull(dispatcherAccessor.getPropertyValue("executor"));

View File

@@ -222,6 +222,10 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
private volatile boolean running;
LifecycleMessageSource() {
super();
}
@Override
public void start() {
this.running = true;

View File

@@ -194,34 +194,37 @@ public class FilterAnnotationPostProcessorTests {
@MessageEndpoint
private static class TestFilterWithBooleanPrimitive {
public static class TestFilterWithBooleanPrimitive {
@Filter(inputChannel = "input", outputChannel = "output")
public boolean filter(String s) {
return !s.contains("bad");
}
}
@MessageEndpoint
private static class TestFilterWithAdviceDiscardWithin {
public static class TestFilterWithAdviceDiscardWithin {
@Filter(inputChannel = "input", outputChannel = "output", adviceChain = "adviceChain")
public boolean filter(String s) {
return !s.contains("bad");
}
}
@MessageEndpoint
private static class TestFilterWithAdviceDiscardWithinTwice {
public static class TestFilterWithAdviceDiscardWithinTwice {
@Filter(inputChannel = "input", outputChannel = "output", adviceChain = {"adviceChain1", "adviceChain2"})
public boolean filter(String s) {
return !s.contains("bad");
}
}
@MessageEndpoint
private static class TestFilterWithAdviceDiscardWithout {
public static class TestFilterWithAdviceDiscardWithout {
@Filter(inputChannel = "input", outputChannel = "output",
adviceChain = "adviceChain", discardWithinAdvice = "false")
@@ -231,31 +234,34 @@ public class FilterAnnotationPostProcessorTests {
}
@MessageEndpoint
private static class TestFilterWithBooleanWrapperClass {
public static class TestFilterWithBooleanWrapperClass {
@Filter(inputChannel = "input", outputChannel = "output")
public Boolean filter(String s) {
return !s.contains("bad");
}
}
@MessageEndpoint
private static class TestFilterWithStringReturnType {
public static class TestFilterWithStringReturnType {
@Filter(inputChannel = "input", outputChannel = "output")
public String filter(String s) {
return s;
}
}
@MessageEndpoint
private static class TestFilterWithVoidReturnType {
public static class TestFilterWithVoidReturnType {
@Filter(inputChannel = "input", outputChannel = "output")
public void filter(String s) {
}
}
public static class TestAdvice extends AbstractRequestHandlerAdvice {
@@ -266,4 +272,5 @@ public class FilterAnnotationPostProcessorTests {
}
}
}

View File

@@ -320,7 +320,7 @@ public class MessagingAnnotationPostProcessorTests {
@MessageEndpoint
private static class OutboundOnlyTestBean {
public static class OutboundOnlyTestBean {
private String messageText;
@@ -343,17 +343,17 @@ public class MessagingAnnotationPostProcessorTests {
}
private static class SimpleAnnotatedEndpointSubclass extends AnnotatedTestService {
public static class SimpleAnnotatedEndpointSubclass extends AnnotatedTestService {
}
@MessageEndpoint
private interface SimpleAnnotatedEndpointInterface {
public interface SimpleAnnotatedEndpointInterface {
String test(String input);
}
private static class SimpleAnnotatedEndpointImplementation implements SimpleAnnotatedEndpointInterface {
public static class SimpleAnnotatedEndpointImplementation implements SimpleAnnotatedEndpointInterface {
@Override
@ServiceActivator(inputChannel = "inputChannel", outputChannel = "outputChannel")
@@ -364,7 +364,7 @@ public class MessagingAnnotationPostProcessorTests {
@MessageEndpoint
private static class ServiceActivatorAnnotatedBean {
public static class ServiceActivatorAnnotatedBean {
public final AtomicBoolean invoked = new AtomicBoolean();
@@ -385,7 +385,7 @@ public class MessagingAnnotationPostProcessorTests {
@MessageEndpoint
private static class TransformerAnnotationTestBean {
public static class TransformerAnnotationTestBean {
@Transformer(inputChannel = "inputChannel", outputChannel = "outputChannel")
public String transformBefore(String input) {

View File

@@ -74,7 +74,6 @@ import org.springframework.integration.transformer.ExpressionEvaluatingTransform
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
@@ -278,13 +277,7 @@ public class MessagingAnnotationsWithBeanAnnotationTests {
@ServiceActivator(inputChannel = "serviceChannel")
public MessageHandler service() {
final List<Message<?>> collector = this.collector();
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
collector.add(message);
}
};
return message -> collector.add(message);
}
@Bean

View File

@@ -83,6 +83,7 @@ public class ConverterParserWithExistingConversionServiceTests {
TestBean1(String text) {
this.text = text;
}
}
@@ -99,6 +100,7 @@ public class ConverterParserWithExistingConversionServiceTests {
public String toString() {
return this.text.replace("-TEST", "_TARGET_CHANNEL");
}
}
private static class TestBean3 {
@@ -113,21 +115,33 @@ public class ConverterParserWithExistingConversionServiceTests {
public String toString() {
return this.text.replace("-TEST", "_TARGET_CHANNEL");
}
}
private static class TestConverter implements Converter<TestBean1, TestBean2> {
TestConverter() {
super();
}
@Override
public TestBean2 convert(TestBean1 source) {
return new TestBean2(source.text.toUpperCase());
}
}
private static class TestConverter3 implements Converter<TestBean1, TestBean3> {
TestConverter3() {
super();
}
@Override
public TestBean3 convert(TestBean1 source) {
return new TestBean3(source.text.toUpperCase());
}
}
}

View File

@@ -28,6 +28,7 @@ import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -201,13 +202,7 @@ public class DelegatingConsumerParserTests {
public static class MyFilter extends MessageFilter {
public MyFilter() {
super(new MessageSelector() {
@Override
public boolean accept(Message<?> message) {
return true;
}
});
super(message -> true);
}
}

View File

@@ -21,10 +21,8 @@ import static org.junit.Assert.fail;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -49,19 +47,14 @@ public class DispatcherMaxSubscribersOverrideDefaultTests extends DispatcherMaxS
@Test
public void testExceed() {
oneSub.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
}
});
oneSub.subscribe(message -> { });
try {
oneSub.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
}
});
oneSub.subscribe(message -> { });
fail("Expected Exception");
}
catch (IllegalArgumentException e) {
assertEquals("Maximum subscribers exceeded", e.getMessage());
}
}
}

View File

@@ -349,40 +349,37 @@ public class GatewayParserTests {
}
private void startResponder(final PollableChannel requestChannel, final MessageChannel replyChannel) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
@Override
public void run() {
Message<?> request = requestChannel.receive(60000);
assertNotNull("Request not received", request);
Message<?> reply = MessageBuilder.fromMessage(request)
.setCorrelationId(request.getHeaders().getId()).build();
Object payload = null;
if (request.getPayload().equals("futureSync")) {
payload = new AsyncResult<Message<?>>(reply);
}
else if (request.getPayload().equals("flowCompletable")) {
payload = CompletableFuture.<String>completedFuture("SYNC_COMPLETABLE");
}
else if (request.getPayload().equals("flowCustomCompletable")) {
MyCompletableFuture myCompletableFuture = new MyCompletableFuture();
myCompletableFuture.complete("SYNC_CUSTOM_COMPLETABLE");
payload = myCompletableFuture;
}
else if (request.getPayload().equals("flowCompletableM")) {
payload = CompletableFuture.<Message<?>>completedFuture(reply);
}
else if (request.getPayload().equals("flowCustomCompletableM")) {
MyCompletableMessageFuture myCompletableFuture = new MyCompletableMessageFuture();
myCompletableFuture.complete(reply);
payload = myCompletableFuture;
}
if (payload != null) {
reply = MessageBuilder.withPayload(payload)
.copyHeaders(reply.getHeaders())
.build();
}
replyChannel.send(reply);
Executors.newSingleThreadExecutor().execute(() -> {
Message<?> request = requestChannel.receive(60000);
assertNotNull("Request not received", request);
Message<?> reply = MessageBuilder.fromMessage(request)
.setCorrelationId(request.getHeaders().getId()).build();
Object payload = null;
if (request.getPayload().equals("futureSync")) {
payload = new AsyncResult<Message<?>>(reply);
}
else if (request.getPayload().equals("flowCompletable")) {
payload = CompletableFuture.<String>completedFuture("SYNC_COMPLETABLE");
}
else if (request.getPayload().equals("flowCustomCompletable")) {
MyCompletableFuture myCompletableFuture1 = new MyCompletableFuture();
myCompletableFuture1.complete("SYNC_CUSTOM_COMPLETABLE");
payload = myCompletableFuture1;
}
else if (request.getPayload().equals("flowCompletableM")) {
payload = CompletableFuture.<Message<?>>completedFuture(reply);
}
else if (request.getPayload().equals("flowCustomCompletableM")) {
MyCompletableMessageFuture myCompletableFuture2 = new MyCompletableMessageFuture();
myCompletableFuture2.complete(reply);
payload = myCompletableFuture2;
}
if (payload != null) {
reply = MessageBuilder.withPayload(payload)
.copyHeaders(reply.getHeaders())
.build();
}
replyChannel.send(reply);
});
}

View File

@@ -97,6 +97,10 @@ public class ObjectToStringTransformerParserTests {
private static class TestBean {
TestBean() {
super();
}
@Override
public String toString() {
return "test";

View File

@@ -32,11 +32,11 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.serializer.Deserializer;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.FileCopyUtils;
@@ -128,6 +128,10 @@ public class PayloadDeserializingTransformerParserTests {
@SuppressWarnings("serial")
private static class TestBean implements Serializable {
TestBean() {
super();
}
public final String name = "test";
}
@@ -135,6 +139,7 @@ public class PayloadDeserializingTransformerParserTests {
public static class TestDeserializer implements Deserializer<Object> {
@Override
public Object deserialize(InputStream source) throws IOException {
return FileCopyUtils.copyToString(new InputStreamReader(source, "UTF-8")).toUpperCase();
}

View File

@@ -28,13 +28,14 @@ import java.io.Serializable;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -125,6 +126,10 @@ public class PayloadSerializingTransformerParserTests {
@SuppressWarnings("serial")
private static class TestBean implements Serializable {
TestBean() {
super();
}
public final String name = "test";
}
@@ -132,6 +137,7 @@ public class PayloadSerializingTransformerParserTests {
public static class TestSerializer implements Serializer<Object> {
@Override
public void serialize(Object source, OutputStream outputStream) throws IOException {
outputStream.write(source.toString().toUpperCase().getBytes("UTF-8"));
outputStream.flush();

View File

@@ -25,8 +25,6 @@ import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
@@ -61,12 +59,10 @@ public class PublishingInterceptorParserTests {
public void validateDefaultChannelPublishing() {
MessageHandler handler = Mockito.mock(MessageHandler.class);
defaultChannel.subscribe(handler);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) {
Message<?> message = (Message<?>) invocation.getArguments()[0];
assertEquals("hello", message.getPayload());
return null;
}
doAnswer(invocation -> {
Message<?> message = (Message<?>) invocation.getArguments()[0];
assertEquals("hello", message.getPayload());
return null;
}).when(handler).handleMessage((Message<?>) anyObject());
testBean.echoDefaultChannel("hello");
verify(handler, times(1)).handleMessage((Message<?>) anyObject());
@@ -76,13 +72,11 @@ public class PublishingInterceptorParserTests {
public void validateEchoChannelPublishing() {
MessageHandler handler = Mockito.mock(MessageHandler.class);
echoChannel.subscribe(handler);
doAnswer(new Answer<Object>() {
public Object answer(InvocationOnMock invocation) {
Message<?> message = (Message<?>) invocation.getArguments()[0];
assertEquals("bar", message.getHeaders().get("foo"));
assertEquals("Echoing: hello", message.getPayload());
return null;
}
doAnswer(invocation -> {
Message<?> message = (Message<?>) invocation.getArguments()[0];
assertEquals("bar", message.getHeaders().get("foo"));
assertEquals("Echoing: hello", message.getPayload());
return null;
}).when(handler).handleMessage((Message<?>) anyObject());
testBean.echo("hello");
verify(handler, times(1)).handleMessage((Message<?>) anyObject());

View File

@@ -211,6 +211,10 @@ public class ServiceActivatorParserTests {
@SuppressWarnings("unused")
private static class TestBean {
TestBean() {
super();
}
public String caps(String s) {
return s.toUpperCase();
}
@@ -224,6 +228,10 @@ public class ServiceActivatorParserTests {
@SuppressWarnings("unused")
private static class TestPayload {
TestPayload() {
super();
}
public String getSimpleClassName(Object o) {
return o.getClass().getSimpleName();
}

View File

@@ -35,11 +35,9 @@ import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -48,8 +46,6 @@ import org.apache.commons.logging.Log;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.FactoryBean;
@@ -103,6 +99,7 @@ import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MutableMessageBuilder;
import org.springframework.integration.support.SmartLifecycleRoleController;
import org.springframework.integration.test.util.OnlyOnceTrigger;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -115,7 +112,6 @@ import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.support.CronTrigger;
@@ -294,15 +290,10 @@ public class EnableIntegrationTests {
Log logger = spy(TestUtils.getPropertyValue(this.serviceActivatorEndpoint, "logger", Log.class));
when(logger.isDebugEnabled()).thenReturn(true);
final CountDownLatch pollerInterruptedLatch = new CountDownLatch(1);
doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
pollerInterruptedLatch.countDown();
invocation.callRealMethod();
return null;
}
doAnswer(invocation -> {
pollerInterruptedLatch.countDown();
invocation.callRealMethod();
return null;
}).when(logger).debug("Received no Message during the poll, returning 'false'");
new DirectFieldAccessor(this.serviceActivatorEndpoint).setPropertyValue("logger", logger);
@@ -719,15 +710,7 @@ public class EnableIntegrationTests {
@Bean
public Trigger onlyOnceTrigger() {
return new Trigger() {
private final AtomicBoolean invoked = new AtomicBoolean();
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
return this.invoked.getAndSet(true) ? null : new Date();
}
};
return new OnlyOnceTrigger();
}
@Bean

View File

@@ -50,7 +50,7 @@ import org.springframework.util.Assert;
public class AsyncMessagingTemplateTests {
// TODO: changed from 0 because of recurrent failure: is this right?
private long safety = 100;
private final long safety = 100;
@Test
public void asyncSendWithDefaultChannel() throws Exception {
@@ -438,19 +438,15 @@ public class AsyncMessagingTemplateTests {
private static void sendMessageAfterDelay(final MessageChannel channel, final GenericMessage<String> message,
final int delay) {
Executors.newSingleThreadExecutor().execute(new Runnable() {
public void run() {
try {
Thread.sleep(delay);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
channel.send(message);
Executors.newSingleThreadExecutor().execute(() -> {
try {
Thread.sleep(delay);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
channel.send(message);
});
}
@@ -490,6 +486,7 @@ public class AsyncMessagingTemplateTests {
private static class TestMessagePostProcessor implements MessagePostProcessor {
@Override
public Message<?> postProcessMessage(Message<?> message) {
return MessageBuilder.fromMessage(message).setHeader("foo", "bar").build();
}

View File

@@ -133,7 +133,7 @@ public class MessageHistoryTests {
private final int id;
private TestComponent(int id) {
TestComponent(int id) {
this.id = id;
}

View File

@@ -158,12 +158,7 @@ public class MessageIdGenerationTests {
Field idGeneratorField = ReflectionUtils.findField(MessageHeaders.class, "idGenerator");
ReflectionUtils.makeAccessible(idGeneratorField);
ReflectionUtils.setField(idGeneratorField, null, new IdGenerator() {
@Override
public UUID generateId() {
return TimeBasedUUIDGenerator.generateId();
}
});
ReflectionUtils.setField(idGeneratorField, null, (IdGenerator) () -> TimeBasedUUIDGenerator.generateId());
watch = new StopWatch();
watch.start();
for (int i = 0; i < times; i++) {

View File

@@ -27,8 +27,6 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -48,15 +46,15 @@ public class BroadcastingDispatcherTests {
private BroadcastingDispatcher dispatcher;
private TaskExecutor taskExecutorMock = Mockito.mock(TaskExecutor.class);
private final TaskExecutor taskExecutorMock = Mockito.mock(TaskExecutor.class);
private Message<?> messageMock = Mockito.mock(Message.class);
private final Message<?> messageMock = Mockito.mock(Message.class);
private MessageHandler targetMock1 = Mockito.mock(MessageHandler.class);
private final MessageHandler targetMock1 = Mockito.mock(MessageHandler.class);
private MessageHandler targetMock2 = Mockito.mock(MessageHandler.class);
private final MessageHandler targetMock2 = Mockito.mock(MessageHandler.class);
private MessageHandler targetMock3 = Mockito.mock(MessageHandler.class);
private final MessageHandler targetMock3 = Mockito.mock(MessageHandler.class);
@Before
@@ -277,13 +275,9 @@ public class BroadcastingDispatcherTests {
}
private void defaultTaskExecutorMock() {
Mockito.doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
((Runnable) invocation.getArguments()[0]).run();
return null;
}
Mockito.doAnswer(invocation -> {
((Runnable) invocation.getArguments()[0]).run();
return null;
}).when(taskExecutorMock).execute(Mockito.any(Runnable.class));
}
@@ -292,15 +286,11 @@ public class BroadcastingDispatcherTests {
*/
private void partialFailingExecutorMock(final boolean... passes) {
final AtomicInteger count = new AtomicInteger();
Mockito.doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
if (passes[count.getAndIncrement()]) {
((Runnable) invocation.getArguments()[0]).run();
}
return null;
Mockito.doAnswer(invocation -> {
if (passes[count.getAndIncrement()]) {
((Runnable) invocation.getArguments()[0]).run();
}
return null;
}).when(taskExecutorMock).execute(Mockito.any(Runnable.class));
}
@@ -313,6 +303,7 @@ public class BroadcastingDispatcherTests {
this.messageList = messageList;
}
@Override
public void handleMessage(Message<?> message) {
this.messageList.add(message);
}

View File

@@ -153,32 +153,26 @@ public class OrderedAwareCopyOnWriteArraySetTests {
final Object o8 = new Foo(Ordered.HIGHEST_PRECEDENCE);
final Object o9 = new Foo(4);
final Object o10 = new Foo(2);
Thread t1 = new Thread(new Runnable() {
public void run() {
setToTest.add(o1);
setToTest.add(o3);
setToTest.add(o5);
setToTest.add(o7);
setToTest.add(o9);
}
Thread t1 = new Thread(() -> {
setToTest.add(o1);
setToTest.add(o3);
setToTest.add(o5);
setToTest.add(o7);
setToTest.add(o9);
});
Thread t2 = new Thread(new Runnable() {
public void run() {
setToTest.add(o2);
setToTest.add(o4);
setToTest.add(o6);
setToTest.add(o8);
setToTest.add(o10);
}
Thread t2 = new Thread(() -> {
setToTest.add(o2);
setToTest.add(o4);
setToTest.add(o6);
setToTest.add(o8);
setToTest.add(o10);
});
Thread t3 = new Thread(new Runnable() {
public void run() {
setToTest.add(1);
setToTest.add(new Foo(2));
setToTest.add(3);
setToTest.add(new Foo(9));
setToTest.add(8);
}
Thread t3 = new Thread(() -> {
setToTest.add(1);
setToTest.add(new Foo(2));
setToTest.add(3);
setToTest.add(new Foo(9));
setToTest.add(8);
});
t1.start();
t2.start();
@@ -230,12 +224,10 @@ public class OrderedAwareCopyOnWriteArraySetTests {
tempList.add(o9);
tempList.add(o10);
final OrderedAwareCopyOnWriteArraySet orderAwareSet = new OrderedAwareCopyOnWriteArraySet();
Thread t1 = new Thread(new Runnable() {
public void run() {
orderAwareSet.addAll(tempList);
orderAwareSet.remove(o5);
orderAwareSet.remove(o7);
}
Thread t1 = new Thread(() -> {
orderAwareSet.addAll(tempList);
orderAwareSet.remove(o5);
orderAwareSet.remove(o7);
});
final List tempList2 = new ArrayList();
final Foo foo5 = new Foo(5);
@@ -249,17 +241,13 @@ public class OrderedAwareCopyOnWriteArraySetTests {
tempList2.add(10);
tempList2.add(13);
tempList2.add(new Foo(63));
Thread t2 = new Thread(new Runnable() {
public void run() {
orderAwareSet.addAll(tempList2);
orderAwareSet.remove(foo5);
}
Thread t2 = new Thread(() -> {
orderAwareSet.addAll(tempList2);
orderAwareSet.remove(foo5);
});
Thread t3 = new Thread(new Runnable() {
public void run() {
orderAwareSet.add("hello");
orderAwareSet.add("hello again");
}
Thread t3 = new Thread(() -> {
orderAwareSet.add("hello");
orderAwareSet.add("hello again");
});
t1.start();
@@ -277,17 +265,24 @@ public class OrderedAwareCopyOnWriteArraySetTests {
Object[] elements = orderAwareSet.toArray();
assertEquals(18, elements.length);
}
private static class Foo implements Ordered {
private final int order;
Foo(int order) {
this.order = order;
}
@Override
public int getOrder() {
return order;
}
@Override
public String toString() {
return "Foo-" + order;
}
}
}

View File

@@ -32,10 +32,10 @@ import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.springframework.messaging.Message;
import org.springframework.integration.MessageRejectedException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
@@ -47,9 +47,9 @@ public class RoundRobinDispatcherConcurrentTests {
private static final int TOTAL_EXECUTIONS = 40;
private UnicastingDispatcher dispatcher = new UnicastingDispatcher();
private final UnicastingDispatcher dispatcher = new UnicastingDispatcher();
private ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor();
private final ThreadPoolTaskExecutor scheduler = new ThreadPoolTaskExecutor();
@Mock
private MessageHandler handler1;
@@ -84,19 +84,17 @@ public class RoundRobinDispatcherConcurrentTests {
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
Runnable messageSenderTask = new Runnable() {
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
if (!dispatcher.dispatch(message)) {
failed.set(true);
}
allDone.countDown();
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
if (!dispatcher.dispatch(message)) {
failed.set(true);
}
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
@@ -116,23 +114,21 @@ public class RoundRobinDispatcherConcurrentTests {
final CountDownLatch start = new CountDownLatch(1);
final CountDownLatch allDone = new CountDownLatch(TOTAL_EXECUTIONS);
final Message<?> message = this.message;
Runnable messageSenderTask = new Runnable() {
public void run() {
try {
start.await();
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
try {
dispatcher.dispatch(message);
fail("this shouldn't happen");
}
catch (MessagingException e) {
// expected
}
allDone.countDown();
Runnable messageSenderTask = () -> {
try {
start.await();
}
catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
try {
dispatcher.dispatch(message);
fail("this shouldn't happen");
}
catch (MessagingException e2) {
// expected
}
allDone.countDown();
};
for (int i = 0; i < TOTAL_EXECUTIONS; i++) {
scheduler.execute(messageSenderTask);
@@ -151,6 +147,7 @@ public class RoundRobinDispatcherConcurrentTests {
final Message<?> message = this.message;
final AtomicBoolean failed = new AtomicBoolean(false);
Runnable messageSenderTask = new Runnable() {
@Override
public void run() {
try {
start.await();

View File

@@ -27,7 +27,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -44,14 +43,10 @@ public class UnicastingDispatcherTests {
public void withInboundGatewayAsyncRequestChannelAndExplicitErrorChannel() throws Exception {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("unicasting-with-async.xml", this.getClass());
SubscribableChannel errorChannel = context.getBean("errorChannel", SubscribableChannel.class);
MessageHandler errorHandler = new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
assertTrue(message.getPayload() instanceof MessageDeliveryException);
replyChannel.send(new GenericMessage<String>("reply"));
}
MessageHandler errorHandler = message -> {
MessageChannel replyChannel = (MessageChannel) message.getHeaders().getReplyChannel();
assertTrue(message.getPayload() instanceof MessageDeliveryException);
replyChannel.send(new GenericMessage<String>("reply"));
};
errorChannel.subscribe(errorHandler);

View File

@@ -20,14 +20,15 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.splitter.MethodInvokingSplitter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Mark Fisher
@@ -132,6 +133,10 @@ public class CorrelationIdTests {
@SuppressWarnings("unused")
private static class TestBean {
TestBean() {
super();
}
public String upperCase(String input) {
return input.toUpperCase();
}

View File

@@ -31,12 +31,11 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.messaging.Message;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
@@ -69,10 +68,8 @@ public class ExpressionEvaluatingMessageSourceIntegrationTests {
adapter.setMaxMessagesPerPoll(3);
adapter.setTrigger(new PeriodicTrigger(60000));
adapter.setOutputChannel(channel);
adapter.setErrorHandler(new ErrorHandler() {
public void handleError(Throwable t) {
throw new IllegalStateException("unexpected exception in test", t);
}
adapter.setErrorHandler(t -> {
throw new IllegalStateException("unexpected exception in test", t);
});
adapter.start();
List<Message<?>> messages = new ArrayList<Message<?>>();

View File

@@ -35,8 +35,6 @@ import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
@@ -53,11 +51,8 @@ public class MessageProducerSupportTests {
public void validateExceptionIfNoErrorChannel() {
DirectChannel outChannel = new DirectChannel();
outChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("problems");
}
outChannel.subscribe(message -> {
throw new RuntimeException("problems");
});
MessageProducerSupport mps = new MessageProducerSupport() { };
mps.setOutputChannel(outChannel);
@@ -70,18 +65,12 @@ public class MessageProducerSupportTests {
@Test(expected = MessageDeliveryException.class)
public void validateExceptionIfSendToErrorChannelFails() {
DirectChannel outChannel = new DirectChannel();
outChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("problems");
}
outChannel.subscribe(message -> {
throw new RuntimeException("problems");
});
PublishSubscribeChannel errorChannel = new PublishSubscribeChannel();
errorChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("ooops");
}
errorChannel.subscribe(message -> {
throw new RuntimeException("ooops");
});
MessageProducerSupport mps = new MessageProducerSupport() { };
mps.setOutputChannel(outChannel);
@@ -95,11 +84,8 @@ public class MessageProducerSupportTests {
@Test
public void validateSuccessfulErrorFlowDoesNotThrowErrors() {
DirectChannel outChannel = new DirectChannel();
outChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("problems");
}
outChannel.subscribe(message -> {
throw new RuntimeException("problems");
});
PublishSubscribeChannel errorChannel = new PublishSubscribeChannel();
SuccessfulErrorService errorService = new SuccessfulErrorService();
@@ -143,11 +129,8 @@ public class MessageProducerSupportTests {
assertTrue(endpoint.isStopped());
endpoint.start();
assertFalse(endpoint.isStopped());
endpoint.stop(new Runnable() {
@Override
public void run() {
// Do nothing
}
endpoint.stop(() -> {
// Do nothing
});
assertEquals(1, endpoint.getCount());
assertTrue(endpoint.isStopped());
@@ -157,6 +140,10 @@ public class MessageProducerSupportTests {
private volatile Message<?> lastMessage;
SuccessfulErrorService() {
super();
}
@SuppressWarnings("unused")
public void handleErrorMessage(Message<?> errorMessage) {
this.lastMessage = errorMessage;
@@ -166,8 +153,13 @@ public class MessageProducerSupportTests {
private static class CustomEndpoint extends AbstractEndpoint {
private final AtomicInteger count = new AtomicInteger(0);
private final AtomicBoolean stopped = new AtomicBoolean(true);
CustomEndpoint() {
super();
}
public int getCount() {
return this.count.get();
}

View File

@@ -31,8 +31,8 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.messaging.Message;
import org.springframework.integration.MessageRejectedException;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
@@ -50,21 +50,21 @@ public class PollingConsumerEndpointTests {
private PollingConsumer endpoint;
private TestTrigger trigger = new TestTrigger();
private final TestTrigger trigger = new TestTrigger();
private TestConsumer consumer = new TestConsumer();
private final TestConsumer consumer = new TestConsumer();
@SuppressWarnings("rawtypes")
private Message message = new GenericMessage<String>("test");
private final Message message = new GenericMessage<String>("test");
@SuppressWarnings("rawtypes")
private Message badMessage = new GenericMessage<String>("bad");
private final Message badMessage = new GenericMessage<String>("bad");
private TestErrorHandler errorHandler = new TestErrorHandler();
private final TestErrorHandler errorHandler = new TestErrorHandler();
private PollableChannel channelMock = Mockito.mock(PollableChannel.class);
private final PollableChannel channelMock = Mockito.mock(PollableChannel.class);
private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
@Before
@@ -176,6 +176,11 @@ public class PollingConsumerEndpointTests {
private volatile AtomicInteger counter = new AtomicInteger();
TestConsumer() {
super();
}
@Override
public void handleMessage(Message<?> message) {
this.counter.incrementAndGet();
if ("bad".equals(message.getPayload().toString())) {
@@ -192,6 +197,11 @@ public class PollingConsumerEndpointTests {
private volatile CountDownLatch latch = new CountDownLatch(1);
TestTrigger() {
super();
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (!this.hasRun.getAndSet(true)) {
return new Date();
@@ -223,6 +233,11 @@ public class PollingConsumerEndpointTests {
private volatile Throwable lastError;
TestErrorHandler() {
super();
}
@Override
public void handleError(Throwable t) {
this.lastError = t;
}

View File

@@ -55,9 +55,9 @@ import org.springframework.scheduling.support.PeriodicTrigger;
*/
public class PollingLifecycleTests {
private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
private TestErrorHandler errorHandler = new TestErrorHandler();
private final TestErrorHandler errorHandler = new TestErrorHandler();
@Before
public void init() throws Exception {
@@ -71,6 +71,7 @@ public class PollingLifecycleTests {
channel.send(new GenericMessage<String>("foo"));
MessageHandler handler = Mockito.spy(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
latch.countDown();
}
@@ -105,6 +106,7 @@ public class PollingLifecycleTests {
pollerMetadata.setTrigger(new PeriodicTrigger(2000));
adapterFactory.setPollerMetadata(pollerMetadata);
MessageSource<String> source = spy(new MessageSource<String>() {
@Override
public Message<String> receive() {
latch.countDown();
return new GenericMessage<String>("hello");
@@ -135,21 +137,19 @@ public class PollingLifecycleTests {
pollerMetadata.setTrigger(new PeriodicTrigger(2000));
adapterFactory.setPollerMetadata(pollerMetadata);
final Runnable coughtInterrupted = mock(Runnable.class);
MessageSource<String> source = new MessageSource<String>() {
public Message<String> receive() {
MessageSource<String> source = () -> {
try {
for (int i = 0; i < 10; i++) {
Thread.sleep(1000);
latch.countDown();
}
try {
for (int i = 0; i < 10; i++) {
Thread.sleep(1000);
latch.countDown();
}
catch (InterruptedException e) {
coughtInterrupted.run();
}
return new GenericMessage<String>("hello");
}
catch (InterruptedException e) {
coughtInterrupted.run();
}
return new GenericMessage<String>("hello");
};
adapterFactory.setSource(source);
adapterFactory.setOutputChannel(channel);

View File

@@ -52,8 +52,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
@@ -217,42 +215,38 @@ public class PseudoTransactionalMessageSourceTests {
public void testCommitWithManager() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
transactionTemplate.execute(new TransactionCallback<Object>() {
transactionTemplate.execute(status -> {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setBeforeCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
syncProcessor.setBeforeCommitChannel(queueChannel);
syncProcessor.setAfterCommitChannel(queueChannel);
syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#baz"));
@Override
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setBeforeCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
syncProcessor.setBeforeCommitChannel(queueChannel);
syncProcessor.setAfterCommitChannel(queueChannel);
syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
IntegrationResourceHolder holder =
(IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this);
holder.addAttribute("baz", "qux");
holder.addAttribute("bix", "qox");
return message;
}
});
@Override
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
IntegrationResourceHolder holder =
(IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this);
holder.addAttribute("baz", "qux");
holder.addAttribute("bix", "qox");
return message;
}
});
doPoll(adapter);
return null;
}
doPoll(adapter);
return null;
});
Message<?> beforeCommitMessage = queueChannel.receive(1000);
assertNotNull(beforeCommitMessage);
@@ -267,57 +261,7 @@ public class PseudoTransactionalMessageSourceTests {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
try {
transactionTemplate.execute(new TransactionCallback<Object>() {
@Override
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
return message;
}
});
doPoll(adapter);
throw new RuntimeException("Force rollback");
}
});
}
catch (Exception e) {
assertEquals("Force rollback", e.getMessage());
}
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
}
@Test
public void testRollbackWithManagerUsingStatus() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
transactionTemplate.execute(new TransactionCallback<Object>() {
@Override
public Object doInTransaction(TransactionStatus status) {
transactionTemplate.execute(status -> {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
@@ -345,9 +289,51 @@ public class PseudoTransactionalMessageSourceTests {
});
doPoll(adapter);
status.setRollbackOnly();
return null;
}
throw new RuntimeException("Force rollback");
});
}
catch (Exception e) {
assertEquals("Force rollback", e.getMessage());
}
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
}
@Test
public void testRollbackWithManagerUsingStatus() {
final PollableChannel queueChannel = new QueueChannel();
TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager());
transactionTemplate.execute(status -> {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
return message;
}
});
doPoll(adapter);
status.setRollbackOnly();
return null;
});
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
@@ -358,13 +344,7 @@ public class PseudoTransactionalMessageSourceTests {
public void testInt2777UnboundResourceAfterTransactionComplete() {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
return null;
}
});
adapter.setSource(() -> null);
TransactionSynchronizationManager.setActualTransactionActive(true);
doPoll(adapter);
@@ -396,13 +376,7 @@ public class PseudoTransactionalMessageSourceTests {
};
adapter.setTransactionSynchronizationFactory(syncFactory);
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
return null;
}
});
adapter.setSource(() -> null);
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);

Some files were not shown because too many files have changed in this diff Show More