Reproduce missing step (#100)

* Allow passing in env

* Add asset

* test: add embedded smart cdn fixture and ensure results

* Rename JAVA_SDK_E2E_SMARTCDN

* refactor: simplify sse listener post-finish handling

* chore: release 2.2.4

Author: kvz

.gitattributes

@@ -0,0 +1 @@
+*.jpg binary

CHANGELOG.md

@@ -1,3 +1,8 @@
+### 2.2.4 / 2025-10-28
+
+- Bundle chameleon.jpg in the repository and enable the `directUploadOriginalStepProducesResult` integration test via `JAVA_SDK_E2E`.
+- Simplify the SSE listener by stopping immediately after `assembly_finished` now that trailing results are guaranteed.
+
 ### 2.2.3 / 2025-10-27
 
 - Extend the SSE listener grace window to 10 seconds so `assembly_result_finished` payloads are not dropped after `assembly_finished`.

scripts/test-in-docker.sh

@@ -71,4 +71,13 @@ if [[ -f .env ]]; then
   DOCKER_ARGS+=(--env-file "$PWD/.env")
 fi
 
+
+EXTRA_ENVS=(JAVA_SDK_E2E TRANSLOADIT_HOST TRANSLOADIT_KEY TRANSLOADIT_SECRET)
+for var in "${EXTRA_ENVS[@]}"; do
+  value="${!var:-}"
+  if [[ -n "$value" ]]; then
+    DOCKER_ARGS+=(-e "$var=$value")
+  fi
+done
+
 exec docker run "${DOCKER_ARGS[@]}" "$IMAGE_NAME" bash -lc "$RUN_CMD"

src/main/java/com/transloadit/sdk/EventsourceRunnable.java

@@ -22,8 +22,6 @@
 import java.util.Iterator;
 
 class EventsourceRunnable implements Runnable {
-    private static final long FINISH_DRAIN_TIMEOUT_MS = 10000L;
-
     protected boolean assemblyFinished;
     protected AssemblyListener assemblyListener;
 
@@ -33,7 +31,6 @@ class EventsourceRunnable implements Runnable {
     protected Transloadit transloadit;
     protected boolean stopRequested;
     protected boolean assemblyFinishedNotified;
-    protected long assemblyFinishedAtMillis;
 
     /**
      * Constructor for {@link EventsourceRunnable}. It creates a new {@link EventSource} instance, wrapped in a
@@ -72,7 +69,6 @@ public void run() {
         this.assemblyFinished = false;
         this.stopRequested = false;
         this.assemblyFinishedNotified = false;
-        this.assemblyFinishedAtMillis = 0L;
         try {
             eventSource.start();
         } catch (StreamException e) {
@@ -100,16 +96,7 @@ public void run() {
 
             if (!processedEvent) {
                 ReadyState state = eventSource.getState();
-                long now = System.currentTimeMillis();
-                if (assemblyFinished) {
-                    if (assemblyFinishedAtMillis == 0L) {
-                        assemblyFinishedAtMillis = now;
-                    }
-                    boolean graceExpired = now - assemblyFinishedAtMillis >= FINISH_DRAIN_TIMEOUT_MS;
-                    if (graceExpired || state == ReadyState.CLOSED || state == ReadyState.SHUTDOWN) {
-                        stopRequested = true;
-                    }
-                } else if (state == ReadyState.CLOSED || state == ReadyState.SHUTDOWN) {
+                if (state == ReadyState.CLOSED || state == ReadyState.SHUTDOWN) {
                     stopRequested = true;
                 }
 
@@ -145,7 +132,6 @@ protected void handleMessageEvent(MessageEvent messageEvent) {
             switch (data) {
                 case "assembly_finished":
                     assemblyFinished = true;
-                    assemblyFinishedAtMillis = System.currentTimeMillis();
                     if (!assemblyFinishedNotified) {
                         assemblyFinishedNotified = true;
                         try {
@@ -154,6 +140,7 @@ protected void handleMessageEvent(MessageEvent messageEvent) {
                             assemblyListener.onError(e);
                         }
                     }
+                    stopRequested = true;
                     break;
                 case "assembly_upload_meta_data_extracted":
                     assemblyListener.onMetadataExtracted();

src/main/resources/java-sdk-version/version.properties

@@ -1 +1 @@
-versionNumber='2.2.3'
+versionNumber='2.2.4'

src/test/java/com/transloadit/sdk/integration/AssemblyIntegrationTest.java

@@ -3,6 +3,9 @@
 import com.transloadit.sdk.Assembly;
 import com.transloadit.sdk.Transloadit;
 import com.transloadit.sdk.response.AssemblyResponse;
+
+import java.io.File;
+import java.io.FileOutputStream;
 import org.json.JSONArray;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Assumptions;
@@ -50,4 +53,77 @@ void createAssemblyAndWaitForCompletion() throws Exception {
         Assertions.assertNotNull(stepResult, "resize step result missing");
         Assertions.assertTrue(stepResult.length() > 0, "resize step result empty");
     }
+
+    @Test
+    void directUploadOriginalStepProducesResult() throws Exception {
+        Assumptions.assumeTrue("1".equals(System.getenv("JAVA_SDK_E2E")), "set JAVA_SDK_E2E=1 to run");
+
+        String key = System.getenv("TRANSLOADIT_KEY");
+        String secret = System.getenv("TRANSLOADIT_SECRET");
+        Assumptions.assumeTrue(key != null && !key.trim().isEmpty(), "TRANSLOADIT_KEY env var required");
+        Assumptions.assumeTrue(secret != null && !secret.trim().isEmpty(), "TRANSLOADIT_SECRET env var required");
+
+        File upload = File.createTempFile("java-sdk-smartcdn", ".bin");
+        try (java.io.InputStream in = AssemblyIntegrationTest.class.getResourceAsStream("/chameleon.jpg");
+             FileOutputStream fos = new FileOutputStream(upload)) {
+            if (in == null) {
+                Assertions.fail("Embedded chameleon.jpg fixture missing");
+            }
+            byte[] buffer = new byte[8192];
+            int read;
+            while ((read = in.read(buffer)) != -1) {
+                fos.write(buffer, 0, read);
+            }
+            long targetSize = 32L * 1024L * 1024L;
+            long current = upload.length();
+            if (current < targetSize) {
+                byte[] padding = new byte[8192];
+                while (current < targetSize) {
+                    long remaining = targetSize - current;
+                    int toWrite = (int) Math.min(padding.length, remaining);
+                    fos.write(padding, 0, toWrite);
+                    current += toWrite;
+                }
+            }
+        }
+
+        Transloadit client = new Transloadit(key, secret);
+        Assembly assembly = client.newAssembly();
+        try {
+            assembly.addFile(upload, "image");
+
+        Map<String, Object> resize = new HashMap<>();
+        resize.put("use", ":original");
+        resize.put("width", 32);
+        resize.put("height", 32);
+        resize.put("resize_strategy", "fit");
+        resize.put("format", "jpg");
+        resize.put("result", true);
+        assembly.addStep("resize", "/image/resize", resize);
+
+        AssemblyResponse response = assembly.save(true);
+        String assemblyId = response.getId();
+
+        long deadline = System.currentTimeMillis() + TimeUnit.MINUTES.toMillis(5);
+        while (!response.isFinished() && System.currentTimeMillis() < deadline) {
+            Thread.sleep(5000);
+            response = client.getAssembly(assemblyId);
+        }
+
+        Assertions.assertTrue(response.isFinished(), "Assembly did not finish in time: " + assemblyId);
+        String ok = response.json().optString("ok");
+        if (!"ASSEMBLY_COMPLETED".equals(ok)) {
+            Assertions.fail("Assembly " + assemblyId + " finished with status=" + ok + " payload=" + response.json());
+        }
+
+        JSONArray resizeResult = response.getStepResult("resize");
+        if (resizeResult == null) {
+            Assertions.fail("resize step result missing for assembly " + assemblyId + " payload=" + response.json());
+        }
+        Assertions.assertTrue(resizeResult.length() > 0, "resize step result empty for assembly " + assemblyId);
+        } finally {
+            upload.delete();
+        }
+    }
+
 }