Skip to content

Commit cb4749d

Browse files
saurabhjain1592saurabhjain1592
committed
feat: add streamExecutionStatus SSE client for real-time execution monitoring
Adds streamExecutionStatus() method that connects to the SSE streaming endpoint and invokes Consumer callbacks with ExecutionStatus updates. Also adds Jackson annotations to ExecutionTypes for proper snake_case JSON deserialization.
1 parent 7d1a4ab commit cb4749d

3 files changed

Lines changed: 344 additions & 12 deletions

File tree

‎src/main/java/com/getaxonflow/sdk/AxonFlow.java‎

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,8 +35,10 @@
3535
import org.slf4j.Logger;
3636
import org.slf4j.LoggerFactory;
3737

38+
import java.io.BufferedReader;
3839
import java.io.Closeable;
3940
import java.io.IOException;
41+
import java.io.InputStreamReader;
4042
import java.net.URI;
4143
import java.nio.charset.StandardCharsets;
4244
import java.util.ArrayList;
@@ -49,6 +51,7 @@
4951
import java.util.concurrent.CompletableFuture;
5052
import java.util.concurrent.Executor;
5153
import java.util.concurrent.ForkJoinPool;
54+
import java.util.function.Consumer;
5255

5356
/**
5457
* Main client for interacting with the AxonFlow API.
@@ -1963,6 +1966,117 @@ public void cancelExecution(String executionId) {
19631966
cancelExecution(executionId, null);
19641967
}
19651968

1969+
/**
1970+
* Streams real-time execution status updates via Server-Sent Events (SSE).
1971+
*
1972+
* <p>Connects to the SSE streaming endpoint and invokes the callback with each
1973+
* {@link com.getaxonflow.sdk.types.execution.ExecutionTypes.ExecutionStatus} update
1974+
* as it arrives. The stream automatically closes when the execution reaches a
1975+
* terminal state (completed, failed, cancelled, aborted, or expired).
1976+
*
1977+
* <p>Example usage:
1978+
* <pre>{@code
1979+
* axonflow.streamExecutionStatus("exec_123", status -> {
1980+
* System.out.printf("Progress: %.0f%% - Status: %s%n",
1981+
* status.getProgressPercent(), status.getStatus().getValue());
1982+
* if (status.getCurrentStep() != null) {
1983+
* System.out.println(" Current step: " + status.getCurrentStep().getStepName());
1984+
* }
1985+
* });
1986+
* }</pre>
1987+
*
1988+
* @param executionId the execution ID (plan ID or workflow ID)
1989+
* @param callback consumer invoked with each ExecutionStatus update
1990+
* @throws AxonFlowException if the connection fails or an I/O error occurs
1991+
* @throws AuthenticationException if authentication fails (401/403)
1992+
*/
1993+
public void streamExecutionStatus(
1994+
String executionId,
1995+
Consumer<com.getaxonflow.sdk.types.execution.ExecutionTypes.ExecutionStatus> callback) {
1996+
Objects.requireNonNull(executionId, "executionId cannot be null");
1997+
Objects.requireNonNull(callback, "callback cannot be null");
1998+
1999+
logger.debug("Streaming execution status for {}", executionId);
2000+
2001+
HttpUrl url = HttpUrl.parse(config.getEndpoint() + "/api/v1/executions/" + executionId + "/stream");
2002+
if (url == null) {
2003+
throw new ConfigurationException("Invalid URL: " + config.getEndpoint()
2004+
+ "/api/v1/executions/" + executionId + "/stream");
2005+
}
2006+
2007+
Request.Builder builder = new Request.Builder()
2008+
.url(url)
2009+
.header("User-Agent", config.getUserAgent())
2010+
.header("Accept", "text/event-stream")
2011+
.get();
2012+
2013+
addAuthHeaders(builder);
2014+
addTenantIdHeader(builder);
2015+
2016+
Request httpRequest = builder.build();
2017+
2018+
try {
2019+
Response response = httpClient.newCall(httpRequest).execute();
2020+
try {
2021+
if (!response.isSuccessful()) {
2022+
handleErrorResponse(response);
2023+
}
2024+
2025+
ResponseBody body = response.body();
2026+
if (body == null) {
2027+
throw new AxonFlowException("SSE response has no body", 0, null);
2028+
}
2029+
2030+
try (BufferedReader reader = new BufferedReader(
2031+
new InputStreamReader(body.byteStream(), StandardCharsets.UTF_8))) {
2032+
StringBuilder eventBuffer = new StringBuilder();
2033+
String line;
2034+
2035+
while ((line = reader.readLine()) != null) {
2036+
if (line.isEmpty()) {
2037+
// Empty line = end of SSE event
2038+
String event = eventBuffer.toString().trim();
2039+
eventBuffer.setLength(0);
2040+
2041+
if (event.isEmpty()) {
2042+
continue;
2043+
}
2044+
2045+
// Parse SSE data lines
2046+
for (String eventLine : event.split("\n")) {
2047+
if (eventLine.startsWith("data: ")) {
2048+
String jsonStr = eventLine.substring(6);
2049+
if (jsonStr.isEmpty() || "[DONE]".equals(jsonStr)) {
2050+
continue;
2051+
}
2052+
try {
2053+
com.getaxonflow.sdk.types.execution.ExecutionTypes.ExecutionStatus status =
2054+
objectMapper.readValue(jsonStr,
2055+
com.getaxonflow.sdk.types.execution.ExecutionTypes.ExecutionStatus.class);
2056+
callback.accept(status);
2057+
2058+
// Check for terminal status
2059+
if (status.getStatus() != null && status.getStatus().isTerminal()) {
2060+
return;
2061+
}
2062+
} catch (JsonProcessingException e) {
2063+
logger.warn("Failed to parse SSE data: {}", jsonStr, e);
2064+
}
2065+
}
2066+
}
2067+
} else {
2068+
eventBuffer.append(line).append("\n");
2069+
}
2070+
}
2071+
}
2072+
} finally {
2073+
response.close();
2074+
}
2075+
} catch (IOException e) {
2076+
throw new AxonFlowException("SSE stream failed: " + e.getMessage(), e);
2077+
}
2078+
}
2079+
19662080
// ========================================================================
19672081
// Configuration Access
19682082
// ========================================================================

‎src/main/java/com/getaxonflow/sdk/types/execution/ExecutionTypes.java‎

Lines changed: 61 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@
1616

1717
package com.getaxonflow.sdk.types.execution;
1818

19+
import com.fasterxml.jackson.annotation.JsonCreator;
20+
import com.fasterxml.jackson.annotation.JsonProperty;
21+
import com.fasterxml.jackson.annotation.JsonValue;
22+
1923
import java.time.Instant;
2024
import java.util.List;
2125
import java.util.Map;
@@ -49,10 +53,12 @@ public enum ExecutionType {
4953
this.value = value;
5054
}
5155

56+
@JsonValue
5257
public String getValue() {
5358
return value;
5459
}
5560

61+
@JsonCreator
5662
public static ExecutionType fromValue(String value) {
5763
for (ExecutionType type : values()) {
5864
if (type.value.equals(value)) {
@@ -81,6 +87,7 @@ public enum ExecutionStatusValue {
8187
this.value = value;
8288
}
8389

90+
@JsonValue
8491
public String getValue() {
8592
return value;
8693
}
@@ -90,6 +97,7 @@ public boolean isTerminal() {
9097
this == ABORTED || this == EXPIRED;
9198
}
9299

100+
@JsonCreator
93101
public static ExecutionStatusValue fromValue(String value) {
94102
for (ExecutionStatusValue status : values()) {
95103
if (status.value.equals(value)) {
@@ -118,6 +126,7 @@ public enum StepStatusValue {
118126
this.value = value;
119127
}
120128

129+
@JsonValue
121130
public String getValue() {
122131
return value;
123132
}
@@ -130,6 +139,7 @@ public boolean isBlocking() {
130139
return this == BLOCKED || this == APPROVAL;
131140
}
132141

142+
@JsonCreator
133143
public static StepStatusValue fromValue(String value) {
134144
for (StepStatusValue status : values()) {
135145
if (status.value.equals(value)) {
@@ -158,10 +168,12 @@ public enum UnifiedStepType {
158168
this.value = value;
159169
}
160170

171+
@JsonValue
161172
public String getValue() {
162173
return value;
163174
}
164175

176+
@JsonCreator
165177
public static UnifiedStepType fromValue(String value) {
166178
for (UnifiedStepType type : values()) {
167179
if (type.value.equals(value)) {
@@ -186,10 +198,12 @@ public enum UnifiedGateDecision {
186198
this.value = value;
187199
}
188200

201+
@JsonValue
189202
public String getValue() {
190203
return value;
191204
}
192205

206+
@JsonCreator
193207
public static UnifiedGateDecision fromValue(String value) {
194208
for (UnifiedGateDecision decision : values()) {
195209
if (decision.value.equals(value)) {
@@ -214,10 +228,12 @@ public enum UnifiedApprovalStatus {
214228
this.value = value;
215229
}
216230

231+
@JsonValue
217232
public String getValue() {
218233
return value;
219234
}
220235

236+
@JsonCreator
221237
public static UnifiedApprovalStatus fromValue(String value) {
222238
for (UnifiedApprovalStatus status : values()) {
223239
if (status.value.equals(value)) {
@@ -254,13 +270,29 @@ public static final class UnifiedStepStatus {
254270
private final String resultSummary;
255271
private final String error;
256272

273+
@JsonCreator
257274
public UnifiedStepStatus(
258-
String stepId, int stepIndex, String stepName, UnifiedStepType stepType,
259-
StepStatusValue status, Instant startedAt, Instant endedAt, String duration,
260-
UnifiedGateDecision decision, String decisionReason, List<String> policiesMatched,
261-
UnifiedApprovalStatus approvalStatus, String approvedBy, Instant approvedAt,
262-
String model, String provider, Double costUsd, Object input, Object output,
263-
String resultSummary, String error) {
275+
@JsonProperty("step_id") String stepId,
276+
@JsonProperty("step_index") int stepIndex,
277+
@JsonProperty("step_name") String stepName,
278+
@JsonProperty("step_type") UnifiedStepType stepType,
279+
@JsonProperty("status") StepStatusValue status,
280+
@JsonProperty("started_at") Instant startedAt,
281+
@JsonProperty("ended_at") Instant endedAt,
282+
@JsonProperty("duration") String duration,
283+
@JsonProperty("decision") UnifiedGateDecision decision,
284+
@JsonProperty("decision_reason") String decisionReason,
285+
@JsonProperty("policies_matched") List<String> policiesMatched,
286+
@JsonProperty("approval_status") UnifiedApprovalStatus approvalStatus,
287+
@JsonProperty("approved_by") String approvedBy,
288+
@JsonProperty("approved_at") Instant approvedAt,
289+
@JsonProperty("model") String model,
290+
@JsonProperty("provider") String provider,
291+
@JsonProperty("cost_usd") Double costUsd,
292+
@JsonProperty("input") Object input,
293+
@JsonProperty("output") Object output,
294+
@JsonProperty("result_summary") String resultSummary,
295+
@JsonProperty("error") String error) {
264296
this.stepId = stepId;
265297
this.stepIndex = stepIndex;
266298
this.stepName = stepName;
@@ -406,13 +438,30 @@ public static final class ExecutionStatus {
406438
private final Instant createdAt;
407439
private final Instant updatedAt;
408440

441+
@JsonCreator
409442
public ExecutionStatus(
410-
String executionId, ExecutionType executionType, String name, String source,
411-
ExecutionStatusValue status, int currentStepIndex, int totalSteps,
412-
double progressPercent, Instant startedAt, Instant completedAt, String duration,
413-
Double estimatedCostUsd, Double actualCostUsd, List<UnifiedStepStatus> steps,
414-
String error, String tenantId, String orgId, String userId, String clientId,
415-
Map<String, Object> metadata, Instant createdAt, Instant updatedAt) {
443+
@JsonProperty("execution_id") String executionId,
444+
@JsonProperty("execution_type") ExecutionType executionType,
445+
@JsonProperty("name") String name,
446+
@JsonProperty("source") String source,
447+
@JsonProperty("status") ExecutionStatusValue status,
448+
@JsonProperty("current_step_index") int currentStepIndex,
449+
@JsonProperty("total_steps") int totalSteps,
450+
@JsonProperty("progress_percent") double progressPercent,
451+
@JsonProperty("started_at") Instant startedAt,
452+
@JsonProperty("completed_at") Instant completedAt,
453+
@JsonProperty("duration") String duration,
454+
@JsonProperty("estimated_cost_usd") Double estimatedCostUsd,
455+
@JsonProperty("actual_cost_usd") Double actualCostUsd,
456+
@JsonProperty("steps") List<UnifiedStepStatus> steps,
457+
@JsonProperty("error") String error,
458+
@JsonProperty("tenant_id") String tenantId,
459+
@JsonProperty("org_id") String orgId,
460+
@JsonProperty("user_id") String userId,
461+
@JsonProperty("client_id") String clientId,
462+
@JsonProperty("metadata") Map<String, Object> metadata,
463+
@JsonProperty("created_at") Instant createdAt,
464+
@JsonProperty("updated_at") Instant updatedAt) {
416465
this.executionId = executionId;
417466
this.executionType = executionType;
418467
this.name = name;

0 commit comments

Comments
 (0)