Skip to content

Commit 24cd34f

Browse files
kvzkvz
authored
Stop reconnecting SSE after assembly_finished (#96)
* Stop SSE reconnect after assembly finishes * bump version * Document 2.2.0 release * Make RequestTest expect dynamic version header
1 parent 3a9bcaf commit 24cd34f

6 files changed

Lines changed: 297 additions & 4 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,9 @@
1+
### 2.2.0 / 2025-10-27
2+
3+
- Prevent the SSE client from reconnecting after `assembly_finished`, eliminating spurious `assembly_error` callbacks and timeouts.
4+
- Add regression coverage with a targeted unit test and a live SSE integration test executed via the Docker harness.
5+
- Confirm the new tests run in CI alongside the existing Gradle `check` workflow.
6+
17
### 2.1.0 / 2025-10-15
28

39
- Added support for external signature generation via `SignatureProvider` interface ([#19](https://github.com/transloadit/android-sdk/issues/19))

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

Lines changed: 31 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,11 @@ protected void handleMessageEvent(MessageEvent messageEvent) {
101101
String eventName = messageEvent.getEventName();
102102
String data = messageEvent.getData();
103103

104+
if (assemblyFinished) {
105+
shutdownEventSource();
106+
return;
107+
}
108+
104109
// Check if the event is a message event without
105110
if (eventName.equals("message")) {
106111
switch (data) {
@@ -110,8 +115,9 @@ protected void handleMessageEvent(MessageEvent messageEvent) {
110115
assemblyListener.onAssemblyFinished(transloadit.getAssemblyByUrl(response.getSslUrl()));
111116
} catch (RequestException | LocalOperationException e) {
112117
assemblyListener.onError(e);
118+
} finally {
119+
shutdownEventSource();
113120
}
114-
this.eventSource.close();
115121
break;
116122
case "assembly_upload_meta_data_extracted":
117123
assemblyListener.onMetadataExtracted();
@@ -139,8 +145,12 @@ protected void handleMessageEvent(MessageEvent messageEvent) {
139145
break;
140146

141147
case "assembly_error":
148+
if (assemblyFinished) {
149+
shutdownEventSource();
150+
break;
151+
}
142152
assemblyListener.onError(new RequestException(data));
143-
this.eventSource.close();
153+
shutdownEventSource();
144154
break;
145155

146156
case "assembly_execution_progress":
@@ -167,10 +177,29 @@ protected void handleStartedEvent(StartedEvent startedEvent) {
167177
}
168178

169179
protected void handleFaultEvent(FaultEvent faultEvent) {
180+
if (assemblyFinished) {
181+
shutdownEventSource();
182+
}
170183
// Debug output, uncomment if needed
171184
// String data = faultEvent.toString();
172185
// System.out.printf("Fault: %s\n", data);
173186
// System.out.println("Starting Over");
174187
}
175188

189+
private void shutdownEventSource() {
190+
if (this.eventSource == null) {
191+
return;
192+
}
193+
try {
194+
this.eventSource.stop();
195+
} catch (Exception ignore) {
196+
// Ignore cleanup exceptions
197+
}
198+
try {
199+
this.eventSource.close();
200+
} catch (Exception ignore) {
201+
// Ignore cleanup exceptions
202+
}
203+
}
204+
176205
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
versionNumber='2.1.0'
1+
versionNumber='2.2.0'

‎src/test/java/com/transloadit/sdk/EventsourceRunnableTest.java‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
package com.transloadit.sdk;
2+
3+
import com.launchdarkly.eventsource.ConnectStrategy;
4+
import com.launchdarkly.eventsource.ErrorStrategy;
5+
import com.launchdarkly.eventsource.MessageEvent;
6+
import com.launchdarkly.eventsource.RetryDelayStrategy;
7+
import com.transloadit.sdk.response.AssemblyResponse;
8+
import org.json.JSONArray;
9+
import org.json.JSONObject;
10+
import org.junit.jupiter.api.Test;
11+
12+
import java.net.URI;
13+
import java.util.concurrent.atomic.AtomicBoolean;
14+
import java.util.concurrent.atomic.AtomicReference;
15+
16+
import static org.junit.jupiter.api.Assertions.*;
17+
import static org.mockito.ArgumentMatchers.anyString;
18+
import static org.mockito.Mockito.mock;
19+
import static org.mockito.Mockito.when;
20+
21+
class EventsourceRunnableTest {
22+
23+
@Test
24+
void assemblyErrorAfterFinishedDoesNotNotifyListener() throws Exception {
25+
Transloadit transloadit = mock(Transloadit.class);
26+
AssemblyResponse initialResponse = mock(AssemblyResponse.class);
27+
AssemblyResponse finalResponse = mock(AssemblyResponse.class);
28+
29+
when(initialResponse.getSslUrl()).thenReturn("https://example.com/assemblies/123");
30+
when(transloadit.getAssemblyByUrl(anyString())).thenReturn(finalResponse);
31+
when(finalResponse.json()).thenReturn(new JSONObject().put("ok", "ASSEMBLY_COMPLETED"));
32+
33+
ConnectStrategy connectStrategy = ConnectStrategy.http(URI.create("http://localhost/sse"));
34+
RetryDelayStrategy retryStrategy = RetryDelayStrategy.defaultStrategy();
35+
ErrorStrategy errorStrategy = ErrorStrategy.alwaysContinue();
36+
37+
RecordingListener listener = new RecordingListener();
38+
39+
EventsourceRunnable runnable = new EventsourceRunnable(
40+
transloadit,
41+
initialResponse,
42+
listener,
43+
connectStrategy,
44+
retryStrategy,
45+
errorStrategy,
46+
false
47+
);
48+
49+
MessageEvent finishedEvent = new MessageEvent("assembly_finished");
50+
MessageEvent errorEvent = new MessageEvent("assembly_error", "{}", null, null);
51+
runnable.handleMessageEvent(finishedEvent);
52+
runnable.handleMessageEvent(errorEvent);
53+
54+
assertTrue(listener.finishedCalled.get(), "Expected assembly_finished to notify listener");
55+
assertFalse(listener.errorCalled.get(), "Unexpected error callback after completion");
56+
assertNotNull(listener.finishedResponse.get(), "Final response missing");
57+
assertEquals(finalResponse, listener.finishedResponse.get());
58+
}
59+
60+
private static final class RecordingListener implements AssemblyListener {
61+
private final AtomicBoolean finishedCalled = new AtomicBoolean(false);
62+
private final AtomicBoolean errorCalled = new AtomicBoolean(false);
63+
private final AtomicReference<AssemblyResponse> finishedResponse = new AtomicReference<>();
64+
65+
@Override
66+
public void onAssemblyFinished(AssemblyResponse response) {
67+
finishedCalled.set(true);
68+
finishedResponse.set(response);
69+
}
70+
71+
@Override
72+
public void onError(Exception error) {
73+
errorCalled.set(true);
74+
}
75+
76+
@Override
77+
public void onMetadataExtracted() {
78+
}
79+
80+
@Override
81+
public void onAssemblyUploadFinished() {
82+
}
83+
84+
@Override
85+
public void onFileUploadFinished(JSONObject uploadInformation) {
86+
}
87+
88+
@Override
89+
public void onFileUploadPaused(String name) {
90+
}
91+
92+
@Override
93+
public void onFileUploadResumed(String name) {
94+
}
95+
96+
@Override
97+
public void onFileUploadProgress(long uploadedBytes, long totalBytes) {
98+
}
99+
100+
@Override
101+
public void onAssemblyProgress(JSONObject progress) {
102+
}
103+
104+
@Override
105+
public void onAssemblyResultFinished(JSONArray result) {
106+
}
107+
}
108+
}

‎src/test/java/com/transloadit/sdk/RequestTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -130,10 +130,11 @@ private String extractMultipartField(String body, String fieldName) {
130130
public void get() throws Exception {
131131
request.get("/foo");
132132

133+
String expectedClientHeader = transloadit.getVersionInfo();
133134
mockServerClient.verify(HttpRequest.request()
134135
.withPath("/foo")
135136
.withMethod("GET")
136-
.withHeader("Transloadit-Client", "java-sdk:2.1.0"));
137+
.withHeader("Transloadit-Client", expectedClientHeader));
137138

138139
}
139140

‎src/test/java/com/transloadit/sdk/integration/AssemblySseIntegrationTest.java‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
package com.transloadit.sdk.integration;
2+
3+
import com.transloadit.sdk.Assembly;
4+
import com.transloadit.sdk.AssemblyListener;
5+
import com.transloadit.sdk.Transloadit;
6+
import com.transloadit.sdk.response.AssemblyResponse;
7+
import org.json.JSONArray;
8+
import org.json.JSONObject;
9+
import org.junit.jupiter.api.Assumptions;
10+
import org.junit.jupiter.api.Test;
11+
12+
import java.io.IOException;
13+
import java.io.InputStream;
14+
import java.io.OutputStream;
15+
import java.net.URL;
16+
import java.nio.file.Files;
17+
import java.nio.file.Path;
18+
import java.util.HashMap;
19+
import java.util.Map;
20+
import java.util.concurrent.CompletableFuture;
21+
import java.util.concurrent.ExecutionException;
22+
import java.util.concurrent.TimeUnit;
23+
import java.util.concurrent.TimeoutException;
24+
import java.util.concurrent.atomic.AtomicReference;
25+
26+
import static org.junit.jupiter.api.Assertions.*;
27+
28+
class AssemblySseIntegrationTest {
29+
30+
@Test
31+
void sseStreamShouldCloseWithoutErrorsAfterAssemblyFinished() throws Exception {
32+
String key = System.getenv("TRANSLOADIT_KEY");
33+
String secret = System.getenv("TRANSLOADIT_SECRET");
34+
Assumptions.assumeTrue(key != null && !key.trim().isEmpty(), "TRANSLOADIT_KEY env var required");
35+
Assumptions.assumeTrue(secret != null && !secret.trim().isEmpty(), "TRANSLOADIT_SECRET env var required");
36+
37+
Transloadit client = new Transloadit(key, secret);
38+
Assembly assembly = client.newAssembly();
39+
40+
Path tempFile = createTempUpload();
41+
try {
42+
assembly.addFile(tempFile.toFile(), "file");
43+
44+
Map<String, Object> resizeStep = new HashMap<>();
45+
resizeStep.put("use", ":original");
46+
resizeStep.put("width", 64);
47+
resizeStep.put("height", 64);
48+
resizeStep.put("resize_strategy", "fit");
49+
assembly.addStep("resize", "/image/resize", resizeStep);
50+
51+
AtomicReference<AssemblyResponse> finishedResponse = new AtomicReference<>();
52+
CompletableFuture<Void> finishedFuture = new CompletableFuture<>();
53+
CompletableFuture<Exception> errorFuture = new CompletableFuture<>();
54+
55+
AssemblyListener listener = new AssemblyListener() {
56+
@Override
57+
public void onAssemblyFinished(AssemblyResponse response) {
58+
System.out.println("[AssemblySseIntegrationTest] assembly_finished event");
59+
finishedResponse.set(response);
60+
finishedFuture.complete(null);
61+
}
62+
63+
@Override
64+
public void onError(Exception error) {
65+
System.out.println("[AssemblySseIntegrationTest] SSE error: " + error);
66+
errorFuture.complete(error);
67+
finishedFuture.completeExceptionally(error);
68+
}
69+
70+
@Override
71+
public void onMetadataExtracted() {
72+
}
73+
74+
@Override
75+
public void onAssemblyUploadFinished() {
76+
}
77+
78+
@Override
79+
public void onFileUploadFinished(JSONObject uploadInformation) {
80+
}
81+
82+
@Override
83+
public void onFileUploadPaused(String name) {
84+
}
85+
86+
@Override
87+
public void onFileUploadResumed(String name) {
88+
}
89+
90+
@Override
91+
public void onFileUploadProgress(long uploadedBytes, long totalBytes) {
92+
}
93+
94+
@Override
95+
public void onAssemblyProgress(JSONObject progress) {
96+
}
97+
98+
@Override
99+
public void onAssemblyResultFinished(JSONArray result) {
100+
}
101+
};
102+
103+
assembly.setAssemblyListener(listener);
104+
105+
AssemblyResponse initialResponse = assembly.save(true);
106+
assertNotNull(initialResponse.getId(), "Assembly ID should be present");
107+
108+
try {
109+
finishedFuture.get(5, TimeUnit.MINUTES);
110+
} catch (ExecutionException executionException) {
111+
Throwable cause = executionException.getCause();
112+
if (cause instanceof Exception) {
113+
throw (Exception) cause;
114+
}
115+
throw executionException;
116+
}
117+
118+
AssemblyResponse completed = finishedResponse.get();
119+
assertNotNull(completed, "Assembly finished response missing");
120+
assertTrue(completed.isFinished(), "Assembly should be finished");
121+
assertEquals("ASSEMBLY_COMPLETED", completed.json().optString("ok"));
122+
123+
try {
124+
Exception unexpected = errorFuture.get(30, TimeUnit.SECONDS);
125+
fail("Unexpected SSE error after completion: " + unexpected);
126+
} catch (TimeoutException ignore) {
127+
// expected: no error surfaced after assembly finished
128+
}
129+
} finally {
130+
try {
131+
Files.deleteIfExists(tempFile);
132+
} catch (IOException ignore) {
133+
}
134+
}
135+
}
136+
137+
private static Path createTempUpload() throws IOException {
138+
Path file = Files.createTempFile("transloadit-sse-test", ".jpg");
139+
URL source = new URL("https://demos.transloadit.com/inputs/chameleon.jpg");
140+
try (InputStream input = source.openStream(); OutputStream output = Files.newOutputStream(file)) {
141+
byte[] buffer = new byte[8192];
142+
int read;
143+
while ((read = input.read(buffer)) != -1) {
144+
output.write(buffer, 0, read);
145+
}
146+
}
147+
return file;
148+
}
149+
}

0 commit comments

Comments
 (0)