diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index 171c53c4f1..9d75517e5b 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2024 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. @@ -106,7 +106,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin private Expression delayExpression; - private ExpressionEvaluatingMessageProcessor delayGenerator; + private ExpressionEvaluatingMessageProcessor delayGenerator; private boolean headersMappedLast; @@ -483,7 +483,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin private void configureDelayGenerator(BeanFactory beanFactory) { if (this.delayExpression != null) { - this.delayGenerator = new ExpressionEvaluatingMessageProcessor<>(this.delayExpression, Integer.class); + this.delayGenerator = new ExpressionEvaluatingMessageProcessor<>(this.delayExpression, Long.class); if (beanFactory != null) { this.delayGenerator.setBeanFactory(beanFactory); } @@ -622,7 +622,7 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin protected void addDelayProperty(Message message, org.springframework.amqp.core.Message amqpMessage) { if (this.delayGenerator != null) { - amqpMessage.getMessageProperties().setDelay(this.delayGenerator.processMessage(message)); + amqpMessage.getMessageProperties().setDelayLong(this.delayGenerator.processMessage(message)); } } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index d1c8d4945f..f4e935286e 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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. @@ -133,7 +133,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper 0, IntegrationMessageHeaderAccessor.PRIORITY, priority, headers::put) - .acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(), headers::put) + .acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelayLong(), + headers::put) .acceptIfNotNull(AmqpHeaders.RECEIVED_EXCHANGE, amqpMessageProperties.getReceivedExchange(), headers::put) .acceptIfHasText(AmqpHeaders.RECEIVED_ROUTING_KEY, amqpMessageProperties.getReceivedRoutingKey(), @@ -196,8 +197,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper("foo")); ArgumentCaptor captor = ArgumentCaptor.forClass(Message.class); verify(amqpTemplate).send(eq("foo"), eq("bar"), captor.capture(), isNull()); - assertThat(captor.getValue().getMessageProperties().getDelay()).isEqualTo(42); + assertThat(captor.getValue().getMessageProperties().getDelayLong()).isEqualTo(42); endpoint.setExpectReply(true); endpoint.setOutputChannel(new NullChannel()); endpoint.handleMessage(new GenericMessage<>("foo")); verify(amqpTemplate).sendAndReceive(eq("foo"), eq("bar"), captor.capture(), isNull()); - assertThat(captor.getValue().getMessageProperties().getDelay()).isEqualTo(42); + assertThat(captor.getValue().getMessageProperties().getDelayLong()).isEqualTo(42); endpoint.setDelay(23); endpoint.setRoutingKey("baz"); endpoint.afterPropertiesSet(); endpoint.handleMessage(new GenericMessage<>("foo")); verify(amqpTemplate).sendAndReceive(eq("foo"), eq("baz"), captor.capture(), isNull()); - assertThat(captor.getValue().getMessageProperties().getDelay()).isEqualTo(23); + assertThat(captor.getValue().getMessageProperties().getDelayLong()).isEqualTo(23); } @Test @@ -114,7 +114,7 @@ public class OutboundEndpointTests { ArgumentCaptor captor = ArgumentCaptor.forClass(Message.class); gateway.handleMessage(new GenericMessage<>("foo")); verify(amqpTemplate).sendAndReceive(eq("foo"), eq("bar"), captor.capture()); - assertThat(captor.getValue().getMessageProperties().getDelay()).isEqualTo(42); + assertThat(captor.getValue().getMessageProperties().getDelayLong()).isEqualTo(42); } @Test diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java index 63b4d95a88..78e178f5e9 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2024 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. @@ -77,7 +77,7 @@ public class DefaultAmqpHeaderMapperTests { headerMap.put(AmqpHeaders.CONTENT_TYPE, "test.contentType"); String testCorrelationId = "foo"; headerMap.put(AmqpHeaders.CORRELATION_ID, testCorrelationId); - headerMap.put(AmqpHeaders.DELAY, 1234); + headerMap.put(AmqpHeaders.DELAY, 1234L); headerMap.put(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT); headerMap.put(AmqpHeaders.DELIVERY_TAG, 1234L); headerMap.put(AmqpHeaders.EXPIRATION, "test.expiration"); @@ -111,7 +111,7 @@ public class DefaultAmqpHeaderMapperTests { assertThat(amqpProperties.getContentLength()).isEqualTo(99L); assertThat(amqpProperties.getContentType()).isEqualTo("test.contentType"); assertThat(amqpProperties.getCorrelationId()).isEqualTo(testCorrelationId); - assertThat(amqpProperties.getDelay()).isEqualTo(Integer.valueOf(1234)); + assertThat(amqpProperties.getDelayLong()).isEqualTo(1234L); assertThat(amqpProperties.getDeliveryMode()).isEqualTo(MessageDeliveryMode.NON_PERSISTENT); assertThat(amqpProperties.getDeliveryTag()).isEqualTo(1234L); assertThat(amqpProperties.getExpiration()).isEqualTo("test.expiration"); @@ -183,7 +183,7 @@ public class DefaultAmqpHeaderMapperTests { amqpProperties.setMessageCount(42); amqpProperties.setMessageId("test.messageId"); amqpProperties.setPriority(22); - amqpProperties.setReceivedDelay(4567); + amqpProperties.setReceivedDelayLong(4567L); amqpProperties.setReceivedExchange("test.receivedExchange"); amqpProperties.setReceivedRoutingKey("test.receivedRoutingKey"); amqpProperties.setRedelivered(true); @@ -206,7 +206,7 @@ public class DefaultAmqpHeaderMapperTests { assertThat(headerMap.get(AmqpHeaders.EXPIRATION)).isEqualTo("test.expiration"); assertThat(headerMap.get(AmqpHeaders.MESSAGE_COUNT)).isEqualTo(42); assertThat(headerMap.get(AmqpHeaders.MESSAGE_ID)).isEqualTo("test.messageId"); - assertThat(headerMap.get(AmqpHeaders.RECEIVED_DELAY)).isEqualTo(4567); + assertThat(headerMap.get(AmqpHeaders.RECEIVED_DELAY)).isEqualTo(4567L); assertThat(headerMap.get(AmqpHeaders.RECEIVED_EXCHANGE)).isEqualTo("test.receivedExchange"); assertThat(headerMap.get(AmqpHeaders.RECEIVED_ROUTING_KEY)).isEqualTo("test.receivedRoutingKey"); assertThat(headerMap.get(AmqpHeaders.REPLY_TO)).isEqualTo("test.replyTo");