From 8dab7409b591df5730516e3409769d76d22778d2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 11 Feb 2025 11:47:33 -0500 Subject: [PATCH] GH-9792: Remove resource header in the `StreamTransformer` Fixes: https://github.com/spring-projects/spring-integration/issues/9792 The `StreamTransformer` closes `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header value, so this becomes unusable afterward. In addition, it may even cause some problems downstream when the message could be serialized for subsequent network interaction. * Add logic to the `StreamTransformer` to build a new message, but remove `IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE` header * Verify that header is removed in the `StreamingInboundTests`. Addition `StreamingInboundTests` clean up for better code style. --- .../transformer/StreamTransformer.java | 25 +++-- .../file/remote/StreamingInboundTests.java | 98 +++++++++++-------- 2 files changed, 73 insertions(+), 50 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java index 222cade96a..3af5b41cda 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/StreamTransformer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2019 the original author or authors. + * Copyright 2016-2025 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,16 +22,18 @@ import java.io.IOException; import java.io.InputStream; import java.io.UncheckedIOException; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.StaticMessageHeaderAccessor; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.FileCopyUtils; /** - * Transforms an InputStream payload to a byte[] or String (if a - * charset is provided). + * Transforms an {@link InputStream} payload to a {@code byte[]} or {@link String} if a charset is provided. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.3 * */ @@ -40,8 +42,7 @@ public class StreamTransformer extends AbstractTransformer { private final String charset; /** - * Construct an instance to transform an {@link InputStream} to - * a {@code byte[]}. + * Construct an instance to transform an {@link InputStream} to a {@code byte[]}. */ public StreamTransformer() { this(null); @@ -49,7 +50,7 @@ public class StreamTransformer extends AbstractTransformer { /** * Construct an instance with the charset to convert the stream to a - * String; if null a {@code byte[]} will be produced instead. + * String; if {@code null} a {@code byte[]} will be produced instead. * @param charset the charset. */ public StreamTransformer(String charset) { @@ -63,14 +64,20 @@ public class StreamTransformer extends AbstractTransformer { InputStream stream = (InputStream) message.getPayload(); ByteArrayOutputStream baos = new ByteArrayOutputStream(); FileCopyUtils.copy(stream, baos); + Object result = this.charset == null ? baos.toByteArray() : baos.toString(this.charset); Closeable closeableResource = StaticMessageHeaderAccessor.getCloseableResource(message); if (closeableResource != null) { closeableResource.close(); + result = getMessageBuilderFactory() + .withPayload(result) + .copyHeaders(message.getHeaders()) + .removeHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE) + .build(); } - return this.charset == null ? baos.toByteArray() : baos.toString(this.charset); + return result; } - catch (IOException e) { - throw new UncheckedIOException(e); + catch (IOException ex) { + throw new UncheckedIOException(ex); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java index ca6795fde3..a17c7241de 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/StreamingInboundTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2023 the original author or authors. + * Copyright 2016-2025 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. @@ -97,37 +97,45 @@ public class StreamingInboundTests { } streamer.afterPropertiesSet(); streamer.start(); - Message received = (Message) this.transformer.transform(streamer.receive()); + Message inputStreamMessage = streamer.receive(); + Message received = (Message) this.transformer.transform(inputStreamMessage); assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "foo") + .doesNotContainKey(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE); String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); - assertThat(fileInfo).contains("remoteDirectory\":\"/foo"); - assertThat(fileInfo).contains("permissions\":\"-rw-rw-rw"); - assertThat(fileInfo).contains("size\":42"); - assertThat(fileInfo).contains("directory\":false"); - assertThat(fileInfo).contains("filename\":\"foo"); - assertThat(fileInfo).contains("modified\":42000"); - assertThat(fileInfo).contains("link\":false"); + assertThat(fileInfo) + .contains("remoteDirectory\":\"/foo") + .contains("permissions\":\"-rw-rw-rw") + .contains("size\":42") + .contains("directory\":false") + .contains("filename\":\"foo") + .contains("modified\":42000") + .contains("link\":false"); // close after list, transform - verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(inputStreamMessage), times(2)).close(); - received = (Message) this.transformer.transform(streamer.receive()); + inputStreamMessage = streamer.receive(); + received = (Message) this.transformer.transform(inputStreamMessage); assertThat(received.getPayload()).isEqualTo("baz\nqux".getBytes()); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "bar") + .doesNotContainKey(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE); fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO); - assertThat(fileInfo).contains("remoteDirectory\":\"/foo"); - assertThat(fileInfo).contains("permissions\":\"-rw-rw-rw"); - assertThat(fileInfo).contains("size\":42"); - assertThat(fileInfo).contains("directory\":false"); - assertThat(fileInfo).contains("filename\":\"bar"); - assertThat(fileInfo).contains("modified\":42000"); - assertThat(fileInfo).contains("link\":false"); + assertThat(fileInfo) + .contains("remoteDirectory\":\"/foo") + .contains("permissions\":\"-rw-rw-rw") + .contains("size\":42") + .contains("directory\":false") + .contains("filename\":\"bar") + .contains("modified\":42000") + .contains("link\":false"); // close after transform - verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(3)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(inputStreamMessage), times(3)).close(); verify(sessionFactory.getSession()).list("/foo"); } @@ -142,21 +150,25 @@ public class StreamingInboundTests { streamer.setFilter(new AcceptOnceFileListFilter<>()); streamer.afterPropertiesSet(); streamer.start(); - Message received = (Message) this.transformer.transform(streamer.receive()); + Message inputStreamMessage = streamer.receive(); + Message received = (Message) this.transformer.transform(inputStreamMessage); assertThat(received.getPayload()).isEqualTo("foo\nbar".getBytes()); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "foo"); // close after list, transform - verify(StaticMessageHeaderAccessor.getCloseableResource(received), times(2)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(inputStreamMessage), times(2)).close(); - received = (Message) this.transformer.transform(streamer.receive()); + inputStreamMessage = streamer.receive(); + received = (Message) this.transformer.transform(inputStreamMessage); assertThat(received.getPayload()).isEqualTo("baz\nqux".getBytes()); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "bar"); // close after transform - verify(new IntegrationMessageHeaderAccessor(received).getCloseableResource(), times(3)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(inputStreamMessage), times(3)).close(); verify(sessionFactory.getSession()).list("/foo"); } @@ -189,31 +201,35 @@ public class StreamingInboundTests { splitter.handleMessage(receivedStream); Message received = out.receive(0); assertThat(received.getPayload()).isEqualTo("foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "foo"); received = out.receive(0); assertThat(received.getPayload()).isEqualTo("bar"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("foo"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "foo"); assertThat(out.receive(0)).isNull(); // close by list, splitter - verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(3)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(receivedStream), times(3)).close(); receivedStream = streamer.receive(); splitter.handleMessage(receivedStream); received = out.receive(0); assertThat(received.getPayload()).isEqualTo("baz"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "bar"); received = out.receive(0); assertThat(received.getPayload()).isEqualTo("qux"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)).isEqualTo("/foo"); - assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE)).isEqualTo("bar"); + assertThat(received.getHeaders()) + .containsEntry(FileHeaders.REMOTE_DIRECTORY, "/foo") + .containsEntry(FileHeaders.REMOTE_FILE, "bar"); assertThat(out.receive(0)).isNull(); // close by splitter - verify(new IntegrationMessageHeaderAccessor(receivedStream).getCloseableResource(), times(5)).close(); + verify(StaticMessageHeaderAccessor.getCloseableResource(receivedStream), times(5)).close(); } @SuppressWarnings("unchecked")