refactor: Update SSEEventListener to use ProtoUtils.FromProto.streamingEventKind

Updated SSEEventListener to use the new ProtoUtils.FromProto.streamingEventKind()
method instead of Jackson for deserializing streaming events.

Changes:
- Replaced Jackson JsonNode parsing with JSONRPCUtils.parseResponseEvent()
- Now uses ProtoUtils.FromProto.streamingEventKind() to convert proto
  StreamResponse to domain StreamingEventKind
- Removed dependency on Utils.OBJECT_MAPPER
- Simplified error handling - JSONRPCUtils.parseResponseEvent() throws
  JSONRPCError which is caught and passed to errorHandler

This ensures consistent use of protobuf JSON parsing throughout the
streaming implementation and leverages the StreamResponseMapper we
created earlier.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: ehsavoie

client/transport/jsonrpc/src/main/java/io/a2a/client/transport/jsonrpc/sse/SSEEventListener.java

@@ -1,7 +1,6 @@
 package io.a2a.client.transport.jsonrpc.sse;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
 import io.a2a.spec.JSONRPCError;
 import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.TaskStatusUpdateEvent;
@@ -10,8 +9,9 @@
 import java.util.function.Consumer;
 import java.util.logging.Logger;
 
-import static io.a2a.util.Utils.OBJECT_MAPPER;
-
+import io.a2a.grpc.StreamResponse;
+import io.a2a.grpc.utils.JSONRPCUtils;
+import io.a2a.grpc.utils.ProtoUtils;
 import org.jspecify.annotations.Nullable;
 
 public class SSEEventListener {
@@ -29,11 +29,7 @@ public SSEEventListener(Consumer<StreamingEventKind> eventHandler,
     }
 
     public void onMessage(String message, @Nullable Future<Void> completableFuture) {
-        try {
-            handleMessage(OBJECT_MAPPER.readTree(message), completableFuture);
-        } catch (JsonProcessingException e) {
-            log.warning("Failed to parse JSON message: " + message);
-        }
+        handleMessage(message, completableFuture);
     }
 
     public void onError(Throwable throwable, @Nullable Future<Void> future) {
@@ -63,25 +59,20 @@ public void onComplete() {
         }
     }
 
-    private void handleMessage(JsonNode jsonNode, @Nullable Future<Void> future) {
+    private void handleMessage(String message, @Nullable Future<Void> future) {
         try {
-            if (jsonNode.has("error")) {
-                JSONRPCError error = OBJECT_MAPPER.treeToValue(jsonNode.get("error"), JSONRPCError.class);
-                if (errorHandler != null) {
-                    errorHandler.accept(error);
-                }
-            } else if (jsonNode.has("result")) {
-                // result can be a Task, Message, TaskStatusUpdateEvent, or TaskArtifactUpdateEvent
-                JsonNode result = jsonNode.path("result");
-                StreamingEventKind event = OBJECT_MAPPER.treeToValue(result, StreamingEventKind.class);
-                eventHandler.accept(event);
-                if (event instanceof TaskStatusUpdateEvent && ((TaskStatusUpdateEvent) event).isFinal()) {
-                    if (future != null) {
-                        future.cancel(true); // close SSE channel
-                    }
+            StreamResponse response = JSONRPCUtils.parseResponseEvent(message);
+
+            StreamingEventKind event = ProtoUtils.FromProto.streamingEventKind(response);
+            eventHandler.accept(event);
+            if (event instanceof TaskStatusUpdateEvent && ((TaskStatusUpdateEvent) event).isFinal()) {
+                if (future != null) {
+                    future.cancel(true); // close SSE channel
                 }
-            } else {
-                throw new IllegalArgumentException("Unknown message type");
+            }
+        } catch (JSONRPCError error) {
+            if (errorHandler != null) {
+                errorHandler.accept(error);
             }
         } catch (JsonProcessingException e) {
             throw new RuntimeException(e);