Merge branch 'OZG-7573-files-weiterleitung-bug' into 'main'

Ozg 7573 files weiterleitung bug

See merge request !12

ozgcloud-common-lib/src/main/java/de/ozgcloud/common/binaryfile/GrpcFileUploadUtils.java

@@ -23,225 +23,69 @@
  */
 package de.ozgcloud.common.binaryfile;
 
-import java.io.IOException;
 import java.io.InputStream;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
-import org.apache.commons.io.IOUtils;
-
-import de.ozgcloud.common.errorhandling.TechnicalException;
 import io.grpc.stub.CallStreamObserver;
 import io.grpc.stub.StreamObserver;
 import lombok.AccessLevel;
-import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.NonNull;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
 
 @Log4j2
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class GrpcFileUploadUtils {
 
-	static final int CHUNK_SIZE = 4 * 1024;
-
-	/*
-	 * Q = Request Type; S = Response Type
+	/**
+	 * @param <Q> Request Type
+	 * @param <S> Response Type
+	 * @deprecated use {@link #createStreamExclusiveFileSender(BiFunction, InputStream, Function)} instead
 	 */
+	@Deprecated(since = "4.13.0")
 	public static <Q, S> FileSender<Q, S> createSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream,
 			Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder) {
-		return createSender(chunkBuilder, inputStream, reqObserverBuilder, true);
+		return new FileSender<>(chunkBuilder, inputStream, reqObserverBuilder);
 	}
 
-	public static <Q, S> FileSender<Q, S> createSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream,
-			Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder, boolean completeOnFileSent) {
-		return new FileSender<>(chunkBuilder, reqObserverBuilder, inputStream, completeOnFileSent);
+	/**
+	 * @param <Q> Request Type
+	 * @param <S> Response Type
+	 */
+	public static <Q, S> StreamingFileSender<Q, S> createStreamExclusiveFileSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream,
+			Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder) {
+		return new StreamExclusiveFileSender<>(chunkBuilder, inputStream, reqObserverBuilder);
 	}
 
-	public static class FileSender<Q, S> {
-		private final BiFunction<byte[], Integer, Q> chunkBuilder;
-		private final InputStream inputStream;
-
-		@Getter
-		private final CompletableFuture<S> resultFuture = new CompletableFuture<>();
-		private final Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder;
-		private CallStreamObserver<Q> requestObserver;
-
-		private Optional<Q> metaData = Optional.empty();
-		private final AtomicBoolean metaDataSent = new AtomicBoolean(false);
-		private final AtomicBoolean done = new AtomicBoolean(false);
-
-		private final StreamReader streamReader;
-		private final boolean completeOnFileSent;
-
-		FileSender(BiFunction<byte[], Integer, Q> chunkBuilder, Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder,
-				InputStream inputStream) {
-			this(chunkBuilder, reqObserverBuilder, inputStream, true);
-		}
+	/**
+	 * @param <Q> Request Type
+	 * @param <S> Response Type
+	 */
+	public static <Q, S> StreamingFileSender<Q, S> createStreamSharingSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream,
+			CallStreamObserver<Q> requestObserver, Consumer<Runnable> onReadyHandlerRegistrar) {
+		return new StreamSharingFileSender<>(chunkBuilder, inputStream, requestObserver, onReadyHandlerRegistrar);
+	}
 
-		FileSender(BiFunction<byte[], Integer, Q> chunkBuilder, Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder,
-				InputStream inputStream, boolean completeOnFileSent) {
-			this.chunkBuilder = chunkBuilder;
-			this.inputStream = inputStream;
-			this.reqObserverBuilder = reqObserverBuilder;
-			this.completeOnFileSent = completeOnFileSent;
+	@Deprecated(since = "4.13.0")
+	public static class FileSender<Q, S> extends StreamExclusiveFileSender<Q, S> {
 
-			this.streamReader = new StreamReader(this.inputStream);
-		}
-
-		public FileSender<Q, S> withMetaData(@NonNull Q metaData) {
-			this.metaData = Optional.of(metaData);
-			return this;
+		FileSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream,
+				Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder) {
+			super(chunkBuilder, inputStream, reqObserverBuilder);
 		}
 
+		@Override
 		public FileSender<Q, S> send() {
-			LOG.debug("Start sending File.");
-			var responseObserver = BinaryFileUploadStreamObserver.create(resultFuture, this::sendNext);
-			requestObserver = reqObserverBuilder.apply(responseObserver);
-
+			super.send();
 			return this;
 		}
 
-		public void cancelOnTimeout() {
-			LOG.warn("File transfer canceled on timeout");
-			resultFuture.cancel(true);
-			requestObserver.onError(new TechnicalException("Timeout on waiting for upload."));
-			closeStreams();
-		}
-
-		public void cancelOnError(Throwable t) {
-			LOG.error("File tranfer canceled on error.", t);
-			resultFuture.cancel(true);
-			requestObserver.onError(t);
-			closeStreams();
-		}
-
-		void sendNext() {
-			if (!done.get()) {
-				waitForOberver();
-				sendMetaData();
-				do {
-					LOG.debug("Sending next chunk.");
-					sendNextChunk();
-				} while (!done.get() && isReady());
-				LOG.debug("Finished or waiting to become ready.");
-			}
-		}
-
-		private boolean isReady() {
-			return requestObserver.isReady();
-		}
-
-		private void waitForOberver() {
-			synchronized (this) {
-				while (Objects.isNull(requestObserver)) {
-					try {
-						LOG.debug("wait for observer");
-						wait(300);
-					} catch (InterruptedException e) {
-						LOG.error("Error on waiting for request Observer.", e);
-						Thread.currentThread().interrupt();
-					}
-				}
-			}
-
-		}
-
-		void sendNextChunk() {
-			byte[] contentToSend = streamReader.getNextData();
-
-			if (streamReader.getLastReadSize() > 0) {
-				sendChunk(contentToSend, streamReader.getLastReadSize());
-			} else {
-				endTransfer();
-			}
-		}
-
-		private void endTransfer() {
-			if (completeOnFileSent)
-				requestObserver.onCompleted();
-			else
-				sendEndOfFile();
-			done.set(true);
-			LOG.debug("File Transfer done.");
-			closeStreams();
-
-		}
-
-		private void sendEndOfFile() {
-			sendChunk(new byte[0], streamReader.getLastReadSize());
-		}
-
-		void closeStreams() {
-			LOG.debug("Closing streams");
-			streamReader.close();
-		}
-
-		void sendChunk(byte[] content, int length) {
-			LOG.debug("Sending {} byte Data.", length);
-			var chunk = chunkBuilder.apply(content, length);
-			requestObserver.onNext(chunk);
-		}
-
-		byte[] readFromStream() {
-			try {
-				return inputStream.readNBytes(CHUNK_SIZE);
-			} catch (IOException e) {
-				throw new TechnicalException("Error on sending a single chunk", e);
-			}
-		}
-
-		void sendMetaData() {
-			metaData.filter(md -> !metaDataSent.get()).ifPresent(this::doSendMetaData);
-		}
-
-		private void doSendMetaData(Q metadata) {
-			LOG.debug("Sending Metadata.");
-			requestObserver.onNext(metadata);
-			metaDataSent.set(true);
-		}
-
-		void checkForEndOfStream(long sentSize) {
-			if (sentSize < CHUNK_SIZE) {
-				LOG.debug("File Transfer done. Closing stream.");
-				IOUtils.closeQuietly(inputStream);
-				requestObserver.onCompleted();
-				done.set(true);
-			} else {
-				LOG.debug("File Transfer not jet done - need to tranfer another chunk.");
-			}
-		}
-
-		@RequiredArgsConstructor
-		private class StreamReader {
-			private final InputStream inStream;
-			private final byte[] buffer = new byte[CHUNK_SIZE];
-			@Getter
-			private int lastReadSize = 0;
-			@Getter
-			private final AtomicBoolean done = new AtomicBoolean(false);
-
-			byte[] getNextData() {
-				readNext();
-				return buffer;
-			}
-
-			void close() {
-				IOUtils.closeQuietly(inStream);
-			}
-
-			void readNext() {
-				try {
-					lastReadSize = inStream.read(buffer, 0, CHUNK_SIZE);
-				} catch (IOException e) {
-					throw new TechnicalException("Error on reading a single chunk", e);
-				}
-			}
+		@Override
+		public FileSender<Q, S> withMetaData(@NonNull Q metaData) {
+			super.withMetaData(metaData);
+			return this;
 		}
 	}
 

ozgcloud-common-lib/src/main/java/de/ozgcloud/common/binaryfile/StreamExclusiveFileSender.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import java.io.InputStream;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import io.grpc.stub.CallStreamObserver;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.log4j.Log4j2;
+
+@Log4j2
+class StreamExclusiveFileSender<Q, S> extends StreamingFileSender<Q, S> {
+
+	private final Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder;
+	private CallStreamObserver<Q> requestObserver;
+
+	StreamExclusiveFileSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream, Function<StreamObserver<S>, CallStreamObserver<Q>> reqObserverBuilder) {
+		super(chunkBuilder, inputStream);
+
+		this.reqObserverBuilder = reqObserverBuilder;
+	}
+
+	@Override
+	public StreamExclusiveFileSender<Q, S> send() {
+		LOG.debug("Start sending File.");
+
+		var responseObserver = BinaryFileUploadStreamObserver.create(getResultFuture(), this::sendNext);
+		requestObserver = reqObserverBuilder.apply(responseObserver);
+
+		return this;
+	}
+
+	@Override
+	void communicateEndOfTransfer() {
+		requestObserver.onCompleted();
+	}
+
+	@Override
+	protected CallStreamObserver<Q> getRequestObserver() {
+		return requestObserver;
+	}
+
+	@Override
+	public void cancelOnTimeout() {
+		super.cancelOnTimeout();
+		requestObserver.onError(new TechnicalException("Timeout on waiting for upload."));
+	}
+
+	@Override
+	public void cancelOnError(Throwable t) {
+		super.cancelOnError(t);
+		requestObserver.onError(t);
+	}
+}

ozgcloud-common-lib/src/main/java/de/ozgcloud/common/binaryfile/StreamSharingFileSender.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import java.io.InputStream;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+import io.grpc.stub.CallStreamObserver;
+import lombok.extern.log4j.Log4j2;
+
+@Log4j2
+class StreamSharingFileSender<Q, S> extends StreamingFileSender<Q, S> {
+
+	private final CallStreamObserver<Q> requestObserver;
+	private final Consumer<Runnable> onReadyHandlerRegistrar;
+
+	StreamSharingFileSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream, CallStreamObserver<Q> requestObserver, Consumer<Runnable> onReadyHandlerRegistrar) {
+		super(chunkBuilder, inputStream);
+
+		this.requestObserver = requestObserver;
+		this.onReadyHandlerRegistrar = onReadyHandlerRegistrar;
+	}
+
+	@Override
+	public StreamSharingFileSender<Q, S> send() {
+		LOG.debug("Register onReadyHandler and start sending File.");
+		onReadyHandlerRegistrar.accept(this::sendNext);
+		return this;
+	}
+
+	@Override
+	void communicateEndOfTransfer() {
+		sendEndOfFile();
+		getResultFuture().complete(null);
+	}
+
+	private void sendEndOfFile() {
+		sendChunk(new byte[0], -1);
+	}
+
+	@Override
+	protected CallStreamObserver<Q> getRequestObserver() {
+		return requestObserver;
+	}
+}

ozgcloud-common-lib/src/main/java/de/ozgcloud/common/binaryfile/StreamingFileSender.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiFunction;
+
+import org.apache.commons.io.IOUtils;
+
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import io.grpc.stub.CallStreamObserver;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+/*
+ * Q = Request Type; S = Response Type
+ */
+@Log4j2
+public abstract class StreamingFileSender<Q, S> {
+
+	static final int CHUNK_SIZE = 4 * 1024;
+	
+	private final BiFunction<byte[], Integer, Q> chunkBuilder;
+
+	@Getter
+	private final CompletableFuture<S> resultFuture = new CompletableFuture<>();
+
+	private Q metaData;
+	private final AtomicBoolean metaDataSent = new AtomicBoolean(false);
+	private final AtomicBoolean done = new AtomicBoolean(false);
+
+	@Getter(AccessLevel.PROTECTED)
+	private final StreamReader streamReader;
+
+	StreamingFileSender(BiFunction<byte[], Integer, Q> chunkBuilder, InputStream inputStream) {
+		this.chunkBuilder = chunkBuilder;
+		this.streamReader = new StreamReader(inputStream);
+	}
+
+	public StreamingFileSender<Q, S> withMetaData(@NonNull Q metaData) {
+		this.metaData = metaData;
+		return this;
+	}
+
+	public abstract StreamingFileSender<Q, S> send();
+
+	public void cancelOnTimeout() {
+		LOG.warn("File transfer canceled on timeout");
+		resultFuture.cancel(true);
+		closeStreamReader();
+	}
+
+	public void cancelOnError(Throwable t) {
+		LOG.error("File transfer canceled on error.", t);
+		resultFuture.cancel(true);
+		closeStreamReader();
+	}
+
+	void sendNext() {
+		if (notFinished()) {
+			waitForObserver();
+			sendMetaData();
+			do {
+				LOG.debug("Sending next chunk.");
+				sendNextChunk();
+			} while (notFinished() && isReady());
+			LOG.debug("Finished or waiting to become ready.");
+		}
+		checkIfFinishedForcefully();
+	}
+
+	private boolean notFinished() {
+		return !done.get() && !resultFuture.isCancelled();
+	}
+
+	private void checkIfFinishedForcefully() {
+		if (resultFuture.isCancelled()) {
+			LOG.warn("File transfer was cancelled");
+			closeStreamReader();
+		}
+	}
+
+	private boolean isReady() {
+		return getRequestObserver().isReady();
+	}
+
+	private void waitForObserver() {
+		synchronized (this) {
+			while (Objects.isNull(getRequestObserver())) {
+				try {
+					LOG.debug("wait for observer");
+					wait(300);
+				} catch (InterruptedException e) {
+					LOG.error("Error on waiting for request Observer.", e);
+					Thread.currentThread().interrupt();
+				}
+			}
+		}
+
+	}
+
+	void sendNextChunk() {
+		byte[] contentToSend = streamReader.getNextData();
+
+		if (streamReader.getLastReadSize() > 0) {
+			sendChunk(contentToSend, streamReader.getLastReadSize());
+		} else {
+			endTransfer();
+		}
+	}
+
+	void endTransfer() {
+		communicateEndOfTransfer();
+		done.set(true);
+		LOG.debug("File Transfer done.");
+		closeStreamReader();
+	}
+
+	abstract void communicateEndOfTransfer();
+
+	void closeStreamReader() {
+		LOG.debug("Closing streams");
+		streamReader.close();
+	}
+
+	void sendChunk(byte[] content, int length) {
+		LOG.debug("Sending {} byte Data.", length);
+		var chunk = chunkBuilder.apply(content, length);
+		getRequestObserver().onNext(chunk);
+	}
+
+	void sendMetaData() {
+		if (metaData != null && !metaDataSent.get()) {
+			doSendMetaData(metaData);
+		}
+	}
+
+	private void doSendMetaData(Q metadata) {
+		LOG.debug("Sending Metadata.");
+		getRequestObserver().onNext(metadata);
+		metaDataSent.set(true);
+	}
+
+	protected abstract CallStreamObserver<Q> getRequestObserver();
+
+	@RequiredArgsConstructor
+	protected static class StreamReader {
+		private final InputStream inStream;
+		private final byte[] buffer = new byte[CHUNK_SIZE];
+		@Getter
+		private int lastReadSize = 0;
+		@Getter
+		private final AtomicBoolean done = new AtomicBoolean(false);
+
+		byte[] getNextData() {
+			readNext();
+			return buffer;
+		}
+
+		void close() {
+			IOUtils.closeQuietly(inStream);
+		}
+
+		void readNext() {
+			try {
+				lastReadSize = inStream.read(buffer, 0, CHUNK_SIZE);
+			} catch (IOException e) {
+				throw new TechnicalException("Error on reading a single chunk", e);
+			}
+		}
+	}
+}

ozgcloud-common-lib/src/test/java/de/ozgcloud/common/binaryfile/GrpcFileUploadUtilsTest.java

@@ -24,289 +24,64 @@
 package de.ozgcloud.common.binaryfile;
 
 import static org.assertj.core.api.Assertions.*;
-import static org.mockito.ArgumentMatchers.*;
-import static org.mockito.Mockito.*;
 
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
-import org.apache.commons.lang3.RandomUtils;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Captor;
-import org.mockito.InjectMocks;
 import org.mockito.Mock;
 
 import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestRequestType;
 import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestResponseType;
-import de.ozgcloud.common.binaryfile.GrpcFileUploadUtils.FileSender;
-import de.ozgcloud.common.errorhandling.TechnicalException;
 import io.grpc.stub.CallStreamObserver;
 import io.grpc.stub.StreamObserver;
-import lombok.SneakyThrows;
 
 class GrpcFileUploadUtilsTest {
 
-	@InjectMocks
-	private GrpcFileUploadUtils service;
-
 	@Mock
 	private BiFunction<byte[], Integer, TestRequestType> chunkBuilder;
 	@Mock
+	private InputStream inputStream;
+	@Mock
 	private Function<StreamObserver<TestResponseType>, CallStreamObserver<TestRequestType>> reqObserverBuilder;
 	@Mock
 	private CallStreamObserver<TestRequestType> requestObserver;
 	@Mock
-	private InputStream inputStream;
-
-	private FileSender<TestRequestType, TestResponseType> fileSender;
-
-	@Mock
-	private TestRequestType metaData;
-
-	@BeforeEach
-	void init() {
-		when(reqObserverBuilder.apply(any())).thenReturn(requestObserver);
-		fileSender = spy(GrpcFileUploadUtils.createSender(chunkBuilder, inputStream, reqObserverBuilder));
-	}
+	private Consumer<Runnable> onReadyHandlerRegistrar;
 
 	@Nested
-	class TestCreateFileSender {
+	class TestCreateSender {
 
 		@Test
-		void shouldCreateRequestObserver() {
-			GrpcFileUploadUtils.createSender(chunkBuilder, inputStream, reqObserverBuilder).send();
+		void shouldReturnInstanceOfStreamExclusiveFileSender() {
+			var createdSender = GrpcFileUploadUtils.createSender(chunkBuilder, inputStream, reqObserverBuilder);
 
-			verify(reqObserverBuilder, atLeastOnce()).apply(notNull());
+			assertThat(createdSender).isInstanceOf(StreamExclusiveFileSender.class);
 		}
 	}
 
 	@Nested
-	class TestSendBinaryFile {
-
-		@Captor
-		private ArgumentCaptor<Runnable> runnableCaptor;
+	class TestCreateStreamExclusiveFileSender {
 
 		@Test
-		void shouldReturnSenderWithFuture() {
-			var result = fileSender.send();
+		void shouldReturnInstanceOfStreamExclusiveFileSender() {
+			var createdSender = GrpcFileUploadUtils.createStreamExclusiveFileSender(chunkBuilder, inputStream, reqObserverBuilder);
 
-			assertThat(result).isNotNull().extracting(FileSender::getResultFuture).isNotNull();
+			assertThat(createdSender).isInstanceOf(StreamExclusiveFileSender.class);
 		}
 	}
 
 	@Nested
-	class TestSendNext {
-
-		@BeforeEach
-		void initObserver() {
-			fileSender.send();
-		}
+	class TestCreateStreamSharingSender {
 
 		@Test
-		void shouldCallSendMetaData() {
-			fileSender.sendNext();
-
-			verify(fileSender).sendMetaData();
-		}
-
-		@Test
-		void shouldSendNextChunk() {
-			fileSender.sendNext();
-
-			verify(fileSender).sendNextChunk();
-		}
-
-	}
-
-	@Nested
-	class TestSendNextChunk {
-
-		private final byte[] content = RandomUtils.insecure().randomBytes(GrpcFileUploadUtils.CHUNK_SIZE / 2);
-		private ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(content);
+		void shouldReturnInstanceOfStreamSharingSender() {
+			var createdSender = GrpcFileUploadUtils.createStreamSharingSender(chunkBuilder, inputStream, requestObserver, onReadyHandlerRegistrar);
 
-		@Captor
-		private ArgumentCaptor<byte[]> chunkCaptor;
-
-		@Nested
-		class TestOnDataAvailable {
-			@BeforeEach
-			void initObserver() {
-				fileSender = spy(GrpcFileUploadUtils.createSender(chunkBuilder, byteArrayInputStream, reqObserverBuilder));
-				fileSender.send();
-			}
-
-			@Test
-			void shouldCallSendChunk() {
-				fileSender.sendNextChunk();
-
-				verify(fileSender).sendChunk(chunkCaptor.capture(), eq(content.length));
-				assertThat(chunkCaptor.getValue()).contains(content);
-			}
+			assertThat(createdSender).isInstanceOf(StreamSharingFileSender.class);
 		}
-
-		@Nested
-		class TestOnNoBytesLeftToRead {
-
-			@Nested
-			class TestOnCompleteOnFileSent {
-				private static final boolean COMPLETE_ON_FILE_SENT = true;
-
-				@BeforeEach
-				void initialize() {
-					var buffer = new byte[GrpcFileUploadUtils.CHUNK_SIZE];
-					byteArrayInputStream.read(buffer, 0, GrpcFileUploadUtils.CHUNK_SIZE);
-					fileSender = spy(GrpcFileUploadUtils.createSender(chunkBuilder, byteArrayInputStream, reqObserverBuilder, COMPLETE_ON_FILE_SENT));
-					fileSender.send();
-				}
-
-				@Test
-				void shouldCallOnCompleted() {
-					fileSender.sendNextChunk();
-
-					verify(requestObserver).onCompleted();
-				}
-
-				@Test
-				void shouldNotCallSendChunk() {
-					fileSender.sendNextChunk();
-
-					verify(fileSender, never()).sendChunk(any(), anyInt());
-				}
-
-				@Test
-				@SneakyThrows
-				void shouldCallCloseStreams() {
-					fileSender.sendNextChunk();
-
-					verify(fileSender).closeStreams();
-				}
-			}
-
-			@Nested
-			class TestOnNotCompleteOnFileSent {
-				private static final boolean COMPLETE_ON_FILE_SENT = false;
-
-				@BeforeEach
-				void initialize() {
-					var buffer = new byte[GrpcFileUploadUtils.CHUNK_SIZE];
-					byteArrayInputStream.read(buffer, 0, GrpcFileUploadUtils.CHUNK_SIZE);
-					fileSender = spy(GrpcFileUploadUtils.createSender(chunkBuilder, byteArrayInputStream, reqObserverBuilder, COMPLETE_ON_FILE_SENT));
-					fileSender.send();
-				}
-
-				@Test
-				void shouldNotCallOnCompleted() {
-					fileSender.sendNextChunk();
-
-					verify(requestObserver, never()).onCompleted();
-				}
-
-				@Test
-				void shouldCallSendChunk() {
-					fileSender.sendNextChunk();
-
-					verify(fileSender).sendChunk(chunkCaptor.capture(), eq(-1));
-					assertThat(chunkCaptor.getValue()).isEmpty();
-				}
-
-				@Test
-				@SneakyThrows
-				void shouldCallCloseStreams() {
-					fileSender.sendNextChunk();
-
-					verify(fileSender).closeStreams();
-				}
-			}
-		}
-
 	}
-
-	@Nested
-	class TestCloseStreams {
-
-		@Test
-		@SneakyThrows
-		void shouldCloseInputStream() {
-			fileSender.send();
-
-			fileSender.closeStreams();
-
-			verify(inputStream).close();
-		}
-	}
-
-	@Nested
-	class TestSendChunk {
-
-		private static final byte[] CHUNK_PART = "ChunkPartContent".getBytes();
-
-		@BeforeEach
-		void initObserver() {
-			fileSender.send();
-		}
-
-		@Test
-		void shouldApplyBuildChunk() throws IOException {
-			fileSender.sendChunk(CHUNK_PART, 5);
-
-			verify(chunkBuilder).apply(CHUNK_PART, 5);
-		}
-
-		@Test
-		void shouldCallOnNext() throws IOException {
-			fileSender.sendChunk(CHUNK_PART, 5);
-
-			verify(requestObserver).onNext(any());
-		}
-	}
-
-	@Nested
-	class TestSendMetaData {
-
-		@BeforeEach
-		void initObserver() {
-			fileSender.send();
-		}
-
-		@Test
-		void shouldNotSendWithoutMetadata() {
-			fileSender.sendMetaData();
-
-			verify(requestObserver, never()).onNext(any());
-		}
-
-		@Test
-		void shouldSendMetadata() {
-			fileSender.withMetaData(metaData).sendMetaData();
-
-			verify(requestObserver).onNext(metaData);
-		}
-
-		@Test
-		void shouldSendMetadataOnlyOnce() {
-			fileSender.withMetaData(metaData).sendMetaData();
-			fileSender.sendMetaData();
-
-			verify(requestObserver).onNext(metaData);
-		}
-	}
-
-	@Disabled("unused")
-	@Nested
-	class TestReadFromStream {
-		@Test
-		void shouldThrowException() throws IOException {
-			doThrow(IOException.class).when(inputStream).read(any(), anyInt(), anyInt());
-
-			assertThatThrownBy(() -> fileSender.readFromStream()).isInstanceOf(TechnicalException.class);
-		}
-	}
-
 }
\ No newline at end of file

ozgcloud-common-lib/src/test/java/de/ozgcloud/common/binaryfile/StreamExclusiveFileSenderTest.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import static org.mockito.Mockito.*;
+
+import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.Spy;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestRequestType;
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestResponseType;
+import de.ozgcloud.common.errorhandling.TechnicalException;
+import io.grpc.stub.CallStreamObserver;
+import io.grpc.stub.StreamObserver;
+
+class StreamExclusiveFileSenderTest {
+
+	@Mock
+	private BiFunction<byte[], Integer, TestRequestType> chunkBuilder;
+	@Mock
+	private InputStream inputStream;
+	@Mock
+	private Function<StreamObserver<TestResponseType>, CallStreamObserver<TestRequestType>> reqObserverBuilder;
+	@Spy
+	@InjectMocks
+	private StreamExclusiveFileSender<TestRequestType, TestResponseType> fileSender;
+
+	@Nested
+	class TestSend {
+
+		private final CompletableFuture<TestResponseType> resultFuture = CompletableFuture.completedFuture(new TestResponseType());
+		@Captor
+		private ArgumentCaptor<Runnable> onReadyHandlerCaptor;
+		@Mock
+		private BinaryFileUploadStreamObserver<TestRequestType, TestResponseType> responseObserver;
+
+		@BeforeEach
+		void init() {
+			doReturn(resultFuture).when(fileSender).getResultFuture();
+		}
+
+		@SuppressWarnings("rawtypes")
+		@Test
+		void shouldCreateResponseObserver() {
+			try (MockedStatic<BinaryFileUploadStreamObserver> mocked = mockStatic(BinaryFileUploadStreamObserver.class)) {
+				doNothing().when(fileSender).sendNext();
+
+				fileSender.send();
+
+				mocked.verify(() -> BinaryFileUploadStreamObserver.create(same(resultFuture), onReadyHandlerCaptor.capture()));
+				verifyCallsSendNext(onReadyHandlerCaptor.getValue());
+			}
+		}
+
+		@SuppressWarnings("rawtypes")
+		@Test
+		void shouldBuildRequestObserver() {
+			try (MockedStatic<BinaryFileUploadStreamObserver> mocked = mockStatic(BinaryFileUploadStreamObserver.class)) {
+				mocked.when(() -> BinaryFileUploadStreamObserver.create(any(), any())).thenReturn(responseObserver);
+
+				fileSender.send();
+
+				verify(reqObserverBuilder).apply(responseObserver);
+			}
+		}
+
+		private void verifyCallsSendNext(Runnable runnable) {
+			runnable.run();
+			verify(fileSender).sendNext();
+		}
+	}
+
+	@Nested
+	class TestCommunicateEndOfTransfer {
+
+		@Mock
+		private CallStreamObserver<TestRequestType> requestObserver;
+
+		@Test
+		void shouldCallOnCompleted() {
+			ReflectionTestUtils.setField(fileSender, "requestObserver", requestObserver);
+
+			fileSender.communicateEndOfTransfer();
+
+			verify(requestObserver).onCompleted();
+		}
+	}
+
+	@Nested
+	class TestCancelOnTimeout {
+
+		@Mock
+		private CompletableFuture<TestResponseType> resultFuture;
+		@Mock
+		private CallStreamObserver<TestRequestType> requestObserver;
+
+		@BeforeEach
+		void init() {
+			ReflectionTestUtils.setField(fileSender, "resultFuture", resultFuture);
+			ReflectionTestUtils.setField(fileSender, "requestObserver", requestObserver);
+		}
+
+		@Test
+		void shouldCancelResultFuture() {
+			fileSender.cancelOnTimeout();
+
+			verify(resultFuture).cancel(true);
+		}
+
+		@Test
+		void shouldCallOnError() {
+			fileSender.cancelOnTimeout();
+
+			verify(requestObserver).onError(any(TechnicalException.class));
+		}
+
+		@Test
+		void shouldCloseStreamReader() {
+			fileSender.cancelOnTimeout();
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+
+	@Nested
+	class TestCancelOnError {
+
+		@Mock
+		private CompletableFuture<TestResponseType> resultFuture;
+		@Mock
+		private CallStreamObserver<TestRequestType> requestObserver;
+		private final Throwable error = new Throwable();
+
+		@BeforeEach
+		void init() {
+			ReflectionTestUtils.setField(fileSender, "resultFuture", resultFuture);
+			ReflectionTestUtils.setField(fileSender, "requestObserver", requestObserver);
+		}
+
+		@Test
+		void shouldCancelResultFuture() {
+			fileSender.cancelOnError(error);
+
+			verify(resultFuture).cancel(true);
+		}
+
+		@Test
+		void shouldCallOnError() {
+			fileSender.cancelOnError(error);
+
+			verify(requestObserver).onError(error);
+		}
+
+		@Test
+		void shouldCloseStreamReader() {
+			fileSender.cancelOnError(error);
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+}

ozgcloud-common-lib/src/test/java/de/ozgcloud/common/binaryfile/StreamSharingFileSenderTest.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.io.InputStream;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestRequestType;
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestResponseType;
+import io.grpc.stub.CallStreamObserver;
+
+class StreamSharingFileSenderTest {
+
+	@Mock
+	private BiFunction<byte[], Integer, TestRequestType> chunkBuilder;
+	@Mock
+	private InputStream inputStream;
+	@Mock
+	private CallStreamObserver<TestRequestType> requestObserver;
+	@Mock
+	private Consumer<Runnable> onReadyHandlerRegistrar;
+	@Spy
+	@InjectMocks
+	private StreamSharingFileSender<TestRequestType, TestResponseType> fileSender;
+
+	@Nested
+	class TestSend {
+
+		@Captor
+		private ArgumentCaptor<Runnable> onReadyHandlerCaptor;
+
+		@Test
+		void shouldRegisterOnReadyHandler() {
+			doNothing().when(fileSender).sendNext();
+
+			fileSender.send();
+
+			verify(onReadyHandlerRegistrar).accept(onReadyHandlerCaptor.capture());
+			verifyCallsSendNext(onReadyHandlerCaptor.getValue());
+		}
+
+		@Test
+		void shouldReturnThis() {
+			var obj = fileSender.send();
+
+			assertThat(obj).isSameAs(fileSender);
+		}
+
+		private void verifyCallsSendNext(Runnable runnable) {
+			runnable.run();
+			verify(fileSender).sendNext();
+		}
+	}
+
+	@Nested
+	class TestCommunicateEndOfTransfer {
+
+		@Test
+		void shouldSendEmptyChunk() {
+			fileSender.communicateEndOfTransfer();
+
+			verify(fileSender).sendChunk(new byte[0], -1);
+		}
+
+		@Test
+		void shouldCompleteResultFuture() {
+			fileSender.communicateEndOfTransfer();
+
+			assertThat(fileSender.getResultFuture().isDone()).isTrue();
+		}
+	}
+}

ozgcloud-common-lib/src/test/java/de/ozgcloud/common/binaryfile/StreamingFileSenderTest.java

+/*
+ * Copyright (C) 2025 Das Land Schleswig-Holstein vertreten durch den
+ * Ministerpräsidenten des Landes Schleswig-Holstein
+ * Staatskanzlei
+ * Abteilung Digitalisierung und zentrales IT-Management der Landesregierung
+ *
+ * Lizenziert unter der EUPL, Version 1.2 oder - sobald
+ * diese von der Europäischen Kommission genehmigt wurden -
+ * Folgeversionen der EUPL ("Lizenz");
+ * Sie dürfen dieses Werk ausschließlich gemäß
+ * dieser Lizenz nutzen.
+ * Eine Kopie der Lizenz finden Sie hier:
+ *
+ * https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
+ *
+ * Sofern nicht durch anwendbare Rechtsvorschriften
+ * gefordert oder in schriftlicher Form vereinbart, wird
+ * die unter der Lizenz verbreitete Software "so wie sie
+ * ist", OHNE JEGLICHE GEWÄHRLEISTUNG ODER BEDINGUNGEN -
+ * ausdrücklich oder stillschweigend - verbreitet.
+ * Die sprachspezifischen Genehmigungen und Beschränkungen
+ * unter der Lizenz sind dem Lizenztext zu entnehmen.
+ */
+package de.ozgcloud.common.binaryfile;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiFunction;
+
+import org.apache.commons.lang3.RandomUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestRequestType;
+import de.ozgcloud.common.binaryfile.BinaryFileTestFactory.TestResponseType;
+import io.grpc.stub.CallStreamObserver;
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.SneakyThrows;
+
+class StreamingFileSenderTest {
+
+	@Mock
+	private BiFunction<byte[], Integer, TestRequestType> chunkBuilder;
+	@Mock
+	private InputStream inputStream;
+	@Mock
+	private CallStreamObserver<TestRequestType> requestObserver;
+
+	private TestFileSender fileSender;
+
+	@BeforeEach
+	void init() {
+		fileSender = spy(new TestFileSender(chunkBuilder, inputStream, requestObserver));
+	}
+
+	@Nested
+	class TestCancelOnTimeout {
+
+		@Mock
+		private CompletableFuture<TestResponseType> resultFuture;
+
+		@BeforeEach
+		void init() {
+			setResultFutureInFileSender(resultFuture);
+		}
+
+		@Test
+		void shouldCancelResultFuture() {
+			fileSender.cancelOnTimeout();
+
+			verify(resultFuture).cancel(true);
+		}
+
+		@Test
+		void shouldCloseStreams() {
+			fileSender.cancelOnTimeout();
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+
+	@Nested
+	class TestCancelOnError {
+
+		@Mock
+		private CompletableFuture<TestResponseType> resultFuture;
+		private final Throwable error = new Throwable();
+
+		@BeforeEach
+		void init() {
+			setResultFutureInFileSender(resultFuture);
+		}
+
+		@Test
+		void shouldCancelResultFuture() {
+			fileSender.cancelOnError(error);
+
+			verify(resultFuture).cancel(true);
+		}
+
+		@Test
+		void shouldCloseStreams() {
+			fileSender.cancelOnError(error);
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+
+	@Nested
+	class TestSendNext {
+
+		@Mock
+		private CompletableFuture<TestResponseType> resultFuture;
+
+		@BeforeEach
+		void init() {
+			fileSender.send();
+		}
+
+		@Test
+		void shouldSendMetaData() {
+			fileSender.sendNext();
+
+			verify(fileSender).sendMetaData();
+		}
+
+		@Test
+		void shouldSendNextChunk() {
+			fileSender.sendNext();
+
+			verify(fileSender).sendNextChunk();
+		}
+
+		@Test
+		void shouldNotSendMetaDataIfDone() {
+			setDoneInFileSender(true);
+
+			fileSender.sendNext();
+
+			verify(fileSender, never()).sendMetaData();
+		}
+
+		@Test
+		void shouldNotSendMetaDataIfCancelled() {
+			fileSender.getResultFuture().cancel(true);
+
+			fileSender.sendNext();
+
+			verify(fileSender, never()).sendMetaData();
+		}
+
+		@Test
+		void shouldSendNextChunkUntilDone() {
+			lenient().when(requestObserver.isReady()).thenReturn(true);
+			doAnswer(invocation -> {
+				setDoneInFileSender(true);
+				return null;
+			}).when(fileSender).sendNextChunk();
+
+			fileSender.sendNext();
+
+			verify(fileSender, times(1)).sendNextChunk();
+		}
+
+		@Test
+		void shouldSendNextChunkUntilCancelled() {
+			lenient().when(requestObserver.isReady()).thenReturn(true);
+			doAnswer(invocation -> {
+				fileSender.getResultFuture().cancel(true);
+				return null;
+			}).when(fileSender).sendNextChunk();
+
+			fileSender.sendNext();
+
+			verify(fileSender, times(1)).sendNextChunk();
+		}
+
+		@Test
+		void closeStreamReaderIfCancelled() {
+			fileSender.getResultFuture().cancel(true);
+
+			fileSender.sendNext();
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+
+	@Nested
+	class TestSendNextChunk {
+
+		private final byte[] content = RandomUtils.insecure().randomBytes(StreamingFileSender.CHUNK_SIZE / 2);
+		private final ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(content);
+
+		@Captor
+		private ArgumentCaptor<byte[]> chunkCaptor;
+
+		@Nested
+		class TestOnDataAvailable {
+
+			@BeforeEach
+			void init() {
+				fileSender = spy(new TestFileSender(chunkBuilder, byteArrayInputStream, requestObserver));
+			}
+
+			@Test
+			void shouldSendChunk() {
+				fileSender.sendNextChunk();
+
+				verify(fileSender).sendChunk(chunkCaptor.capture(), eq(content.length));
+				assertThat(chunkCaptor.getValue()).contains(content);
+			}
+		}
+
+		@Nested
+		class TestOnNoBytesLeftToRead {
+
+			@BeforeEach
+			void init() {
+				var buffer = new byte[StreamingFileSender.CHUNK_SIZE];
+				byteArrayInputStream.read(buffer, 0, StreamingFileSender.CHUNK_SIZE);
+				fileSender = spy(new TestFileSender(chunkBuilder, byteArrayInputStream, requestObserver));
+			}
+
+			@Test
+			void shouldNotSendChunk() {
+				fileSender.sendNextChunk();
+
+				verify(fileSender, never()).sendChunk(any(), anyInt());
+			}
+
+			@Test
+			void shouldEndTransfer() {
+				fileSender.sendNextChunk();
+
+				verify(fileSender).endTransfer();
+			}
+		}
+	}
+
+	@Nested
+	class TestEndTransfer {
+
+		@Test
+		void shouldCommunicateEndOfTransfer() {
+			fileSender.endTransfer();
+
+			verify(fileSender).communicateEndOfTransfer();
+		}
+
+		@Test
+		void shouldSetDoneToTrue() {
+			fileSender.endTransfer();
+
+			fileSender.sendNext();
+			verify(fileSender, never()).sendNextChunk();
+		}
+
+		@Test
+		void shouldCloseStreamReader() {
+			fileSender.endTransfer();
+
+			verify(fileSender).closeStreamReader();
+		}
+	}
+
+	@Nested
+	class TestCloseStreamReader {
+
+		@Test
+		@SneakyThrows
+		void shouldCloseInputStream() {
+			fileSender.closeStreamReader();
+
+			verify(inputStream).close();
+		}
+	}
+
+	@Nested
+	class TestSendChunk {
+
+		private static final byte[] CHUNK_PART = "ChunkPartContent".getBytes();
+
+		@BeforeEach
+		void initObserver() {
+			fileSender.send();
+		}
+
+		@Test
+		void shouldApplyBuildChunk() {
+			fileSender.sendChunk(CHUNK_PART, 5);
+
+			verify(chunkBuilder).apply(CHUNK_PART, 5);
+		}
+
+		@Test
+		void shouldCallOnNext() {
+			fileSender.sendChunk(CHUNK_PART, 5);
+
+			verify(requestObserver).onNext(any());
+		}
+	}
+
+	@Nested
+	class TestSendMetaData {
+
+		@Mock
+		private TestRequestType metaData;
+
+		@BeforeEach
+		void initObserver() {
+			fileSender.send();
+		}
+
+		@Test
+		void shouldNotSendWithoutMetadata() {
+			fileSender.sendMetaData();
+
+			verify(requestObserver, never()).onNext(any());
+		}
+
+		@Test
+		void shouldSendMetadata() {
+			fileSender.withMetaData(metaData).sendMetaData();
+
+			verify(requestObserver).onNext(metaData);
+		}
+
+		@Test
+		void shouldSendMetadataOnlyOnce() {
+			fileSender.withMetaData(metaData).sendMetaData();
+
+			fileSender.sendMetaData();
+
+			verify(requestObserver).onNext(metaData);
+		}
+	}
+
+	private void setResultFutureInFileSender(CompletableFuture<TestResponseType> resultFuture) {
+		ReflectionTestUtils.setField(fileSender, "resultFuture", resultFuture);
+	}
+
+	private void setDoneInFileSender(boolean done) {
+		((AtomicBoolean) ReflectionTestUtils.getField(fileSender, null, "done")).set(done);
+	}
+
+	static class TestFileSender extends StreamingFileSender<TestRequestType, TestResponseType> {
+
+		@Getter(AccessLevel.PROTECTED)
+		private final CallStreamObserver<TestRequestType> requestObserver;
+
+		TestFileSender(BiFunction<byte[], Integer, TestRequestType> chunkBuilder, InputStream inputStream,
+				CallStreamObserver<TestRequestType> requestObserver) {
+			super(chunkBuilder, inputStream);
+			this.requestObserver = requestObserver;
+		}
+
+		@Override
+		public StreamingFileSender<TestRequestType, TestResponseType> send() {
+			return this;
+		}
+
+		@Override
+		void communicateEndOfTransfer() {
+		}
+	}
+}