GH-3174: Fix expiration property conversion for Stream module

Fixes: #3174

The Stream message comes with an `absoluteExpiryTime` according to the AMQP 1.0 message format.
The AMQP 0.9.1 deals with a period instead of absolute time.

* Fix `DefaultStreamMessageConverter` to calculate `expiration` from the difference between
`absoluteExpiryTime` and `creationTime` (which is also absolute).
* Use `System.currentTimeMillis()` if `creationTime` is not provided.
* Fix an `absoluteExpiryTime` calculation based on the `creationTime` and `expiration`

(cherry picked from commit a788393)

Author: artembilan

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java

@@ -28,6 +28,7 @@
 import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder;
 import com.rabbitmq.stream.Properties;
 import com.rabbitmq.stream.codec.WrapperMessageBuilder;
+import org.jspecify.annotations.Nullable;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
@@ -36,12 +37,15 @@
 import org.springframework.lang.Nullable;
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 /**
  * Default {@link StreamMessageConverter}.
  *
  * @author Gary Russell
  * @author Ngoc Nhan
+ * @author Artem Bilan
+ *
  * @since 2.4
  *
  */
@@ -92,30 +96,37 @@ public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageCo
 		Assert.isInstanceOf(StreamMessageProperties.class, props);
 		StreamMessageProperties mProps = (StreamMessageProperties) props;
 		JavaUtils.INSTANCE
-				.acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId) // TODO different types
+				.acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId)
 				.acceptIfNotNull(mProps.getUserId(), usr -> propsBuilder.userId(usr.getBytes(this.charset)))
 				.acceptIfNotNull(mProps.getTo(), propsBuilder::to)
 				.acceptIfNotNull(mProps.getSubject(), propsBuilder::subject)
 				.acceptIfNotNull(mProps.getReplyTo(), propsBuilder::replyTo)
-				.acceptIfNotNull(mProps.getCorrelationId(), propsBuilder::correlationId) // TODO different types
+				.acceptIfNotNull(mProps.getCorrelationId(), propsBuilder::correlationId)
 				.acceptIfNotNull(mProps.getContentType(), propsBuilder::contentType)
 				.acceptIfNotNull(mProps.getContentEncoding(), propsBuilder::contentEncoding)
-				.acceptIfNotNull(mProps.getExpiration(), exp -> propsBuilder.absoluteExpiryTime(Long.parseLong(exp)))
 				.acceptIfNotNull(mProps.getCreationTime(), propsBuilder::creationTime)
 				.acceptIfNotNull(mProps.getGroupId(), propsBuilder::groupId)
 				.acceptIfNotNull(mProps.getGroupSequence(), propsBuilder::groupSequence)
 				.acceptIfNotNull(mProps.getReplyToGroupId(), propsBuilder::replyToGroupId);
 		ApplicationPropertiesBuilder appPropsBuilder = builder.applicationProperties();
-		if (!mProps.getHeaders().isEmpty()) {
-			mProps.getHeaders().forEach((key, val) -> {
-				mapProp(key, val, appPropsBuilder);
-			});
+
+		long creationTime = mProps.getCreationTime();
+		if (creationTime <= 0) {
+			creationTime = System.currentTimeMillis();
+		}
+		propsBuilder.creationTime(creationTime);
+
+		String expiration = mProps.getExpiration();
+		if (StringUtils.hasText(expiration)) {
+			propsBuilder.absoluteExpiryTime(creationTime + Long.parseLong(expiration));
 		}
+
+		mProps.getHeaders().forEach((key, val) -> mapProp(key, val, appPropsBuilder));
 		builder.addData(message.getBody());
 		return builder.build();
 	}
 
-	private void mapProp(String key, Object val, ApplicationPropertiesBuilder builder) { // NOSONAR - complexity
+	private void mapProp(String key, @Nullable Object val, ApplicationPropertiesBuilder builder) {
 		if (val instanceof String string) {
 			builder.entry(key, string);
 		}
@@ -155,19 +166,28 @@ private void toMessageProperties(com.rabbitmq.stream.Message streamMessage,
 		if (properties != null) {
 			JavaUtils.INSTANCE
 					.acceptIfNotNull(properties.getMessageIdAsString(), mProps::setMessageId)
-					.acceptIfNotNull(properties.getUserId(), usr -> mProps.setUserId(new String(usr, this.charset)))
+					.acceptIfNotNull(properties.getUserId(),
+							usr -> mProps.setUserId(new String(usr, this.charset)))
 					.acceptIfNotNull(properties.getTo(), mProps::setTo)
 					.acceptIfNotNull(properties.getSubject(), mProps::setSubject)
 					.acceptIfNotNull(properties.getReplyTo(), mProps::setReplyTo)
 					.acceptIfNotNull(properties.getCorrelationIdAsString(), mProps::setCorrelationId)
 					.acceptIfNotNull(properties.getContentType(), mProps::setContentType)
 					.acceptIfNotNull(properties.getContentEncoding(), mProps::setContentEncoding)
-					.acceptIfNotNull(properties.getAbsoluteExpiryTime(),
-							exp -> mProps.setExpiration(Long.toString(exp)))
-					.acceptIfNotNull(properties.getCreationTime(), mProps::setCreationTime)
 					.acceptIfNotNull(properties.getGroupId(), mProps::setGroupId)
 					.acceptIfNotNull(properties.getGroupSequence(), mProps::setGroupSequence)
 					.acceptIfNotNull(properties.getReplyToGroupId(), mProps::setReplyToGroupId);
+
+			long creationTime = properties.getCreationTime();
+			if (creationTime <= 0) {
+				creationTime = System.currentTimeMillis();
+			}
+			mProps.setCreationTime(creationTime);
+
+			long absoluteExpiryTime = properties.getAbsoluteExpiryTime();
+			if (absoluteExpiryTime > creationTime) {
+				mProps.setExpiration(Long.toString(absoluteExpiryTime - creationTime));
+			}
 		}
 		Map<String, Object> applicationProperties = streamMessage.getApplicationProperties();
 		if (applicationProperties != null) {

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverterTests.java

@@ -24,10 +24,13 @@
 import org.springframework.rabbit.stream.support.StreamMessageProperties;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.withinPercentage;
 import static org.mockito.Mockito.mock;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.4
  *
  */
@@ -45,7 +48,7 @@ void toAndFrom() {
 		smp.setContentType("application/json");
 		smp.setContentEncoding("UTF-8");
 		smp.setExpiration("42");
-		smp.setCreationTime(43L);
+		smp.setCreationTime(System.currentTimeMillis());
 		smp.setGroupId("groupId");
 		smp.setGroupSequence(44L);
 		smp.setReplyToGroupId("replyGroupId");
@@ -62,8 +65,8 @@ void toAndFrom() {
 		assertThat(props.getCorrelationIdAsString()).isEqualTo("correlation");
 		assertThat(props.getContentType()).isEqualTo("application/json");
 		assertThat(props.getContentEncoding()).isEqualTo("UTF-8");
-		assertThat(props.getAbsoluteExpiryTime()).isEqualTo(42L);
-		assertThat(props.getCreationTime()).isEqualTo(43L);
+		assertThat(props.getAbsoluteExpiryTime()).isCloseTo(System.currentTimeMillis() + 42, withinPercentage(5));
+		assertThat(props.getCreationTime()).isCloseTo(System.currentTimeMillis(), withinPercentage(5));
 		assertThat(props.getGroupId()).isEqualTo("groupId");
 		assertThat(props.getGroupSequence()).isEqualTo(44L);
 		assertThat(props.getReplyToGroupId()).isEqualTo("replyGroupId");