Event Bus in Native Java #324

taldekar · 2025-01-15T18:30:17Z

Issue #322

Description of changes:
This PR introduces an Event Bus implementation utilizing the Java 9 Reactive Flow library. The event bus extends the functionality provided by the Flow API by adding custom concurrency constructs to enable asynchronous publishing and subscribing.

Brief Design Overview:
This implementation builds upon the basic publisher provided by the Flow API library, introducing a two-tier concurrency model to optimize performance and responsiveness. Here's a breakdown of the solution:

Publisher Side:

Publisher-side Event Queuing: When an event of a particular type is submitted, it's added to an ordered event queue specific to that event type.
Publisher Thread Pool: A dedicated thread pool is maintained for publishing events.
Publisher Callback Job: If no active job exists for an event type, a new job is submitted to the thread pool that will process the events in the events queue for the relevant type.
Registering Callback: A callback is registered for each event type, which determines how the publishing job handles that event. The job will call the registered callback for each event in the event queue.
Callback Invocation: The job systematically calls the registered callback, which is the SubmissionPublisher::submit method on the relevant publisher for each event in the order that the events were received.

Subscriber Side:

Subscriber-side Event Queuing: When a publishing thread submits an event, the subscriber’s Subscriber::onNext method is called. Inside the subscriber, the published event is submitted to another queue unique to that subscriber.
Subscriber Thread Pool: A distinct thread pool is maintained for processing subscriber callbacks.
Subscriber Callback Job: If no active job exists to handle a particular subscriber’s callback, a new job is submitted to the thread pool to process the events in the event queue for that subscriber.
Registering Callback: A callback is registered for the subscriber, which determines how the above job handles the event. For subscribers, this is the EventObserver::onEvent method provided during subscription.
Callback Invocation: The job sequentially invokes the registered callbacks for the events in the queue for each subscriber.

In essence, this solution creates a pipeline where a dedicated thread handles event publishing for each event type, while on the subscriber side, each subscriber's callbacks are processed by a single thread, regardless of the number of events. By maintaining an event queue, we can guarantee event ordering for each subscriber.

Key features:

Utilizes Java 9 Reactive Flow for event publishing and subscription.
Demonstrates integration with existing codebase.
Provides asynchronous publish and subscribe functionality using custom thread pools for each respectively.

Current limitations and areas for further investigation:

Thread pool size's need to be fine-tuned.
Load testing needs to be performed to ensure that the implementation is scalable and performant.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

plugin/src/software/aws/toolkits/eclipse/amazonq/broker/EventBroker.java

plugin/src/software/aws/toolkits/eclipse/amazonq/events/TestEvent.java

shruti0085 · 2025-01-23T21:34:25Z

plugin/src/software/aws/toolkits/eclipse/amazonq/observers/EventObserver.java

+
+    @SuppressWarnings("unchecked")
+    default Class<T> getEventType() {
+        Class<?> currentClass = getClass();


Are there any references that you took inspiration from to compose this class? If yes, those would be useful to link in the description here

Yeah I used stack overflow for this, but it wasn't exactly this. I can try to find it. It think it was this one: https://stackoverflow.com/questions/3437897/how-do-i-get-a-class-instance-of-generic-type-t. I'll add it to the code.

shruti0085 · 2025-01-23T21:40:26Z

plugin/src/software/aws/toolkits/eclipse/amazonq/observers/StreamObserver.java

+
+    @Override
+    @SuppressWarnings("unchecked")
+    default Class<T> getEventType() {


What is the difference between StreamObserver and EventObserver? I see a few new functions in this class but the getEventType looks the same

The EventObserver is the base bones version of a subscriber that only cares about events on the data channel. The StreamObserver extends this functionality with hooks for when the stream errors out or completes. The interfaces were introduced because the standard subscriber method onNext wasn't descriptive or I felt like it didn't communicate that we were listening to/handling events. Also we'd be forced to also implement the onError and onComplete methods everytime.

shruti0085 · 2025-01-23T21:41:12Z

plugin/src/software/aws/toolkits/eclipse/amazonq/plugin/Activator.java

@@ -44,6 +50,16 @@ public Activator() {
                .initializeOnStartUp()
                .build();
        codeReferenceLoggingService = DefaultCodeReferenceLoggingService.getInstance();
+
+        List<TestSubscribers> testSubscriberList = new ArrayList<>(3);


Do we expect Activator to be the source of truth for event broker subscriptions?

No, this was just to test that the event bus is working. Subscriptions can come from anywhere in the codebase.

shruti0085 · 2025-01-23T21:46:28Z

plugin/src/software/aws/toolkits/eclipse/amazonq/subscriber/TestSubscribers.java

+    public void onEvent(final TestEvent event) {
+//        Activator.getLogger().info(event.getMessage());
+
+        if (event.getSequenceNumber() - previousSequenceNumber != 1) {


I am unclear on the significance of the sequence number here

The sequence numbers are more for testing the POC on my end. They helped ensure that the events arrived in the order in which they were published to all subscribers.

shruti0085 · 2025-01-23T21:47:55Z

plugin/src/software/aws/toolkits/eclipse/amazonq/views/AmazonQView.java

-        AuthStatusProvider.addAuthStatusChangeListener(amazonQCommonActions.getFeedbackDialogContributionAction());
-        AuthStatusProvider.addAuthStatusChangeListener(amazonQCommonActions.getCustomizationDialogContributionAction());
+        authStateSubscription = EventBroker.getInstance().subscribe(this);
+        EventBroker.getInstance().subscribe(amazonQCommonActions.getSignoutAction());


Cancels are missing for l88-90 right?

Yup, but they were not implemented in the original code either. To provide cancel functionality, we'd just need to keep track of the Subscription returned.

shruti0085 · 2025-01-23T21:48:39Z

plugin/src/software/aws/toolkits/eclipse/amazonq/publishers/TestPublisher.java

+public final class TestPublisher {
+
+    public TestPublisher() {
+        Thread publisherThread = new Thread(() -> {


Can we not map this to the auth events that get posted?

We have separate publishers for auth events. This was just to demo that the event bus was performant and could deliver events in the order they were posted.

shruti0085 · 2025-01-23T21:49:19Z

plugin/src/software/aws/toolkits/eclipse/amazonq/lsp/auth/AuthStatusProvider.java

-    }
-
-    public static void notifyAuthStatusChanged(final AuthState authState) {
-        if (prevAuthState != null && prevAuthState.equals(authState)) {


I think this check is missing from the publisher calls when an event is published to the broker

No, I added de-duplication logic to the event broker itself here.

Simple event bus in native Java

5d59e9d

taldekar requested a review from breedloj January 15, 2025 18:30

taldekar added 2 commits January 16, 2025 22:57

Concurrency with one thread for each subscriber

2d85818

Asynchronous Event Bus

a0dacfe

taldekar changed the title ~~Simple event bus in native Java~~ Event Bus in Native Java Jan 17, 2025

Improved API and event handling

968aa47

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch from e49aaa4 to 968aa47 Compare January 17, 2025 16:18

Fixed issues with type identification

b915d76

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch 2 times, most recently from c957b00 to 082e09f Compare January 19, 2025 21:37

Improve internal API readability

2c3a4ca

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch from 082e09f to 2c3a4ca Compare January 21, 2025 15:18

Add batch processing of events from queue for improved performance

8fc1f43

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch 2 times, most recently from 6ee52a1 to 7916673 Compare January 21, 2025 23:33

Integrate event broker to publish auth state

1c682f3

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch from 7916673 to 1c682f3 Compare January 21, 2025 23:43

Fix bug and improve performance

dc8b198

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch from 2930b70 to a316ac0 Compare January 22, 2025 18:32

Add auth state dedup logic

98841fd

taldekar force-pushed the taldekar/NativeJavaEventBusPOC branch from a316ac0 to 98841fd Compare January 22, 2025 19:25

taldekar added 3 commits January 22, 2025 18:37

Remove unused auth state provider and listener interface

63f4cc9

Remove unused generic types

048ce63

Remove unused generic types

89934a8

justinmk3 reviewed Jan 23, 2025

View reviewed changes

plugin/src/software/aws/toolkits/eclipse/amazonq/broker/EventBroker.java Show resolved Hide resolved

justinmk3 reviewed Jan 23, 2025

View reviewed changes

plugin/src/software/aws/toolkits/eclipse/amazonq/events/TestEvent.java Outdated Show resolved Hide resolved

Document code as necessary

9c12e8c

taldekar requested a review from justinmk3 January 23, 2025 17:21

taldekar added 2 commits January 23, 2025 12:48

Document code as necessary

4b9f100

Document code as necessary

3531106

shruti0085 reviewed Jan 23, 2025

View reviewed changes

taldekar requested a review from shruti0085 January 23, 2025 22:38

taldekar closed this Jan 23, 2025

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Event Bus in Native Java #324

Event Bus in Native Java #324

taldekar commented Jan 15, 2025 •

edited

Loading

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025 •

edited

Loading

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

shruti0085 Jan 23, 2025

taldekar Jan 23, 2025

Event Bus in Native Java #324

Event Bus in Native Java #324

Conversation

taldekar commented Jan 15, 2025 • edited Loading

Choose a reason for hiding this comment

taldekar Jan 23, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

taldekar commented Jan 15, 2025 •

edited

Loading

taldekar Jan 23, 2025 •

edited

Loading