diff --git a/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java b/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java index 58681cf44..e05b48d51 100644 --- a/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java +++ b/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java @@ -25,6 +25,7 @@ import io.a2a.spec.GetTaskPushNotificationConfigParams; import io.a2a.spec.Message; import io.a2a.spec.PushNotificationConfig; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; import io.a2a.spec.TaskPushNotificationConfig; import io.a2a.spec.TextPart; @@ -88,9 +89,9 @@ public void testDirectNotificationTrigger() { mockPushNotificationSender.sendNotification(testTask); // Verify it was captured - Queue captured = mockPushNotificationSender.getCapturedTasks(); + Queue captured = mockPushNotificationSender.getCapturedEvents(); assertEquals(1, captured.size()); - assertEquals("direct-test-task", captured.peek().getId()); + assertEquals("direct-test-task", ((Task)captured.peek()).getId()); } @Test @@ -151,7 +152,7 @@ public void testJpaDatabasePushNotificationConfigStoreIntegration() throws Excep boolean notificationReceived = false; while (System.currentTimeMillis() < end) { - if (!mockPushNotificationSender.getCapturedTasks().isEmpty()) { + if (!mockPushNotificationSender.getCapturedEvents().isEmpty()) { notificationReceived = true; break; } @@ -161,10 +162,12 @@ public void testJpaDatabasePushNotificationConfigStoreIntegration() throws Excep assertTrue(notificationReceived, "Timeout waiting for push notification."); // Step 6: Verify the captured notification - Queue capturedTasks = mockPushNotificationSender.getCapturedTasks(); + Queue capturedTasks = mockPushNotificationSender.getCapturedEvents(); // Verify the notification contains the correct task with artifacts Task notifiedTaskWithArtifact = capturedTasks.stream() + .filter(e -> Task.TASK.equals(e.getKind())) + .map(e -> (Task)e) .filter(t -> taskId.equals(t.getId()) && t.getArtifacts() != null && t.getArtifacts().size() > 0) .findFirst() .orElse(null); diff --git a/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java b/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java index 0a6bba415..c957f419b 100644 --- a/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java +++ b/extras/push-notification-config-store-database-jpa/src/test/java/io/a2a/extras/pushnotificationconfigstore/database/jpa/MockPushNotificationSender.java @@ -8,7 +8,7 @@ import jakarta.enterprise.inject.Alternative; import io.a2a.server.tasks.PushNotificationSender; -import io.a2a.spec.Task; +import io.a2a.spec.StreamingEventKind; /** * Mock implementation of PushNotificationSender for integration testing. @@ -19,18 +19,18 @@ @Priority(100) public class MockPushNotificationSender implements PushNotificationSender { - private final Queue capturedTasks = new ConcurrentLinkedQueue<>(); + private final Queue capturedEvents = new ConcurrentLinkedQueue<>(); @Override - public void sendNotification(Task task) { - capturedTasks.add(task); + public void sendNotification(StreamingEventKind kind) { + capturedEvents.add(kind); } - public Queue getCapturedTasks() { - return capturedTasks; + public Queue getCapturedEvents() { + return capturedEvents; } public void clear() { - capturedTasks.clear(); + capturedEvents.clear(); } } diff --git a/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java b/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java index e400b7ba0..343a0a7d2 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java +++ b/server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java @@ -3,25 +3,33 @@ import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON; import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE; import static io.a2a.common.A2AHeaders.X_A2A_NOTIFICATION_TOKEN; -import jakarta.enterprise.context.ApplicationScoped; -import jakarta.inject.Inject; +import static io.a2a.spec.Message.MESSAGE; +import static io.a2a.spec.Task.TASK; +import static io.a2a.spec.TaskArtifactUpdateEvent.ARTIFACT_UPDATE; +import static io.a2a.spec.TaskStatusUpdateEvent.STATUS_UPDATE; import java.io.IOException; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; + import com.fasterxml.jackson.core.JsonProcessingException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.JdkA2AHttpClient; +import io.a2a.spec.Message; import io.a2a.spec.PushNotificationConfig; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; +import io.a2a.spec.TaskArtifactUpdateEvent; +import io.a2a.spec.TaskStatusUpdateEvent; import io.a2a.util.Utils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - @ApplicationScoped public class BasePushNotificationSender implements PushNotificationSender { @@ -42,15 +50,25 @@ public BasePushNotificationSender(PushNotificationConfigStore configStore, A2AHt } @Override - public void sendNotification(Task task) { - List pushConfigs = configStore.getInfo(task.getId()); + public void sendNotification(StreamingEventKind kind) { + String taskId = switch (kind.getKind()) { + case TASK -> ((Task)kind).getId(); + case MESSAGE -> ((Message)kind).getTaskId(); + case STATUS_UPDATE -> ((TaskStatusUpdateEvent)kind).getTaskId(); + case ARTIFACT_UPDATE -> ((TaskArtifactUpdateEvent)kind).getTaskId(); + default -> null; + }; + if (taskId == null) { + return; + } + List pushConfigs = configStore.getInfo(taskId); if (pushConfigs == null || pushConfigs.isEmpty()) { return; } List> dispatchResults = pushConfigs .stream() - .map(pushConfig -> dispatch(task, pushConfig)) + .map(pushConfig -> dispatch(kind, pushConfig)) .toList(); CompletableFuture allFutures = CompletableFuture.allOf(dispatchResults.toArray(new CompletableFuture[0])); CompletableFuture dispatchResult = allFutures.thenApply(v -> dispatchResults.stream() @@ -58,18 +76,18 @@ public void sendNotification(Task task) { try { boolean allSent = dispatchResult.get(); if (! allSent) { - LOGGER.warn("Some push notifications failed to send for taskId: " + task.getId()); + LOGGER.warn("Some push notifications failed to send for taskId: " + taskId); } } catch (InterruptedException | ExecutionException e) { - LOGGER.warn("Some push notifications failed to send for taskId " + task.getId() + ": {}", e.getMessage(), e); + LOGGER.warn("Some push notifications failed to send for taskId " + taskId + ": {}", e.getMessage(), e); } } - private CompletableFuture dispatch(Task task, PushNotificationConfig pushInfo) { - return CompletableFuture.supplyAsync(() -> dispatchNotification(task, pushInfo)); + private CompletableFuture dispatch(StreamingEventKind kind, PushNotificationConfig pushInfo) { + return CompletableFuture.supplyAsync(() -> dispatchNotification(kind, pushInfo)); } - private boolean dispatchNotification(Task task, PushNotificationConfig pushInfo) { + private boolean dispatchNotification(StreamingEventKind kind, PushNotificationConfig pushInfo) { String url = pushInfo.url(); String token = pushInfo.token(); @@ -80,7 +98,7 @@ private boolean dispatchNotification(Task task, PushNotificationConfig pushInfo) String body; try { - body = Utils.OBJECT_MAPPER.writeValueAsString(task); + body = Utils.marshalFrom(kind); } catch (JsonProcessingException e) { LOGGER.debug("Error writing value as string: {}", e.getMessage(), e); return false; diff --git a/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java b/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java index 81d577f46..74a5ac259 100644 --- a/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java +++ b/server-common/src/main/java/io/a2a/server/tasks/PushNotificationSender.java @@ -1,6 +1,6 @@ package io.a2a.server.tasks; -import io.a2a.spec.Task; +import io.a2a.spec.StreamingEventKind; /** * Interface for sending push notifications for tasks. @@ -8,8 +8,8 @@ public interface PushNotificationSender { /** - * Sends a push notification containing the latest task state. - * @param task the task + * Sends a push notification containing payload about a task. + * @param kind the payload to push */ - void sendNotification(Task task); + void sendNotification(StreamingEventKind kind); } diff --git a/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java b/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java index d654a83a6..50bd0fac6 100644 --- a/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java +++ b/server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Properties; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; @@ -15,6 +16,11 @@ import jakarta.enterprise.context.Dependent; +import io.quarkus.arc.profile.IfBuildProfile; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; + import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.A2AHttpResponse; import io.a2a.server.agentexecution.AgentExecutor; @@ -30,20 +36,15 @@ import io.a2a.server.tasks.TaskStore; import io.a2a.spec.AgentCapabilities; import io.a2a.spec.AgentCard; +import io.a2a.spec.Event; import io.a2a.spec.JSONRPCError; import io.a2a.spec.Message; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; import io.a2a.spec.TaskState; import io.a2a.spec.TaskStatus; -import io.a2a.spec.Event; import io.a2a.spec.TextPart; import io.a2a.util.Utils; -import io.quarkus.arc.profile.IfBuildProfile; -import java.util.Map; - -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeEach; public class AbstractA2ARequestHandlerTest { @@ -199,7 +200,10 @@ public PostBuilder body(String body) { @Override public A2AHttpResponse post() throws IOException, InterruptedException { - tasks.add(Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE)); + StreamingEventKind kind = Utils.unmarshalStreamingEventKindFrom(body); + if (kind instanceof Task task) { + tasks.add(task); + } try { return new A2AHttpResponse() { @Override diff --git a/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java b/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java index c52a4cf20..8bbd6991c 100644 --- a/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java +++ b/server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java @@ -2,6 +2,11 @@ import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON; import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE; +import static io.a2a.spec.Message.MESSAGE; +import static io.a2a.spec.Task.TASK; +import static io.a2a.spec.TaskArtifactUpdateEvent.ARTIFACT_UPDATE; +import static io.a2a.spec.TaskState.WORKING; +import static io.a2a.spec.TaskStatusUpdateEvent.STATUS_UPDATE; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -22,11 +27,18 @@ import io.a2a.client.http.A2AHttpClient; import io.a2a.client.http.A2AHttpResponse; import io.a2a.common.A2AHeaders; -import io.a2a.util.Utils; +import io.a2a.spec.Artifact; +import io.a2a.spec.Message; import io.a2a.spec.PushNotificationConfig; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; +import io.a2a.spec.TaskArtifactUpdateEvent; import io.a2a.spec.TaskState; import io.a2a.spec.TaskStatus; +import io.a2a.spec.TaskStatusUpdateEvent; +import io.a2a.spec.TextPart; +import io.a2a.spec.UpdateEvent; +import io.a2a.util.Utils; public class PushNotificationSenderTest { @@ -39,6 +51,9 @@ public class PushNotificationSenderTest { */ private static class TestHttpClient implements A2AHttpClient { final List tasks = Collections.synchronizedList(new ArrayList<>()); + final List messages = Collections.synchronizedList(new ArrayList<>()); + final List statusUpdateEvents = Collections.synchronizedList(new ArrayList<>()); + final List artifactUpdateEvents = Collections.synchronizedList(new ArrayList<>()); final List urls = Collections.synchronizedList(new ArrayList<>()); final List> headers = Collections.synchronizedList(new ArrayList<>()); volatile CountDownLatch latch; @@ -77,8 +92,21 @@ public A2AHttpResponse post() throws IOException, InterruptedException { } try { - Task task = Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE); - tasks.add(task); + StreamingEventKind kind = Utils.unmarshalStreamingEventKindFrom(body); + switch (kind.getKind()) { + case TASK: + tasks.add((Task) kind); + break; + case MESSAGE: + messages.add((Message) kind); + break; + case STATUS_UPDATE: + statusUpdateEvents.add((TaskStatusUpdateEvent)kind); + break; + case ARTIFACT_UPDATE: + artifactUpdateEvents.add((TaskArtifactUpdateEvent) kind); + break; + } urls.add(url); headers.add(new java.util.HashMap<>(requestHeaders)); @@ -316,4 +344,139 @@ public void testSendNotificationHttpError() { // Verify no tasks were successfully processed due to the error assertEquals(0, testHttpClient.tasks.size()); } + + @Test + public void testSendNotificationWithMessage() throws InterruptedException { + String taskId = "task_send_notification_with_message"; + String messageId = taskId + "msg-0"; + PushNotificationConfig config = createSamplePushConfig("http://notify.me/here", "cfg1", "unique_token"); + + // Set up the configuration in the store + configStore.setInfo(taskId, config); + + // Set up latch to wait for async completion + testHttpClient.latch = new CountDownLatch(1); + + Message message = new Message.Builder() + .taskId(taskId) + .messageId(messageId) + .role(Message.Role.USER) + .parts(List.of(new TextPart("This is a message sent as a push notification"))) + .build(); + + sender.sendNotification(message); + + // Wait for the async operation to complete + assertTrue(testHttpClient.latch.await(5, TimeUnit.SECONDS), "HTTP call should complete within 5 seconds"); + + // Verify the message was sent via HTTP + assertEquals(0, testHttpClient.tasks.size()); + assertEquals(1, testHttpClient.messages.size()); + + Message sentMessage = testHttpClient.messages.get(0); + assertEquals(taskId, sentMessage.getTaskId()); + assertEquals(messageId, sentMessage.getMessageId()); + + // Verify that the X-A2A-Notification-Token header is sent with the correct token + assertEquals(1, testHttpClient.headers.size()); + Map sentHeaders = testHttpClient.headers.get(0); + assertEquals(2, sentHeaders.size()); + assertTrue(sentHeaders.containsKey(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + assertEquals(config.token(), sentHeaders.get(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + // Content-Type header should always be present + assertTrue(sentHeaders.containsKey(CONTENT_TYPE)); + assertEquals(APPLICATION_JSON, sentHeaders.get(CONTENT_TYPE)); + } + + @Test + public void testSendNotificationWithStatusUpdateEvent() throws InterruptedException { + String taskId = "task_send_notification_with_status_update_event"; + String contextId = taskId + "-context"; + + PushNotificationConfig config = createSamplePushConfig("http://notify.me/here", "cfg1", "unique_token"); + + // Set up the configuration in the store + configStore.setInfo(taskId, config); + + // Set up latch to wait for async completion + testHttpClient.latch = new CountDownLatch(1); + + TaskStatusUpdateEvent event = new TaskStatusUpdateEvent.Builder() + .taskId(taskId) + .status(new TaskStatus(WORKING)) + .contextId(contextId) + .build(); + + sender.sendNotification(event); + + // Wait for the async operation to complete + assertTrue(testHttpClient.latch.await(5, TimeUnit.SECONDS), "HTTP call should complete within 5 seconds"); + + // Verify the event was sent via HTTP + assertEquals(1, testHttpClient.statusUpdateEvents.size()); + + TaskStatusUpdateEvent updateEvent = testHttpClient.statusUpdateEvents.get(0); + assertEquals(taskId, updateEvent.getTaskId()); + assertEquals(contextId, updateEvent.getContextId()); + assertEquals(WORKING, updateEvent.getStatus().state()); + + // Verify that the X-A2A-Notification-Token header is sent with the correct token + assertEquals(1, testHttpClient.headers.size()); + Map sentHeaders = testHttpClient.headers.get(0); + assertEquals(2, sentHeaders.size()); + assertTrue(sentHeaders.containsKey(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + assertEquals(config.token(), sentHeaders.get(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + // Content-Type header should always be present + assertTrue(sentHeaders.containsKey(CONTENT_TYPE)); + assertEquals(APPLICATION_JSON, sentHeaders.get(CONTENT_TYPE)); + } + + @Test + public void testSendNotificationWithArtifactUpdateEvent() throws InterruptedException { + String taskId = "task_send_notification_with_artifact_update_event"; + String contextId = taskId + "-context"; + String artifactId = taskId + "-artifact"; + + PushNotificationConfig config = createSamplePushConfig("http://notify.me/here", "cfg1", "unique_token"); + + // Set up the configuration in the store + configStore.setInfo(taskId, config); + + // Set up latch to wait for async completion + testHttpClient.latch = new CountDownLatch(1); + + TaskArtifactUpdateEvent event = new TaskArtifactUpdateEvent.Builder() + .taskId(taskId) + .contextId(contextId) + .artifact(new Artifact.Builder() + .artifactId(artifactId) + .parts(List.of(new TextPart("This is a part of the artifact that is updated"))) + .build()) + .lastChunk(false) + .build(); + + sender.sendNotification(event); + + // Wait for the async operation to complete + assertTrue(testHttpClient.latch.await(5, TimeUnit.SECONDS), "HTTP call should complete within 5 seconds"); + + // Verify the event was sent via HTTP + assertEquals(1, testHttpClient.artifactUpdateEvents.size()); + + TaskArtifactUpdateEvent updateEvent = testHttpClient.artifactUpdateEvents.get(0); + assertEquals(taskId, updateEvent.getTaskId()); + assertEquals(contextId, updateEvent.getContextId()); + assertEquals(artifactId, updateEvent.getArtifact().artifactId()); + assertFalse(updateEvent.isLastChunk()); + + // Verify that the X-A2A-Notification-Token header is sent with the correct token + assertEquals(1, testHttpClient.headers.size()); + Map sentHeaders = testHttpClient.headers.get(0); + assertEquals(2, sentHeaders.size()); + assertTrue(sentHeaders.containsKey(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + assertEquals(config.token(), sentHeaders.get(A2AHeaders.X_A2A_NOTIFICATION_TOKEN)); + // Content-Type header should always be present + assertTrue(sentHeaders.containsKey(CONTENT_TYPE)); + assertEquals(APPLICATION_JSON, sentHeaders.get(CONTENT_TYPE)); + } } diff --git a/spec/src/main/java/io/a2a/spec/TaskArtifactUpdateEvent.java b/spec/src/main/java/io/a2a/spec/TaskArtifactUpdateEvent.java index 6cd073eb6..d9be26835 100644 --- a/spec/src/main/java/io/a2a/spec/TaskArtifactUpdateEvent.java +++ b/spec/src/main/java/io/a2a/spec/TaskArtifactUpdateEvent.java @@ -7,6 +7,8 @@ import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonTypeName; +import com.fasterxml.jackson.core.type.TypeReference; + import io.a2a.util.Assert; import static io.a2a.spec.TaskArtifactUpdateEvent.ARTIFACT_UPDATE; @@ -44,6 +46,8 @@ @JsonIgnoreProperties(ignoreUnknown = true) public final class TaskArtifactUpdateEvent implements EventKind, StreamingEventKind, UpdateEvent { + public static final TypeReference TYPE_REFERENCE = new TypeReference<>() {}; + public static final String ARTIFACT_UPDATE = "artifact-update"; private final String taskId; private final Boolean append; diff --git a/spec/src/main/java/io/a2a/spec/TaskStatusUpdateEvent.java b/spec/src/main/java/io/a2a/spec/TaskStatusUpdateEvent.java index 25e2cd170..e3e696d06 100644 --- a/spec/src/main/java/io/a2a/spec/TaskStatusUpdateEvent.java +++ b/spec/src/main/java/io/a2a/spec/TaskStatusUpdateEvent.java @@ -7,6 +7,8 @@ import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonTypeName; +import com.fasterxml.jackson.core.type.TypeReference; + import io.a2a.util.Assert; import static io.a2a.spec.TaskStatusUpdateEvent.STATUS_UPDATE; @@ -20,6 +22,8 @@ @JsonIgnoreProperties(ignoreUnknown = true) public final class TaskStatusUpdateEvent implements EventKind, StreamingEventKind, UpdateEvent { + public static final TypeReference TYPE_REFERENCE = new TypeReference<>() {}; + public static final String STATUS_UPDATE = "status-update"; private final String taskId; private final TaskStatus status; diff --git a/spec/src/main/java/io/a2a/util/Utils.java b/spec/src/main/java/io/a2a/util/Utils.java index 59db4fb67..37f72b313 100644 --- a/spec/src/main/java/io/a2a/util/Utils.java +++ b/spec/src/main/java/io/a2a/util/Utils.java @@ -1,18 +1,29 @@ package io.a2a.util; +import static io.a2a.spec.Message.MESSAGE; +import static io.a2a.spec.Task.TASK; +import static io.a2a.spec.TaskArtifactUpdateEvent.ARTIFACT_UPDATE; +import static io.a2a.spec.TaskStatusUpdateEvent.STATUS_UPDATE; + import java.util.ArrayList; import java.util.List; +import java.util.Map; import java.util.logging.Logger; +import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import io.a2a.spec.Artifact; +import io.a2a.spec.Message; import io.a2a.spec.Part; +import io.a2a.spec.StreamingEventKind; import io.a2a.spec.Task; import io.a2a.spec.TaskArtifactUpdateEvent; +import io.a2a.spec.TaskStatusUpdateEvent; /** * Utility class providing common helper methods for A2A Protocol operations. @@ -69,6 +80,60 @@ public static T unmarshalFrom(String data, TypeReference typeRef) throws return OBJECT_MAPPER.readValue(data, typeRef); } + /** + * Deserializes JSON string into a subtype of StreamingEventKind using Jackson. + *

+ * This method parses JSON containing a StreamingEventKind wrapped into a JSON field + * corresponding to the kind of payload: + *

    + *
  • "task" → {@link Task}
    • + *
  • "message" → {@link Message}
    • + *
  • "status-update" → {@link TaskStatusUpdateEvent}
    • + *
  • "artifact-update" → {@link TaskArtifactUpdateEvent}
    • + *
+ * + * @param data JSON string to deserialize + * @return deserialized StreamingEventKind subclass instance + * @throws JsonProcessingException if JSON parsing fails or kind is unrecognized + */ + public static K unmarshalStreamingEventKindFrom(String data) throws JsonProcessingException { + JsonNode root = OBJECT_MAPPER.readTree(data); + + // Validate that there is exactly one field at the root level + if (root.size() != 1) { + throw new JsonParseException(null, + String.format("Expected exactly one field at root level, but found %d fields", root.size())); + } + + // Get the "kind" field to determine which type to deserialize to + Map.Entry entry = root.fields().next(); + String kind = entry.getKey(); + JsonNode payload = entry.getValue(); + + return (K) switch (kind) { + case TASK -> OBJECT_MAPPER.treeToValue(payload, Task.TYPE_REFERENCE); + case MESSAGE -> OBJECT_MAPPER.treeToValue(payload, Message.TYPE_REFERENCE); + case STATUS_UPDATE -> OBJECT_MAPPER.treeToValue(payload, TaskStatusUpdateEvent.TYPE_REFERENCE); + case ARTIFACT_UPDATE -> OBJECT_MAPPER.treeToValue(payload, TaskArtifactUpdateEvent.TYPE_REFERENCE); + default -> throw new JsonParseException(null, "Unexpected kind of payload: " + kind); + }; + } + + /** + * Serializes a StreamingEventKind in a JSON string + *

+ * The StreamingEventKind object is wrapped in a JSON field named from its kind (e.g. "task") before + * it is serialized + * + * @param kind the StreamingEventKind to deserialize + * @return a JSON String + * @throws JsonProcessingException if JSON parsing fails + */ + public static String marshalFrom(StreamingEventKind kind) throws JsonProcessingException { + Map wrapper = Map.of(kind.getKind(), kind); + return OBJECT_MAPPER.writeValueAsString(wrapper); + } + /** * Returns the provided value if non-null, otherwise returns the default value. *