feat: Implement PushNotifications as per the 1.0 spec#622
feat: Implement PushNotifications as per the 1.0 spec#622kabir wants to merge 1 commit intoa2aproject:mainfrom
Conversation
Summary of ChangesHello @kabir, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly refactors the event processing and persistence architecture to align with the A2A 1.0 specification, particularly for streaming and push notifications. The introduction of a centralized event bus and processor ensures that all client-visible events and notifications reflect a durably stored state, enhancing consistency and reliability. Resource management for streaming connections has been improved with automatic client-side closing and server-side cancellation on disconnects. These changes, along with a more robust threading model, contribute to a more scalable and compliant system. Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a significant and well-executed architectural refactoring of the server's event processing and queueing system. The introduction of a centralized MainEventBus and a dedicated MainEventBusProcessor is a major improvement, enforcing a "persist-before-visibility" pattern that enhances robustness and eliminates potential race conditions. The changes also address critical issues like resource leaks on client disconnects and thread pool exhaustion under load, and improve compliance with the A2A specification for push notifications and event streaming. The code quality is high, with clear documentation for the new architecture. I have a couple of minor suggestions regarding code duplication in the client and a potential inconsistency in the replicated event handling logic, but overall this is an excellent set of changes.
| boolean shouldClose = false; | ||
| if (event instanceof TaskStatusUpdateEvent tue && tue.isFinal()) { | ||
| shouldClose = true; | ||
| } else if (event instanceof Task task) { | ||
| TaskState state = task.status().state(); | ||
| if (state.isFinal()) { | ||
| shouldClose = true; | ||
| } | ||
| } |
There was a problem hiding this comment.
This logic for determining if the connection should be closed based on a final event state is duplicated in client/transport/jsonrpc/src/main/java/io/a2a/client/transport/jsonrpc/sse/SSEEventListener.java. To improve maintainability and avoid future inconsistencies, consider extracting this logic into a shared static utility method. For example, a method StreamingEventUtils.isFinal(StreamingEventKind event) could be placed in a common client module.
| * IMPORTANT: We send TaskStatusUpdateEvent instead of full Task to maintain consistency | ||
| * with local event distribution. Clients expect TaskStatusUpdateEvent for status changes, | ||
| * and sending the full Task causes issues in remote instances where clients don't handle | ||
| * bare Task objects the same way they handle TaskStatusUpdateEvent. |
There was a problem hiding this comment.
The comment here states that sending a full Task object can cause issues for clients. However, other changes in this PR, specifically in SSEEventListener and RestSSEEventListener, add logic to handle receiving a Task with a final state to close the connection. This suggests clients are now expected to handle Task objects in the stream. If clients can indeed handle final Task objects, it might be simpler and more consistent to replicate the final Task directly from the TaskFinalizedEvent, rather than converting it to a TaskStatusUpdateEvent. This would also make the event flow more uniform. If the conversion is still necessary for other reasons, could the comment be updated to clarify the specific issues that remain?
kabir
force-pushed
the
event-queue-bus-improve-pnc
branch
from
62e99c5 to
9414bfb
Compare
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request refactors the push notification mechanism to align with the 1.0 spec, allowing any StreamingEventKind (like Task, Message, TaskStatusUpdateEvent, and TaskArtifactUpdateEvent) to be sent as a notification, instead of just Task snapshots. The changes are well-implemented across the PushNotificationSender interface and its base implementation, ensuring events are serialized correctly into the StreamResponse format. The tests have been comprehensively updated to validate the new behavior for all event types, and documentation has been improved to reflect these changes. The overall implementation is solid and enhances the push notification feature significantly. I have one minor suggestion to improve the extensibility of the BasePushNotificationSender.
| * @param event the streaming event | ||
| * @return the task ID, or null if not available | ||
| */ | ||
| private @Nullable String extractTaskId(StreamingEventKind event) { |
There was a problem hiding this comment.
To improve extensibility, consider changing the visibility of extractTaskId from private to protected. This would allow custom PushNotificationSender implementations that extend BasePushNotificationSender to reuse this helpful utility method. The Javadoc example in the PushNotificationSender interface already implies the existence of such a reusable helper.
| private @Nullable String extractTaskId(StreamingEventKind event) { | |
| protected @Nullable String extractTaskId(StreamingEventKind event) { |
1 participant
Fixes #490
Fixes: #594
Includes #611 (until that is merged)