PR comments - keep transformerCount in FileAsyncResponseTransformerPublisher

Author: L-Applin

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisher.java

@@ -19,7 +19,6 @@
 import java.nio.file.Path;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.reactivestreams.Subscriber;
 import software.amazon.awssdk.annotations.SdkInternalApi;
@@ -46,6 +45,8 @@ public class FileAsyncResponseTransformerPublisher<T extends SdkResponse>
     private final FileTransformerConfiguration initialConfig;
     private final long initialPosition;
     private Subscriber<?> subscriber;
+    private final AtomicLong transformerCount;
+
 
     public FileAsyncResponseTransformerPublisher(FileAsyncResponseTransformer<?> responseTransformer) {
         this.path = Validate.paramNotNull(responseTransformer.path(), "path");
@@ -54,6 +55,7 @@ public FileAsyncResponseTransformerPublisher(FileAsyncResponseTransformer<?> res
                         "CREATE_OR_APPEND_TO_EXISTING is not supported for non-serial operations");
         this.initialConfig = Validate.paramNotNull(responseTransformer.config(), "fileTransformerConfiguration");
         this.initialPosition = responseTransformer.position();
+        this.transformerCount = new AtomicLong(0);
     }
 
     @Override
@@ -129,10 +131,14 @@ public void onResponse(T response) {
             CompletableFuture<T> delegateFuture = delegate.prepare();
             CompletableFutureUtils.forwardResultTo(delegateFuture, future);
             CompletableFutureUtils.forwardExceptionTo(future, delegateFuture);
+            transformerCount.incrementAndGet();
             delegate.onResponse(response);
         }
 
         private AsyncResponseTransformer<T, T> getDelegateTransformer(Long startAt) {
+            if (transformerCount.get() == 0) {
+                return AsyncResponseTransformer.toFile(path, initialConfig);
+            }
             switch (initialConfig.fileWriteOption()) {
                 case CREATE_NEW:
                 case CREATE_OR_REPLACE_EXISTING: {

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/async/FileAsyncResponseTransformerPublisherTest.java

@@ -20,15 +20,13 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.google.common.jimfs.Configuration;
 import com.google.common.jimfs.Jimfs;
 import java.nio.ByteBuffer;
 import java.nio.file.FileSystem;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -54,7 +52,7 @@ class FileAsyncResponseTransformerPublisherTest {
 
     @BeforeEach
     void setUp() throws Exception {
-        fileSystem = Jimfs.newFileSystem(Configuration.unix());
+        fileSystem = Jimfs.newFileSystem();
         testFile = fileSystem.getPath("/test-file.txt");
     }