Skip to content

Commit

Permalink
Tar aggregation strategy null bodies (#15732)
Browse files Browse the repository at this point in the history 
* Handle null bodies in TarAggregationStrategy

* Reduce cognitive complexity

* Remove deprecated doPreSetup/doPostSetup methods and some cosmetic edits

---------

Co-authored-by: Thomas Gantenbein <[email protected]>
  • Loading branch information
@thomas-gantenbein-tga
thomas-gantenbein-tga and Thomas Gantenbein authored Sep 27, 2024
1 parent bd0fb08 commit 1a7924a
Show file tree
Hide file tree
Showing 12 changed files with 238 additions and 92 deletions.
62 changes: 35 additions & 27 deletions ...le/src/main/java/org/apache/camel/processor/aggregate/tarfile/TarAggregationStrategy.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -151,13 +151,13 @@ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
File tarFile;
Exchange answer = oldExchange;

boolean isFirstTimeInAggregation = oldExchange == null;
// Guard against empty new exchanges
if (newExchange == null) {
if (newExchange.getIn().getBody() == null && !isFirstTimeInAggregation) {
return oldExchange;
}

// First time for this aggregation
if (oldExchange == null) {
if (isFirstTimeInAggregation) {
try {
tarFile = FileUtil.createTempFile(this.filePrefix, this.fileSuffix, this.parentDir);
LOG.trace("Created temporary file: {}", tarFile);
Expand All @@ -171,25 +171,23 @@ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
}

Object body = newExchange.getIn().getBody();
if (body instanceof WrappedFile) {
body = ((WrappedFile) body).getFile();
if (body instanceof WrappedFile wrappedFile) {
body = wrappedFile.getFile();
}

if (body instanceof File) {
try {
File appendFile = (File) body;
// do not try to append empty files
if (appendFile.length() > 0) {
String entryName = preserveFolderStructure
? newExchange.getIn().getHeader(Exchange.FILE_NAME, String.class)
: newExchange.getIn().getMessageId();
addFileToTar(tarFile, appendFile, this.preserveFolderStructure ? entryName : null);
}
} catch (Exception e) {
throw new GenericFileOperationFailedException(e.getMessage(), e);
}
if (body instanceof File appendFile) {
addFileToTar(newExchange, appendFile, tarFile);
} else {
// Handle all other messages
appendIncomingBodyAsBytesToTar(newExchange, tarFile);
}
GenericFile<File> genericFile = FileConsumer.asGenericFile(
tarFile.getParent(), tarFile, Charset.defaultCharset().toString(), false);
genericFile.bindToExchange(answer);
return answer;
}

private void appendIncomingBodyAsBytesToTar(Exchange newExchange, File tarFile) {
if (newExchange.getIn().getBody() != null) {
try {
byte[] buffer = newExchange.getIn().getMandatoryBody(byte[].class);
// do not try to append empty data
Expand All @@ -203,10 +201,20 @@ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
throw new GenericFileOperationFailedException(e.getMessage(), e);
}
}
GenericFile<File> genericFile = FileConsumer.asGenericFile(
tarFile.getParent(), tarFile, Charset.defaultCharset().toString(), false);
genericFile.bindToExchange(answer);
return answer;
}

private void addFileToTar(Exchange newExchange, File appendFile, File tarFile) {
try {
// do not try to append empty files
if (appendFile.length() > 0) {
String entryName = preserveFolderStructure
? newExchange.getIn().getHeader(Exchange.FILE_NAME, String.class)
: newExchange.getIn().getMessageId();
addFileToTar(tarFile, appendFile, this.preserveFolderStructure ? entryName : null);
}
} catch (Exception e) {
throw new GenericFileOperationFailedException(e.getMessage(), e);
}
}

@Override
Expand All @@ -219,13 +227,13 @@ public void onCompletion(Exchange exchange, Exchange inputExchange) {

private void addFileToTar(File source, File file, String fileName) throws IOException, ArchiveException {
File tmpTar = Files.createTempFile(parentDir.toPath(), source.getName(), null).toFile();
tmpTar.delete();
Files.delete(tmpTar.toPath());
if (!source.renameTo(tmpTar)) {
throw new IOException("Could not make temp file (" + source.getName() + ")");
}

try (FileInputStream fis = new FileInputStream(tmpTar)) {
try (TarArchiveInputStream tin = (TarArchiveInputStream) new ArchiveStreamFactory()
try (TarArchiveInputStream tin = new ArchiveStreamFactory()
.createArchiveInputStream(ArchiveStreamFactory.TAR, fis)) {
try (TarArchiveOutputStream tos = new TarArchiveOutputStream(new FileOutputStream(source))) {
tos.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX);
Expand Down Expand Up @@ -264,13 +272,13 @@ private void copyExistingEntries(TarArchiveInputStream tin, TarArchiveOutputStre

private void addEntryToTar(File source, String entryName, byte[] buffer, int length) throws IOException, ArchiveException {
File tmpTar = Files.createTempFile(parentDir.toPath(), source.getName(), null).toFile();
tmpTar.delete();
Files.delete(tmpTar.toPath());
if (!source.renameTo(tmpTar)) {
throw new IOException("Cannot create temp file: " + source.getName());
}

try (FileInputStream fis = new FileInputStream(tmpTar)) {
try (TarArchiveInputStream tin = (TarArchiveInputStream) new ArchiveStreamFactory()
try (TarArchiveInputStream tin = new ArchiveStreamFactory()
.createArchiveInputStream(ArchiveStreamFactory.TAR, fis)) {
try (TarArchiveOutputStream tos = new TarArchiveOutputStream(new FileOutputStream(source))) {
tos.setLongFileMode(TarArchiveOutputStream.LONGFILE_POSIX);
Expand Down
23 changes: 12 additions & 11 deletions ...arfile/src/test/java/org/apache/camel/dataformat/tarfile/SpringTarFileDataFormatTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.camel.builder.NotifyBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.camel.test.spring.junit5.CamelSpringTestSupport;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;

Expand All @@ -40,11 +41,11 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class SpringTarFileDataFormatTest extends CamelSpringTestSupport {
class SpringTarFileDataFormatTest extends CamelSpringTestSupport {
private static final File TEST_DIR = new File("target/springtar");

@Test
public void testTarWithoutFileName() throws Exception {
void testTarWithoutFileName() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);

Expand All @@ -65,7 +66,7 @@ public void testTarWithoutFileName() throws Exception {
}

@Test
public void testTarWithFileName() throws Exception {
void testTarWithFileName() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);
mock.expectedHeaderReceived(FILE_NAME, "poem.txt.tar");
Expand All @@ -87,7 +88,7 @@ public void testTarWithFileName() throws Exception {
}

@Test
public void testUntar() throws Exception {
void testUntar() throws Exception {
getMockEndpoint("mock:untar").expectedBodiesReceived(TEXT);
getMockEndpoint("mock:untar").expectedHeaderReceived(FILE_NAME, "file");

Expand All @@ -97,7 +98,7 @@ public void testUntar() throws Exception {
}

@Test
public void testTarAndUntar() throws Exception {
void testTarAndUntar() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tarAndUntar");
mock.expectedMessageCount(1);

Expand All @@ -111,7 +112,7 @@ public void testTarAndUntar() throws Exception {
}

@Test
public void testTarToFileWithoutFileName() throws Exception {
void testTarToFileWithoutFileName() throws Exception {
NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

String[] files = TEST_DIR.list();
Expand Down Expand Up @@ -142,7 +143,7 @@ public void testTarToFileWithoutFileName() throws Exception {
}

@Test
public void testTarToFileWithFileName() throws Exception {
void testTarToFileWithFileName() throws Exception {
NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

MockEndpoint mock = getMockEndpoint("mock:tarToFile");
Expand Down Expand Up @@ -172,7 +173,7 @@ public void testTarToFileWithFileName() throws Exception {
}

@Test
public void testDslTar() throws Exception {
void testDslTar() throws Exception {
getMockEndpoint("mock:dslTar").expectedHeaderReceived(FILE_NAME, "poem.txt.tar");

template.sendBodyAndHeader("direct:dslTar", TEXT, FILE_NAME, "poem.txt");
Expand All @@ -181,7 +182,7 @@ public void testDslTar() throws Exception {
}

@Test
public void testDslUntar() throws Exception {
void testDslUntar() throws Exception {
getMockEndpoint("mock:dslUntar").expectedBodiesReceived(TEXT);
getMockEndpoint("mock:dslUntar").expectedHeaderReceived(FILE_NAME, "test.txt");

Expand All @@ -190,8 +191,8 @@ public void testDslUntar() throws Exception {
MockEndpoint.assertIsSatisfied(context);
}

@Override
public void doPostSetup() {
@AfterEach
public void cleanOutputDirectory() {
deleteDirectory(TEST_DIR);
}

Expand Down
4 changes: 2 additions & 2 deletions ...tarfile/src/test/java/org/apache/camel/dataformat/tarfile/SpringTarSplitterRouteTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,10 @@
import org.junit.jupiter.api.Test;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class SpringTarSplitterRouteTest extends CamelSpringTestSupport {
class SpringTarSplitterRouteTest extends CamelSpringTestSupport {

@Test
public void testSplitter() throws InterruptedException {
void testSplitter() throws InterruptedException {
MockEndpoint processTarEntry = getMockEndpoint("mock:processTarEntry");

processTarEntry.expectedBodiesReceivedInAnyOrder("chau", "hi", "hola", "hello", "greetings");
Expand Down
37 changes: 19 additions & 18 deletions ...amel-tarfile/src/test/java/org/apache/camel/dataformat/tarfile/TarFileDataFormatTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import org.apache.commons.compress.archivers.ArchiveStreamFactory;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;

import static org.apache.camel.Exchange.FILE_NAME;
Expand All @@ -56,13 +57,13 @@
/**
* Unit tests for {@link TarFileDataFormat}.
*/
public class TarFileDataFormatTest extends CamelTestSupport {
class TarFileDataFormatTest extends CamelTestSupport {

private static final File TEST_DIR = new File("target/tar");
private TarFileDataFormat tar;

@Test
public void testTarWithoutFileName() throws Exception {
void testTarWithoutFileName() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);

Expand All @@ -84,7 +85,7 @@ public void testTarWithoutFileName() throws Exception {
}

@Test
public void testTarWithFileName() throws Exception {
void testTarWithFileName() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);
mock.expectedHeaderReceived(FILE_NAME, "poem.txt.tar");
Expand All @@ -106,7 +107,7 @@ public void testTarWithFileName() throws Exception {
}

@Test
public void testTarWithPathElements() throws Exception {
void testTarWithPathElements() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);
mock.expectedHeaderReceived(FILE_NAME, "poem.txt.tar");
Expand All @@ -127,7 +128,7 @@ public void testTarWithPathElements() throws Exception {
}

@Test
public void testTarWithPreservedPathElements() throws Exception {
void testTarWithPreservedPathElements() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tar");
mock.expectedMessageCount(1);
mock.expectedHeaderReceived(FILE_NAME, "poem.txt.tar");
Expand Down Expand Up @@ -155,7 +156,7 @@ public void testTarWithPreservedPathElements() throws Exception {
}

@Test
public void testUntar() throws Exception {
void testUntar() throws Exception {
getMockEndpoint("mock:untar").expectedBodiesReceived(TEXT);
getMockEndpoint("mock:untar").expectedHeaderReceived(FILE_NAME, "file");

Expand All @@ -165,15 +166,15 @@ public void testUntar() throws Exception {
}

@Test
public void testUntarWithCorruptedTarFile() {
void testUntarWithCorruptedTarFile() {
final File body = new File("src/test/resources/data/corrupt.tar");

assertThrows(CamelExecutionException.class,
() -> template.sendBody("direct:corruptUntar", body));
}

@Test
public void testTarAndUntar() throws Exception {
void testTarAndUntar() throws Exception {
MockEndpoint mock = getMockEndpoint("mock:tarAndUntar");
mock.expectedMessageCount(1);

Expand All @@ -187,7 +188,7 @@ public void testTarAndUntar() throws Exception {
}

@Test
public void testTarToFileWithoutFileName() throws Exception {
void testTarToFileWithoutFileName() throws Exception {
NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

String[] files = TEST_DIR.list();
Expand Down Expand Up @@ -218,7 +219,7 @@ public void testTarToFileWithoutFileName() throws Exception {
}

@Test
public void testTarToFileWithFileName() throws Exception {
void testTarToFileWithFileName() throws Exception {
NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

MockEndpoint mock = getMockEndpoint("mock:tarToFile");
Expand Down Expand Up @@ -249,7 +250,7 @@ public void testTarToFileWithFileName() throws Exception {
}

@Test
public void testDslTar() throws Exception {
void testDslTar() throws Exception {
getMockEndpoint("mock:dslTar").expectedHeaderReceived(FILE_NAME, "poem.txt.tar");

template.sendBodyAndHeader("direct:dslTar", TEXT, FILE_NAME, "poem.txt");
Expand All @@ -258,7 +259,7 @@ public void testDslTar() throws Exception {
}

@Test
public void testDslUntar() throws Exception {
void testDslUntar() throws Exception {
getMockEndpoint("mock:dslUntar").expectedBodiesReceived(TEXT);
getMockEndpoint("mock:dslUntar").expectedHeaderReceived(FILE_NAME, "test.txt");

Expand All @@ -268,7 +269,7 @@ public void testDslUntar() throws Exception {
}

@Test
public void testUntarWithEmptyDirectorySupported() {
void testUntarWithEmptyDirectorySupported() {
deleteDirectory(new File("hello_out"));
tar.setUsingIterator(true);
tar.setAllowEmptyDirectory(true);
Expand All @@ -278,7 +279,7 @@ public void testUntarWithEmptyDirectorySupported() {
}

@Test
public void testUntarWithEmptyDirectoryUnsupported() {
void testUntarWithEmptyDirectoryUnsupported() {
deleteDirectory(new File("hello_out"));
tar.setUsingIterator(true);
tar.setAllowEmptyDirectory(false);
Expand All @@ -288,16 +289,16 @@ public void testUntarWithEmptyDirectoryUnsupported() {
}

@Test
public void testUnzipMaxDecompressedSize() throws Exception {
void testUnzipMaxDecompressedSize() throws Exception {
final byte[] files = getTaredText("file");

// We are only allowing 10 bytes to be decompressed, so we expect an error
assertThrows(CamelExecutionException.class,
() -> template.sendBody("direct:untarMaxDecompressedSize", files));
}

@Override
public void doPostSetup() {
@AfterEach
public void cleanOutputDirectory() {
deleteDirectory(TEST_DIR);
}

Expand Down Expand Up @@ -352,7 +353,7 @@ public void process(Exchange exchange) throws Exception {
} else {
outputFile.getParentFile().mkdirs();
try (TarArchiveInputStream debInputStream
= (TarArchiveInputStream) new ArchiveStreamFactory().createArchiveInputStream("tar",
= new ArchiveStreamFactory().createArchiveInputStream("tar",
is)) {
copy(debInputStream, outputFile);
}
Expand Down
Loading

0 comments on commit 1a7924a

Please sign in to comment.