compression service to use with InputStreams

Adds compression service with methods for gzip, bzip2 and lzma2. The
service is designed to be used with InputStreams. For the compression
methods this means that the incoming InputStream needs to be read,
converted to an OuputStream, used in the compression implementations and
than converted back to an InputStream.
Adds v2 plugin integration tests and a direct service test.

Author: hhund

dsf-bpe/dsf-bpe-process-api-v2-impl/pom.xml

@@ -48,6 +48,14 @@
 			<groupId>org.apache.commons</groupId>
 			<artifactId>commons-lang3</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.apache.commons</groupId>
+			<artifactId>commons-compress</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.tukaani</groupId>
+			<artifactId>xz</artifactId>
+		</dependency>
 		<dependency>
 			<groupId>org.apache.tika</groupId>
 			<artifactId>tika-core</artifactId>
@@ -196,10 +204,14 @@
 									<artifactId>jakarta-regexp</artifactId>
 									<version>1.4</version>
 								</artifactItem>
-								<!--<artifactItem>
+								<artifactItem>
 									<groupId>org.apache.commons</groupId>
 									<artifactId>commons-compress</artifactId>
-								</artifactItem>-->
+								</artifactItem>
+								<artifactItem>
+									<groupId>org.tukaani</groupId>
+									<artifactId>xz</artifactId>
+								</artifactItem>
 								<artifactItem>
 									<groupId>org.fhir</groupId>
 									<artifactId>ucum</artifactId>

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/ProcessPluginApiImpl.java

@@ -8,6 +8,7 @@
 
 import ca.uhn.fhir.context.FhirContext;
 import dev.dsf.bpe.v2.config.ProxyConfig;
+import dev.dsf.bpe.v2.service.CompressionService;
 import dev.dsf.bpe.v2.service.CryptoService;
 import dev.dsf.bpe.v2.service.DataLogger;
 import dev.dsf.bpe.v2.service.DsfClientProvider;
@@ -39,6 +40,7 @@ public class ProcessPluginApiImpl implements ProcessPluginApi, InitializingBean
 	private final QuestionnaireResponseHelper questionnaireResponseHelper;
 	private final ReadAccessHelper readAccessHelper;
 	private final TaskHelper taskHelper;
+	private final CompressionService compressionService;
 	private final CryptoService cryptoService;
 	private final TargetProvider targetProvider;
 	private final DataLogger dataLogger;
@@ -49,7 +51,8 @@ public ProcessPluginApiImpl(ProxyConfig proxyConfig, EndpointProvider endpointPr
 			ObjectMapper objectMapper, OrganizationProvider organizationProvider,
 			ProcessAuthorizationHelper processAuthorizationHelper,
 			QuestionnaireResponseHelper questionnaireResponseHelper, ReadAccessHelper readAccessHelper,
-			TaskHelper taskHelper, CryptoService cryptoService, TargetProvider targetProvider, DataLogger dataLogger)
+			TaskHelper taskHelper, CompressionService compressionService, CryptoService cryptoService,
+			TargetProvider targetProvider, DataLogger dataLogger)
 	{
 		this.proxyConfig = proxyConfig;
 		this.endpointProvider = endpointProvider;
@@ -65,6 +68,7 @@ public ProcessPluginApiImpl(ProxyConfig proxyConfig, EndpointProvider endpointPr
 		this.questionnaireResponseHelper = questionnaireResponseHelper;
 		this.readAccessHelper = readAccessHelper;
 		this.taskHelper = taskHelper;
+		this.compressionService = compressionService;
 		this.cryptoService = cryptoService;
 		this.targetProvider = targetProvider;
 		this.dataLogger = dataLogger;
@@ -87,6 +91,7 @@ public void afterPropertiesSet() throws Exception
 		Objects.requireNonNull(questionnaireResponseHelper, "questionnaireResponseHelper");
 		Objects.requireNonNull(readAccessHelper, "readAccessHelper");
 		Objects.requireNonNull(taskHelper, "taskHelper");
+		Objects.requireNonNull(compressionService, "compressionService");
 		Objects.requireNonNull(cryptoService, "cryptoService");
 		Objects.requireNonNull(targetProvider, "targetProvider");
 		Objects.requireNonNull(dataLogger, "dataLogger");
@@ -176,6 +181,12 @@ public TaskHelper getTaskHelper()
 		return taskHelper;
 	}
 
+	@Override
+	public CompressionService getCompressionService()
+	{
+		return compressionService;
+	}
+
 	@Override
 	public CryptoService getCryptoService()
 	{

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/service/CompressionServiceImpl.java

@@ -0,0 +1,126 @@
+package dev.dsf.bpe.v2.service;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
+
+import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
+import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
+import org.tukaani.xz.LZMA2Options;
+import org.tukaani.xz.XZInputStream;
+import org.tukaani.xz.XZOutputStream;
+
+public class CompressionServiceImpl implements CompressionService
+{
+	private static final int BUFFER_SIZE = 8192;
+
+	@FunctionalInterface
+	private static interface Compressor
+	{
+		OutputStream withCompression(OutputStream o) throws IOException;
+	}
+
+	@Override
+	public InputStream toGzip(InputStream in) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+
+		return to(in, GZIPOutputStream::new);
+	}
+
+	@Override
+	public InputStream toBzip2(InputStream in) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+
+		return toBzip2(in, 9);
+	}
+
+	@Override
+	public InputStream toBzip2(InputStream in, int blockSize) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+		if (blockSize < 1 || blockSize > 9)
+			throw new IllegalArgumentException("blockSize < 1 or > 9");
+
+		return to(in, out -> new BZip2CompressorOutputStream(out, blockSize));
+	}
+
+	@Override
+	public InputStream toLzma2(InputStream in) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+
+		return toLzma2(in, 6);
+	}
+
+	@Override
+	public InputStream toLzma2(InputStream in, int preset) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+		if (preset < 0 || preset > 9)
+			throw new IllegalArgumentException("preset < 0 or > 9");
+
+		return to(in, out -> new XZOutputStream(out, new LZMA2Options(preset)));
+	}
+
+	private InputStream to(InputStream in, Compressor compressor) throws IOException
+	{
+		PipedOutputStream pipedOut = new PipedOutputStream();
+		PipedInputStream pipedIn = new PipedInputStream(pipedOut, BUFFER_SIZE);
+
+		List<IOException> capturedExceptions = new ArrayList<>();
+		Thread worker = new Thread(() ->
+		{
+			try (in; OutputStream out = compressor.withCompression(pipedOut))
+			{
+				byte[] buffer = new byte[BUFFER_SIZE];
+				int bytesRead;
+				while ((bytesRead = in.read(buffer)) != -1)
+					out.write(buffer, 0, bytesRead);
+			}
+			catch (IOException e)
+			{
+				capturedExceptions.add(e);
+			}
+		});
+		worker.setDaemon(true);
+		worker.start();
+
+		if (!capturedExceptions.isEmpty())
+		{
+			IOException e = capturedExceptions.removeFirst();
+			capturedExceptions.stream().forEach(e::addSuppressed);
+			throw e;
+		}
+
+		return pipedIn;
+	}
+
+	@Override
+	public InputStream fromGzip(InputStream in) throws IOException
+	{
+		return new GZIPInputStream(in);
+	}
+
+	@Override
+	public InputStream fromBzip2(InputStream in) throws IOException
+	{
+		Objects.requireNonNull(in, "in");
+
+		return new BZip2CompressorInputStream(in);
+	}
+
+	@Override
+	public InputStream fromLzma2(InputStream in) throws IOException
+	{
+		return new XZInputStream(in);
+	}
+}

dsf-bpe/dsf-bpe-process-api-v2-impl/src/main/java/dev/dsf/bpe/v2/spring/ApiServiceConfig.java

@@ -37,6 +37,8 @@
 import dev.dsf.bpe.v2.listener.ListenerVariables;
 import dev.dsf.bpe.v2.listener.StartListener;
 import dev.dsf.bpe.v2.plugin.ProcessPluginFactoryImpl;
+import dev.dsf.bpe.v2.service.CompressionService;
+import dev.dsf.bpe.v2.service.CompressionServiceImpl;
 import dev.dsf.bpe.v2.service.CryptoService;
 import dev.dsf.bpe.v2.service.CryptoServiceImpl;
 import dev.dsf.bpe.v2.service.DataLogger;
@@ -103,7 +105,7 @@ public ProcessPluginApi processPluginApiV2()
 		return new ProcessPluginApiImpl(proxyConfigDelegate(), endpointProvider(), fhirContext(), dsfClientProvider(),
 				fhirClientProvider(), oidcClientProvider(), mailService(), mimeTypeService(), objectMapper(),
 				organizationProvider(), processAuthorizationHelper(), questionnaireResponseHelper(), readAccessHelper(),
-				taskHelper(), cryptoService(), targetProvider(), dataLogger());
+				taskHelper(), compressionService(), cryptoService(), targetProvider(), dataLogger());
 	}
 
 	@Bean
@@ -284,6 +286,12 @@ public ListenerFactory listenerFactory()
 				continueListener());
 	}
 
+	@Bean
+	public CompressionService compressionService()
+	{
+		return new CompressionServiceImpl();
+	}
+
 	@Bean
 	public CryptoService cryptoService()
 	{

dsf-bpe/dsf-bpe-process-api-v2-impl/src/test/java/dev/dsf/bpe/v2/service/CompressionServiceTest.java

@@ -0,0 +1,81 @@
+package dev.dsf.bpe.v2.service;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+
+import org.junit.Test;
+
+public class CompressionServiceTest
+{
+	@FunctionalInterface
+	private static interface Converter
+	{
+		InputStream convert(InputStream i) throws IOException;
+	}
+
+	private static final byte[] TEST_DATA_1 = "Hello compression World!".getBytes(StandardCharsets.UTF_8);
+	private static final byte[] TEST_DATA_2 = new byte[1_000_000];
+	private static final byte[] TEST_DATA_3 = """
+			Lorem ipsum dolor sit amet, consectetur adipiscing elit. Nulla a mauris ligula. Maecenas finibus magna mi, at placerat neque sollicitudin et. Nunc iaculis eros sagittis, rutrum lectus quis, eleifend nunc. Interdum et malesuada fames ac ante ipsum primis in faucibus. Aenean quam risus, blandit in mauris eu, imperdiet tristique neque. Curabitur varius nisl risus, mattis dapibus est vulputate quis. Mauris cursus vel justo id consectetur. Vivamus mollis orci sit amet erat sagittis sodales.
+			Maecenas interdum erat et ipsum fermentum lobortis. Orci varius natoque penatibus et magnis dis parturient montes, nascetur ridiculus mus. Proin et ipsum velit. Donec efficitur commodo ipsum, in condimentum dui facilisis non. Donec urna ipsum, tempus non finibus quis, condimentum at mi. Aliquam at porta ante. Integer luctus nec tellus in ornare. Quisque vel elit posuere, eleifend turpis eget, molestie nulla. Sed volutpat porttitor aliquet. Quisque eu mauris eget nisl porta cursus non sed nisl.
+			Mauris mi nisi, pellentesque malesuada magna interdum, pretium feugiat erat. Mauris et velit sed erat efficitur pulvinar a in dui. Mauris id sem purus. In eget efficitur elit. Mauris a eros sodales, interdum libero at, rutrum sapien. Ut vitae nibh at augue mollis hendrerit a ut odio. Phasellus quis diam id est interdum facilisis. Nullam vitae turpis nisl. Pellentesque eu venenatis diam, ac mattis nisl. Maecenas libero dui, volutpat id eros ut, pulvinar fringilla ligula. Nam maximus orci a lectus suscipit ornare. Praesent vehicula, eros ut scelerisque faucibus, sem sapien pellentesque enim, nec suscipit risus lorem eget odio. Vivamus condimentum urna eget urna feugiat ornare. Donec vitae tempor sem. Duis scelerisque ipsum ac nunc elementum maximus id sed turpis.
+			Proin finibus sapien lacus, eu iaculis neque tempus sed. Nam aliquam odio quis vestibulum sodales. Nam vehicula dictum libero id venenatis. Duis vulputate ligula in risus ultricies tincidunt. Nulla dignissim augue enim, eget interdum sapien interdum at. Nulla enim ligula, facilisis finibus elit euismod, tempor aliquet ante. Vivamus malesuada quis nisi vitae varius. Maecenas id velit convallis, vehicula nisl et, volutpat dui.
+			Pellentesque et dignissim arcu. Cras eleifend egestas tellus, sit amet pellentesque quam vestibulum in. Duis id hendrerit tortor, eget dapibus sapien. Nulla massa tellus, sollicitudin sit amet mauris vitae, iaculis commodo sem. Fusce gravida eros in lorem imperdiet vestibulum. Nullam quis dolor nec est dictum euismod. Aliquam id tempus quam. Phasellus sed fringilla dui, aliquet semper nisi.
+			"""
+			.getBytes(StandardCharsets.UTF_8);
+
+	private CompressionService compressionService = new CompressionServiceImpl();
+
+	@Test
+	public void testGzip() throws Exception
+	{
+		test(compressionService::toGzip, 44, compressionService::fromGzip, TEST_DATA_1);
+		test(compressionService::toGzip, 1003, compressionService::fromGzip, TEST_DATA_2);
+		test(compressionService::toGzip, 1120, compressionService::fromGzip, TEST_DATA_3);
+	}
+
+	@Test
+	public void testBzip2() throws Exception
+	{
+		test(compressionService::toBzip2, 63, compressionService::fromBzip2, TEST_DATA_1);
+		test(compressionService::toBzip2, 48, compressionService::fromBzip2, TEST_DATA_2);
+		test(compressionService::toBzip2, 1078, compressionService::fromBzip2, TEST_DATA_3);
+	}
+
+	@Test
+	public void testLzma2() throws Exception
+	{
+		test(compressionService::toLzma2, 80, compressionService::fromLzma2, TEST_DATA_1);
+		test(compressionService::toLzma2, 276, compressionService::fromLzma2, TEST_DATA_2);
+		test(compressionService::toLzma2, 1228, compressionService::fromLzma2, TEST_DATA_3);
+	}
+
+	private void test(Converter compressor, int compressedSize, Converter decompressor, byte[] testData)
+			throws IOException
+	{
+		byte[] compressed;
+		try (InputStream in = compressor.convert(new ByteArrayInputStream(testData)))
+		{
+			compressed = in.readAllBytes();
+		}
+
+		assertNotNull(compressed);
+		assertEquals(compressedSize, compressed.length);
+
+		byte[] uncompressed;
+		try (InputStream in = decompressor.convert(new ByteArrayInputStream(compressed)))
+		{
+			uncompressed = in.readAllBytes();
+		}
+
+		assertNotNull(uncompressed);
+		assertEquals(testData.length, uncompressed.length);
+		assertArrayEquals(testData, uncompressed);
+	}
+}

dsf-bpe/dsf-bpe-process-api-v2/src/main/java/dev/dsf/bpe/v2/ProcessPluginApi.java

@@ -7,6 +7,7 @@
 
 import ca.uhn.fhir.context.FhirContext;
 import dev.dsf.bpe.v2.config.ProxyConfig;
+import dev.dsf.bpe.v2.service.CompressionService;
 import dev.dsf.bpe.v2.service.CryptoService;
 import dev.dsf.bpe.v2.service.DataLogger;
 import dev.dsf.bpe.v2.service.DsfClientProvider;
@@ -59,6 +60,8 @@ public interface ProcessPluginApi
 
 	TaskHelper getTaskHelper();
 
+	CompressionService getCompressionService();
+
 	CryptoService getCryptoService();
 
 	TargetProvider getTargetProvider();