Introduce RxJava-based Event Bus

plugin/META-INF/MANIFEST.MF

@@ -61,6 +61,7 @@ Bundle-Classpath: target/classes/,
  target/dependency/regions-2.28.26.jar,
  target/dependency/retries-2.28.26.jar,
  target/dependency/retries-spi-2.28.26.jar,
+ target/dependency/rxjava-3.1.5.jar,
  target/dependency/sdk-core-2.28.26.jar,
  target/dependency/third-party-jackson-core-2.28.26.jar,
  target/dependency/utils-2.28.26.jar

plugin/pom.xml

@@ -42,6 +42,11 @@
     </dependencyManagement>
 
     <dependencies>
+		<dependency>
+		    <groupId>io.reactivex.rxjava3</groupId>
+		    <artifactId>rxjava</artifactId>
+		    <version>3.1.5</version>
+		</dependency>
         <dependency>
             <groupId>jakarta.inject</groupId>
             <artifactId>jakarta.inject-api</artifactId>
@@ -134,7 +139,7 @@
                         </goals>
                         <configuration>
                             <outputDirectory>${project.build.directory}/dependency</outputDirectory>
-                            <includeGroupIds>software.amazon.awssdk,com.fasterxml.jackson,com.nimbusds,jakarta.inject,commons-codec,org.apache.httpcomponents,org.reactivestreams</includeGroupIds>
+                            <includeGroupIds>io.reactivex,software.amazon.awssdk,com.fasterxml.jackson,com.nimbusds,jakarta.inject,commons-codec,org.apache.httpcomponents,org.reactivestreams</includeGroupIds>
                         </configuration>
                     </execution>
                     <execution>

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

@@ -0,0 +1,51 @@
+// Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package software.aws.toolkits.eclipse.amazonq.broker;
+
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subjects.PublishSubject;
+import io.reactivex.rxjava3.subjects.Subject;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
+import software.aws.toolkits.eclipse.amazonq.broker.api.Subscription;
+
+public final class EventBroker {
+
+    private static final EventBroker INSTANCE;
+    private final Subject<Object> eventBus = PublishSubject.create().toSerialized();
+
+    static {
+        INSTANCE = new EventBroker();
+    }
+
+    public static EventBroker getInstance() {
+        return INSTANCE;
+    }
+
+    public <T> void post(final T event) {
+        eventBus.onNext(event);
+    }
+
+    public <T> Subscription subscribe(final EventObserver<T> observer) {
+        Consumer<T> consumer = new Consumer<>() {
+            @Override
+            public void accept(final T event) {
+                observer.onEvent(event);
+            }
+        };
+        Disposable disposable = eventBus.ofType(observer.getEventType()).distinct()
+                .observeOn(Schedulers.computation())
+                .subscribe(consumer);
+
+        Subscription subscription = new Subscription() {
+            @Override
+            public void cancel() {
+                disposable.dispose();
+            }
+        };
+        return subscription;
+    }
+
+}

plugin/src/software/aws/toolkits/eclipse/amazonq/broker/api/EventObserver.java

@@ -0,0 +1,33 @@
+// Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package software.aws.toolkits.eclipse.amazonq.broker.api;
+
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+
+public interface EventObserver<T> {
+
+    // Reference:
+    // https://stackoverflow.com/questions/3437897/how-do-i-get-a-class-instance-of-generic-type-t
+    @SuppressWarnings("unchecked")
+    default Class<T> getEventType() {
+        Class<?> currentClass = getClass();
+        while (currentClass != null) {
+            for (Type type : currentClass.getGenericInterfaces()) {
+                if (type instanceof ParameterizedType paramType && (paramType.getRawType() == EventObserver.class)) {
+                    Type typeArg = paramType.getActualTypeArguments()[0];
+                    if (typeArg instanceof Class<?>) {
+                        return (Class<T>) typeArg;
+                    }
+                    throw new IllegalStateException("Generic type parameter is not a Class");
+                }
+            }
+            currentClass = currentClass.getSuperclass();
+        }
+        throw new IllegalStateException("Could not determine generic type");
+    }
+
+    void onEvent(T event);
+
+}

plugin/src/software/aws/toolkits/eclipse/amazonq/broker/api/Subscription.java

@@ -0,0 +1,10 @@
+// Copyright 2024 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package software.aws.toolkits.eclipse.amazonq.broker.api;
+
+public interface Subscription {
+
+    void cancel();
+
+}

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

@@ -13,6 +13,9 @@
 import org.eclipse.ui.PlatformUI;
 import org.eclipse.ui.services.ISourceProviderService;
 
+import software.aws.toolkits.eclipse.amazonq.broker.EventBroker;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
+import software.aws.toolkits.eclipse.amazonq.broker.api.Subscription;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
 
@@ -30,12 +33,13 @@
  * @see software.aws.toolkits.eclipse.amazonq.toolbar
  * @see software.aws.toolkits.eclipse.amazonq.toolbar-unauthenticated
  */
-public final class AuthSourceProvider extends AbstractSourceProvider implements AuthStatusChangedListener {
+public final class AuthSourceProvider extends AbstractSourceProvider implements EventObserver<AuthState> {
     public static final String IS_LOGGED_IN_VARIABLE_ID = "is_logged_in";
     private boolean isLoggedIn = false;
+    private Subscription authStateSubscription;
 
     public AuthSourceProvider() {
-        AuthStatusProvider.addAuthStatusChangeListener(this);
+        authStateSubscription = EventBroker.getInstance().subscribe(this);
         isLoggedIn = Activator.getLoginService().getAuthState().isLoggedIn();
     }
 
@@ -54,7 +58,7 @@ public void dispose() {
         // Notify listeners that this provider is being disposed
         fireSourceChanged(ISources.WORKBENCH, IS_LOGGED_IN_VARIABLE_ID, null);
 
-        AuthStatusProvider.removeAuthStatusChangeListener(this);
+        authStateSubscription.cancel();
     }
 
     @Override
@@ -69,15 +73,15 @@ public void setIsLoggedIn(final Boolean isLoggedIn) {
 
     public static AuthSourceProvider getProvider() {
         IWorkbench workbench = PlatformUI.getWorkbench();
-        ISourceProviderService sourceProviderService = (ISourceProviderService) workbench
+        ISourceProviderService sourceProviderService = workbench
                 .getService(ISourceProviderService.class);
         AuthSourceProvider provider = (AuthSourceProvider) sourceProviderService
                 .getSourceProvider(AuthSourceProvider.IS_LOGGED_IN_VARIABLE_ID);
         return provider;
     }
 
     @Override
-    public void onAuthStatusChanged(final AuthState authState) {
+    public void onEvent(final AuthState authState) {
         boolean isLoggedIn = authState.isLoggedIn();
         Display.getDefault().asyncExec(() -> {
             setIsLoggedIn(isLoggedIn);

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

@@ -3,6 +3,7 @@
 
 package software.aws.toolkits.eclipse.amazonq.lsp.auth;
 
+import software.aws.toolkits.eclipse.amazonq.broker.EventBroker;
 import software.aws.toolkits.eclipse.amazonq.configuration.PluginStore;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthStateType;
@@ -119,7 +120,7 @@ private void updateState(final AuthStateType authStatusType, final LoginType log
          * This notification is critical for ensuring all plugin components reflect the current
          * authentication state.
          */
-        AuthStatusProvider.notifyAuthStatusChanged(getAuthState());
+        EventBroker.getInstance().post(getAuthState());
     }
 
     private void syncAuthStateWithPluginStore() {

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

@@ -117,7 +117,7 @@ public void completed(final ProgressEvent event) {
         });
 
         // Check if user is authenticated and build view accordingly
-        onAuthStatusChanged(authState);
+        onEvent(authState);
     }
 
     private Browser getAndUpdateStateManager() {
@@ -127,7 +127,7 @@ private Browser getAndUpdateStateManager() {
     }
 
     @Override
-    public final void onAuthStatusChanged(final AuthState authState) {
+    public final void onEvent(final AuthState authState) {
         Display.getDefault().asyncExec(() -> {
             amazonQCommonActions.updateActionVisibility(authState, getViewSite());
             if (authState.isExpired()) {

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

@@ -9,20 +9,23 @@
 import org.eclipse.swt.widgets.Display;
 import org.eclipse.ui.part.ViewPart;
 
+import software.aws.toolkits.eclipse.amazonq.broker.EventBroker;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
+import software.aws.toolkits.eclipse.amazonq.broker.api.Subscription;
 import software.aws.toolkits.eclipse.amazonq.controllers.AmazonQViewController;
-import software.aws.toolkits.eclipse.amazonq.lsp.auth.AuthStatusChangedListener;
-import software.aws.toolkits.eclipse.amazonq.lsp.auth.AuthStatusProvider;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
 import software.aws.toolkits.eclipse.amazonq.util.ThemeDetector;
 import software.aws.toolkits.eclipse.amazonq.views.actions.AmazonQCommonActions;
 
-public abstract class AmazonQView extends ViewPart implements AuthStatusChangedListener {
+public abstract class AmazonQView extends ViewPart implements EventObserver<AuthState> {
 
     private AmazonQViewController viewController;
     private AmazonQCommonActions amazonQCommonActions;
     private static final ThemeDetector THEME_DETECTOR = new ThemeDetector();
 
+    private Subscription authStateSubscription;
+
     protected AmazonQView() {
         this.viewController = new AmazonQViewController();
     }
@@ -81,10 +84,10 @@ private void setupActions(final AuthState authState) {
     }
 
     private void setupAuthStatusListeners() {
-        AuthStatusProvider.addAuthStatusChangeListener(this);
-        AuthStatusProvider.addAuthStatusChangeListener(amazonQCommonActions.getSignoutAction());
-        AuthStatusProvider.addAuthStatusChangeListener(amazonQCommonActions.getFeedbackDialogContributionAction());
-        AuthStatusProvider.addAuthStatusChangeListener(amazonQCommonActions.getCustomizationDialogContributionAction());
+        authStateSubscription = EventBroker.getInstance().subscribe(this);
+        EventBroker.getInstance().subscribe(amazonQCommonActions.getSignoutAction());
+        EventBroker.getInstance().subscribe(amazonQCommonActions.getFeedbackDialogContributionAction());
+        EventBroker.getInstance().subscribe(amazonQCommonActions.getCustomizationDialogContributionAction());
     }
 
     @Override
@@ -123,7 +126,7 @@ function waitForFunction(functionName, timeout = 30000) {
      */
     @Override
     public void dispose() {
-        AuthStatusProvider.removeAuthStatusChangeListener(this);
+        authStateSubscription.cancel();
         super.dispose();
     }
 

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

@@ -18,18 +18,19 @@
 import org.eclipse.swt.widgets.Display;
 import org.eclipse.swt.widgets.Link;
 
-import software.aws.toolkits.eclipse.amazonq.lsp.auth.AuthStatusChangedListener;
-import software.aws.toolkits.eclipse.amazonq.lsp.auth.AuthStatusProvider;
+import software.aws.toolkits.eclipse.amazonq.broker.EventBroker;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
+import software.aws.toolkits.eclipse.amazonq.broker.api.Subscription;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
+import software.aws.toolkits.eclipse.amazonq.telemetry.UiTelemetryProvider;
 import software.aws.toolkits.eclipse.amazonq.util.Constants;
 import software.aws.toolkits.eclipse.amazonq.util.PluginUtils;
 import software.aws.toolkits.eclipse.amazonq.util.ThreadingUtils;
 import software.aws.toolkits.eclipse.amazonq.views.actions.SignoutAction;
-import software.aws.toolkits.eclipse.amazonq.telemetry.UiTelemetryProvider;
 
 
-public final class ReauthenticateView extends CallToActionView implements AuthStatusChangedListener {
+public final class ReauthenticateView extends CallToActionView implements EventObserver<AuthState> {
     public static final String ID = "software.aws.toolkits.eclipse.amazonq.views.ReauthenticateView";
 
     private static final String ICON_PATH = "icons/AmazonQ64.png";
@@ -38,10 +39,12 @@ public final class ReauthenticateView extends CallToActionView implements AuthSt
     private static final String BUTTON_LABEL = "Re-authenticate";
     private static final String LINK_LABEL = "Sign out";
 
+    private Subscription authStateSubscription;
+
     public ReauthenticateView() {
          // It is necessary for this view to be an `AuthStatusChangedListener` to switch the view back to Q Chat after the authentication
          // flow is successful. Without this listener, the re-authentication will succeed but the view will remain present.
-        AuthStatusProvider.addAuthStatusChangeListener(this);
+         authStateSubscription = EventBroker.getInstance().subscribe(this);
     }
 
     @Override
@@ -98,7 +101,7 @@ public void widgetSelected(final SelectionEvent e) {
     }
 
     @Override
-    public void onAuthStatusChanged(final AuthState authState) {
+    public void onEvent(final AuthState authState) {
         Display.getDefault().asyncExec(() -> {
             if (authState.isLoggedIn()) {
                 ViewVisibilityManager.showChatView("update");
@@ -138,7 +141,7 @@ protected void updateButtonStyle(final Button button) {
 
     @Override
     public void dispose() {
-        AuthStatusProvider.removeAuthStatusChangeListener(this);
+        authStateSubscription.cancel();
     }
 
     @Override

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

@@ -92,11 +92,11 @@ public Object function(final Object[] arguments) {
         amazonQCommonActions = getAmazonQCommonActions();
 
         // Check if user is authenticated and build view accordingly
-        onAuthStatusChanged(authState);
+        onEvent(authState);
     }
 
     @Override
-    public void onAuthStatusChanged(final AuthState authState) {
+    public void onEvent(final AuthState authState) {
         var browser = getBrowser();
         Display.getDefault().asyncExec(() -> {
             amazonQCommonActions.updateActionVisibility(authState, getViewSite());