test: add test for streaming with multiple events (#326)

## Summary
Implements tests for streaming message scenarios where multiple events
are received from the server, addressing Issue #52. This test validates
the proper handling of event streams in AI agent-to-agent communication
using the A2A Protocol.

## Overview
The `testOnMessageStreamNewMessageMultipleEventsSuccess()` method
verifies that the JSON-RPC handler correctly processes a sequence of
multiple events during streaming communication:

1. **Task Event (WORKING state)** - Initial task state update
2. **TaskArtifactUpdateEvent** - Artifact generation during task
execution
3. **TaskStatusUpdateEvent (COMPLETED state)** - Final task completion
signal


- [x] Follow the [`CONTRIBUTING` Guide](../CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [x] Ensure the tests pass
- [x] Appropriate READMEs were updated (if necessary)

Fixes #52 🦕

---

Author: amithsourya

transport/jsonrpc/src/test/java/io/a2a/transport/jsonrpc/handler/JSONRPCHandlerTest.java

@@ -335,6 +335,114 @@ public void onComplete() {
         Assertions.assertSame(message, results.get(0));
     }
 
+    @Test
+    public void testOnMessageStreamNewMessageMultipleEventsSuccess() throws InterruptedException {
+        JSONRPCHandler handler = new JSONRPCHandler(CARD, requestHandler);
+
+        // Create multiple events to be sent during streaming
+        Task taskEvent = new Task.Builder(MINIMAL_TASK)
+                .status(new TaskStatus(TaskState.WORKING))
+                .build();
+
+        TaskArtifactUpdateEvent artifactEvent = new TaskArtifactUpdateEvent.Builder()
+                .taskId(MINIMAL_TASK.getId())
+                .contextId(MINIMAL_TASK.getContextId())
+                .artifact(new Artifact.Builder()
+                        .artifactId("artifact-1")
+                        .parts(new TextPart("Generated artifact content"))
+                        .build())
+                .build();
+
+        TaskStatusUpdateEvent statusEvent = new TaskStatusUpdateEvent.Builder()
+                .taskId(MINIMAL_TASK.getId())
+                .contextId(MINIMAL_TASK.getContextId())
+                .status(new TaskStatus(TaskState.COMPLETED))
+                .build();
+
+        // Configure the agent executor to enqueue multiple events
+        agentExecutorExecute = (context, eventQueue) -> {
+            // Enqueue the task with WORKING state
+            eventQueue.enqueueEvent(taskEvent);
+            // Enqueue an artifact update event
+            eventQueue.enqueueEvent(artifactEvent);
+            // Enqueue a status update event to complete the task (this is the "final" event)
+            eventQueue.enqueueEvent(statusEvent);
+        };
+
+        Message message = new Message.Builder(MESSAGE)
+                .taskId(MINIMAL_TASK.getId())
+                .contextId(MINIMAL_TASK.getContextId())
+                .build();
+
+        SendStreamingMessageRequest request = new SendStreamingMessageRequest(
+                "1", new MessageSendParams(message, null, null));
+        Flow.Publisher<SendStreamingMessageResponse> response = handler.onMessageSendStream(request, callContext);
+
+        List<StreamingEventKind> results = new ArrayList<>();
+        CountDownLatch latch = new CountDownLatch(3); // Expect 3 events
+        AtomicReference<Throwable> error = new AtomicReference<>();
+
+        response.subscribe(new Flow.Subscriber<>() {
+            private Flow.Subscription subscription;
+
+            @Override
+            public void onSubscribe(Flow.Subscription subscription) {
+                this.subscription = subscription;
+                subscription.request(1);
+            }
+
+            @Override
+            public void onNext(SendStreamingMessageResponse item) {
+                results.add(item.getResult());
+                subscription.request(1);
+                latch.countDown();
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                error.set(throwable);
+                subscription.cancel();
+                // Release latch to prevent timeout
+                while (latch.getCount() > 0) {
+                    latch.countDown();
+                }
+            }
+
+            @Override
+            public void onComplete() {
+                subscription.cancel();
+            }
+        });
+
+        // Wait for all events to be received
+        Assertions.assertTrue(latch.await(2, TimeUnit.SECONDS),
+                "Expected to receive 3 events within timeout");
+
+        // Assert no error occurred during streaming
+        Assertions.assertNull(error.get(), "No error should occur during streaming");
+
+        // Verify that all 3 events were received
+        assertEquals(3, results.size(), "Should have received exactly 3 events");
+
+        // Verify the first event is the task
+        Task receivedTask = assertInstanceOf(Task.class, results.get(0), "First event should be a Task");
+        assertEquals(MINIMAL_TASK.getId(), receivedTask.getId());
+        assertEquals(MINIMAL_TASK.getContextId(), receivedTask.getContextId());
+        assertEquals(TaskState.WORKING, receivedTask.getStatus().state());
+
+        // Verify the second event is the artifact update
+        TaskArtifactUpdateEvent receivedArtifact = assertInstanceOf(TaskArtifactUpdateEvent.class, results.get(1),
+                "Second event should be a TaskArtifactUpdateEvent");
+        assertEquals(MINIMAL_TASK.getId(), receivedArtifact.getTaskId());
+        assertEquals("artifact-1", receivedArtifact.getArtifact().artifactId());
+
+        // Verify the third event is the status update
+        TaskStatusUpdateEvent receivedStatus = assertInstanceOf(TaskStatusUpdateEvent.class, results.get(2),
+                "Third event should be a TaskStatusUpdateEvent");
+        assertEquals(MINIMAL_TASK.getId(), receivedStatus.getTaskId());
+        assertEquals(TaskState.COMPLETED, receivedStatus.getStatus().state());
+    }
+
     @Test
     public void testOnMessageStreamNewMessageSuccessMocks() {
         JSONRPCHandler handler = new JSONRPCHandler(CARD, requestHandler);