diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index 63287f267b..e2ae27b564 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -292,7 +292,8 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem this.targetEvaluationContext = targetContext; if (beanFactory != null) { - configureHeaderExpressions(beanFactory); + configureHeaderExpressions(beanFactory, this.headerExpressions); + configureHeaderExpressions(beanFactory, this.nullResultHeaderExpressions); } } @@ -316,29 +317,18 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem } } - private void configureHeaderExpressions(BeanFactory beanFactory) { + private void configureHeaderExpressions(BeanFactory beanFactory, + Map> headerExpressions) { + boolean checkReadOnlyHeaders = getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory; - for (Map.Entry> entry : this.headerExpressions.entrySet()) { + for (Map.Entry> entry : headerExpressions.entrySet()) { if (checkReadOnlyHeaders && (MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) { - throw new BeanInitializationException( - "ContentEnricher cannot override 'id' and 'timestamp' read-only headers.\n" + - "Wrong 'headerExpressions' [" + this.headerExpressions - + "] configuration for " + getComponentName()); - } - if (entry.getValue() instanceof BeanFactoryAware) { - ((BeanFactoryAware) entry.getValue()).setBeanFactory(beanFactory); - } - } - for (Map.Entry> entry : - this.nullResultHeaderExpressions.entrySet()) { - if (checkReadOnlyHeaders && - (MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) { throw new BeanInitializationException( "ContentEnricher cannot override 'id' and 'timestamp' read-only headers.\n" + - "Wrong 'nullResultHeaderExpressions' [" + this.nullResultHeaderExpressions + "Wrong 'headerExpressions' [" + headerExpressions + "] configuration for " + getComponentName()); } if (entry.getValue() instanceof BeanFactoryAware) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java index 7d1831adfe..8125bd97dc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ConcurrentAggregatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,13 +17,14 @@ package org.springframework.integration.aggregator; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.junit.Before; -import org.junit.Ignore; -import org.junit.Test; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; @@ -53,7 +54,7 @@ public class ConcurrentAggregatorTests { private final MessageGroupStore store = new SimpleMessageStore(); - @Before + @BeforeEach public void configureAggregator() { this.taskExecutor = new SimpleAsyncTaskExecutor(); this.aggregator = new AggregatingMessageHandler(new MultiplyingProcessor(), this.store); @@ -68,12 +69,9 @@ public class ConcurrentAggregatorTests { Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, null); CountDownLatch latch = new CountDownLatch(3); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message1, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message2, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message3, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -84,16 +82,13 @@ public class ConcurrentAggregatorTests { } @Test - @Ignore + @Disabled // dropped backwards compatibility for duplicate ID's public void testCompleteGroupWithinTimeoutWithSameId() { QueueChannel replyChannel = new QueueChannel(); - Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, - "ID#1"); - Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, - "ID#1"); - Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, - "ID#1"); + Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, "ID#1"); + Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, "ID#1"); + Message message3 = createMessage(7, "ABC", 3, 3, replyChannel, "ID#1"); CountDownLatch latch = new CountDownLatch(3); // for testing the duplication scenario, the messages must be processed // synchronously @@ -107,20 +102,17 @@ public class ConcurrentAggregatorTests { @Test public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException { - QueueChannel discardChannel = new QueueChannel(); this.aggregator.setDiscardChannel(discardChannel); QueueChannel replyChannel = new QueueChannel(); Message message = createMessage(3, "ABC", 2, 1, replyChannel, null); CountDownLatch latch = new CountDownLatch(1); - AggregatorTestTask task = new AggregatorTestTask(this.aggregator, - message, latch); + AggregatorTestTask task = new AggregatorTestTask(this.aggregator, message, latch); this.taskExecutor.execute(task); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(latch - .getCount()).as("Task should have completed within timeout").isEqualTo(0); + assertThat(latch.getCount()).as("Task should have completed within timeout").isEqualTo(0); Message reply = replyChannel.receive(10); assertThat(reply).as("No message should have been sent normally").isNull(); this.store.expireMessageGroups(-10000); @@ -136,10 +128,8 @@ public class ConcurrentAggregatorTests { Message message1 = createMessage(3, "ABC", 3, 1, replyChannel, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel, null); CountDownLatch latch = new CountDownLatch(2); - AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, - message1, latch); - AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, - message2, latch); + AggregatorTestTask task1 = new AggregatorTestTask(this.aggregator, message1, latch); + AggregatorTestTask task2 = new AggregatorTestTask(this.aggregator, message2, latch); this.taskExecutor.execute(task1); this.taskExecutor.execute(task2); @@ -161,25 +151,16 @@ public class ConcurrentAggregatorTests { Message message1 = createMessage(3, "ABC", 3, 1, replyChannel1, null); Message message2 = createMessage(5, "ABC", 3, 2, replyChannel1, null); Message message3 = createMessage(7, "ABC", 3, 3, replyChannel1, null); - Message message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, - null); - Message message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, - null); - Message message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, - null); + Message message4 = createMessage(11, "XYZ", 3, 1, replyChannel2, null); + Message message5 = createMessage(13, "XYZ", 3, 2, replyChannel2, null); + Message message6 = createMessage(17, "XYZ", 3, 3, replyChannel2, null); CountDownLatch latch = new CountDownLatch(6); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message1, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message6, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message2, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message5, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message3, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message4, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message6, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message5, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -194,7 +175,7 @@ public class ConcurrentAggregatorTests { } @Test - @Ignore + @Disabled // dropped backwards compatibility for setting capacity limit (it's always // Integer.MAX_VALUE) public void testTrackedCorrelationIdsCapacityAtLimit() { @@ -202,23 +183,19 @@ public class ConcurrentAggregatorTests { QueueChannel discardChannel = new QueueChannel(); // this.aggregator.setTrackedCorrelationIdCapacity(3); this.aggregator.setDiscardChannel(discardChannel); - this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(1); - this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(3, 2, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(3); - this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(4, 3, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(4); // next message with same correlation ID is discarded - this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(2, 1, 1, 1, replyChannel, null)); assertThat(discardChannel.receive(1000).getPayload()).isEqualTo(2); } @Test - @Ignore + @Disabled // dropped backwards compatibility for setting capacity limit (it's always // Integer.MAX_VALUE) public void testTrackedCorrelationIdsCapacityPassesLimit() { @@ -226,29 +203,24 @@ public class ConcurrentAggregatorTests { QueueChannel discardChannel = new QueueChannel(); // this.aggregator.setTrackedCorrelationIdCapacity(3); this.aggregator.setDiscardChannel(discardChannel); - this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(1, 1, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(1); - this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(2, 2, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(2); - this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(3, 3, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(3); - this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(4, 4, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(4); - this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, - null)); + this.aggregator.handleMessage(createMessage(5, 1, 1, 1, replyChannel, null)); assertThat(replyChannel.receive(1000).getPayload()).isEqualTo(5); assertThat(discardChannel.receive(0)).isNull(); } - @Test(expected = MessageHandlingException.class) + @Test public void testExceptionThrownIfNoCorrelationId() { - Message message = createMessage(3, null, 2, 1, new QueueChannel(), - null); - this.aggregator.handleMessage(message); + Message message = createMessage(3, null, 2, 1, new QueueChannel(), null); + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> this.aggregator.handleMessage(message)); } @Test @@ -262,16 +234,12 @@ public class ConcurrentAggregatorTests { this.aggregator.setReleaseStrategy(new SequenceSizeReleaseStrategy()); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message1, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message2, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message3, latch)); - this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, - message4, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message1, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message2, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message3, latch)); + this.taskExecutor.execute(new AggregatorTestTask(this.aggregator, message4, latch)); - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(latch.await(20, TimeUnit.SECONDS)).isTrue(); Message reply = replyChannel.receive(10000); assertThat(reply).as("A message should be aggregated").isNotNull(); @@ -283,10 +251,11 @@ public class ConcurrentAggregatorTests { Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel replyChannel, String predefinedId) { - MessageBuilder builder = MessageBuilder.withPayload(payload) - .setCorrelationId(correlationId).setSequenceSize(sequenceSize) - .setSequenceNumber(sequenceNumber) - .setReplyChannel(replyChannel); + MessageBuilder builder = + MessageBuilder.withPayload(payload) + .setCorrelationId(correlationId).setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(replyChannel); if (predefinedId != null) { builder.setHeader(MessageHeaders.ID, predefinedId); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 2858ce8d20..3c7fce832b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -24,8 +24,8 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; -import org.junit.Before; -import org.junit.Test; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.spel.standard.SpelExpressionParser; @@ -56,7 +56,7 @@ public class ResequencerTests { private final MessageGroupStore store = new SimpleMessageStore(); - @Before + @BeforeEach public void configureResequencer() { this.resequencer = new ResequencingMessageHandler(processor, store, null, null); this.resequencer.setBeanFactory(mock(BeanFactory.class)); @@ -64,7 +64,7 @@ public class ResequencerTests { } @Test - public void testBasicResequencing() throws InterruptedException { + public void testBasicResequencing() { QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); @@ -84,7 +84,7 @@ public class ResequencerTests { } @Test - public void testBasicResequencingA() throws InterruptedException { + public void testBasicResequencingA() { SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); releaseStrategy.setReleasePartialSequences(true); this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy); @@ -103,7 +103,7 @@ public class ResequencerTests { } @Test - public void testBasicUnboundedResequencing() throws InterruptedException { + public void testBasicUnboundedResequencing() { SequenceSizeReleaseStrategy releaseStrategy = new SequenceSizeReleaseStrategy(); releaseStrategy.setReleasePartialSequences(true); this.resequencer = new ResequencingMessageHandler(processor, store, null, releaseStrategy); @@ -158,7 +158,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { + public void testResequencingWithIncompleteSequenceRelease() { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); // INT-3846 this.resequencer.setMessageStore(new SimpleMessageStore(3)); @@ -190,7 +190,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithCapacity() throws InterruptedException { + public void testResequencingWithCapacity() { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); // INT-3846 this.resequencer.setMessageStore(new SimpleMessageStore(3, 2)); @@ -210,7 +210,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithPartialSequenceAndComparator() throws InterruptedException { + public void testResequencingWithPartialSequenceAndComparator() { this.resequencer.setReleaseStrategy(new SequenceSizeReleaseStrategy(true)); QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("456", "ABC", 4, 2, replyChannel); @@ -240,7 +240,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithDiscard() throws InterruptedException { + public void testResequencingWithDiscard() { QueueChannel discardChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 4, 2, null); Message message2 = createMessage("456", "ABC", 4, 1, null); @@ -269,7 +269,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithDifferentSequenceSizes() throws InterruptedException { + public void testResequencingWithDifferentSequenceSizes() { QueueChannel discardChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 4, 2, null); Message message2 = createMessage("456", "ABC", 5, 1, null); @@ -288,7 +288,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithWrongSequenceSizeAndNumber() throws InterruptedException { + public void testResequencingWithWrongSequenceSizeAndNumber() { QueueChannel discardChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 2, 4, null); this.resequencer.setSendPartialResultOnExpiry(false); @@ -301,7 +301,7 @@ public class ResequencerTests { } @Test - public void testResequencingWithCompleteSequenceRelease() throws InterruptedException { + public void testResequencingWithCompleteSequenceRelease() { QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); @@ -343,7 +343,7 @@ public class ResequencerTests { } @Test - public void testTimeoutDefaultExpiry() throws InterruptedException { + public void testTimeoutDefaultExpiry() { this.resequencer.setGroupTimeoutExpression(new SpelExpressionParser().parseExpression("100")); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); @@ -359,7 +359,7 @@ public class ResequencerTests { this.resequencer.handleMessage(message2); Message out1 = replyChannel.receive(10); assertThat(out1).isNull(); - out1 = discardChannel.receive(10000); + out1 = discardChannel.receive(20000); assertThat(out1).isNotNull(); Message out2 = discardChannel.receive(10); assertThat(out2).isNotNull(); @@ -370,7 +370,7 @@ public class ResequencerTests { } @Test - public void testTimeoutDontExpire() throws InterruptedException { + public void testTimeoutDontExpire() { this.resequencer.setGroupTimeoutExpression(new SpelExpressionParser().parseExpression("100")); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); @@ -387,7 +387,7 @@ public class ResequencerTests { this.resequencer.handleMessage(message2); Message out1 = replyChannel.receive(0); assertThat(out1).isNull(); - out1 = discardChannel.receive(10_000); + out1 = discardChannel.receive(20_000); assertThat(out1).isNotNull(); Message out2 = discardChannel.receive(10_000); assertThat(out2).isNotNull(); @@ -401,8 +401,13 @@ public class ResequencerTests { private static Message createMessage(String payload, Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { - return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize) - .setSequenceNumber(sequenceNumber).setReplyChannel(replyChannel).build(); + + return MessageBuilder.withPayload(payload) + .setCorrelationId(correlationId) + .setSequenceSize(sequenceSize) + .setSequenceNumber(sequenceNumber) + .setReplyChannel(replyChannel) + .build(); } }