Introduce RxJava-based Event Bus (#330)

* Add event bus

* Integrate event bus

* Remove auth status change listener and provider

* Fix indentation in pom

* Fix pom.xml indentation

* Remove singleton implementation and add null check to post method

* Move EventBroker access to Activator

* Update event broker calls to go through Activator

* Fix bug caused by null EventBroker most likely because Activator hasn't been intitialized yet

* Fix bug caused by null EventBroker most likely because Activator hasn't been intitialized yet

* Add event bus

* Remove subscription and type introspection

* Fix artifact versioning bug

* Added test coverage for cleanup method (#310)

* Add tests for EventBroker component

* Delete accidental RemoteLspFetcherTest addition

* Revert "Delete accidental RemoteLspFetcherTest addition"

This reverts commit 15ae75a.

* Remove implicit deduplication

---------

Co-authored-by: Nicolas <[email protected]>
Co-authored-by: Jonathan Breedlove <[email protected]>

plugin/META-INF/MANIFEST.MF

@@ -62,6 +62,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,11 +42,21 @@
     </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>
             <version>2.0.1</version>
         </dependency>
+	<dependency>
+	    <groupId>io.reactivex.rxjava3</groupId>
+	    <artifactId>rxjava</artifactId>
+	    <version>3.1.5</version>
+	</dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
@@ -139,7 +149,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,org.apache.maven</includeGroupIds>
+                            <includeGroupIds>io.reactivex,software.amazon.awssdk,com.fasterxml.jackson,com.nimbusds,jakarta.inject,commons-codec,org.apache.httpcomponents,org.reactivestreams,org.apache.maven</includeGroupIds>
                         </configuration>
                     </execution>
                     <execution>

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

@@ -0,0 +1,37 @@
+// 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;
+
+public final class EventBroker {
+
+    private final Subject<Object> eventBus = PublishSubject.create().toSerialized();
+
+    public <T> void post(final T event) {
+        if (event == null) {
+            return;
+        }
+        eventBus.onNext(event);
+    }
+
+    public <T> Disposable subscribe(final Class<T> eventType, final EventObserver<T> observer) {
+        Consumer<T> consumer = new Consumer<>() {
+            @Override
+            public void accept(final T event) {
+                observer.onEvent(event);
+            }
+        };
+
+        return eventBus.ofType(eventType)
+                .observeOn(Schedulers.computation())
+                .subscribe(consumer);
+        }
+
+}

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

@@ -0,0 +1,9 @@
+// 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;
+
+@FunctionalInterface
+public interface EventObserver<T> {
+    void onEvent(T event);
+}

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

@@ -13,6 +13,8 @@
 import org.eclipse.ui.PlatformUI;
 import org.eclipse.ui.services.ISourceProviderService;
 
+import io.reactivex.rxjava3.disposables.Disposable;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
 
@@ -30,12 +32,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 Disposable authStateSubscription;
 
     public AuthSourceProvider() {
-        AuthStatusProvider.addAuthStatusChangeListener(this);
+        authStateSubscription = Activator.getEventBroker().subscribe(AuthState.class, this);
         isLoggedIn = Activator.getLoginService().getAuthState().isLoggedIn();
     }
 
@@ -54,7 +57,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.dispose();
     }
 
     @Override
@@ -69,15 +72,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

@@ -41,6 +41,7 @@ public final class DefaultAuthStateManager implements AuthStateManager {
     private LoginParams loginParams; // used in login's getSsoToken params
     private String issuerUrl; // used in AmazonQLspClientImpl.getConnectionMetadata()
     private String ssoTokenId; // used in logout's invalidateSsoToken params
+    private AuthState previousAuthState = null;
 
     public DefaultAuthStateManager(final PluginStore pluginStore) {
         this.authPluginStore = new AuthPluginStore(pluginStore);
@@ -82,7 +83,6 @@ public void toExpired() {
             toLoggedOut();
             return;
         }
-
         updateState(AuthStateType.EXPIRED, loginType, loginParams, ssoTokenId);
     }
 
@@ -119,7 +119,11 @@ 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());
+        AuthState newAuthState = getAuthState();
+        if (previousAuthState == null || newAuthState.authStateType() != previousAuthState.authStateType()) {
+            Activator.getEventBroker().post(newAuthState);
+        }
+        previousAuthState = newAuthState;
     }
 
     private void syncAuthStateWithPluginStore() {

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

@@ -6,6 +6,8 @@
 import org.eclipse.ui.plugin.AbstractUIPlugin;
 import org.osgi.framework.BundleContext;
 
+import software.aws.toolkits.eclipse.amazonq.broker.EventBroker;
+import software.aws.toolkits.eclipse.amazonq.chat.ChatStateManager;
 import software.aws.toolkits.eclipse.amazonq.configuration.DefaultPluginStore;
 import software.aws.toolkits.eclipse.amazonq.configuration.PluginStore;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.DefaultLoginService;
@@ -14,11 +16,10 @@
 import software.aws.toolkits.eclipse.amazonq.providers.LspProviderImpl;
 import software.aws.toolkits.eclipse.amazonq.telemetry.service.DefaultTelemetryService;
 import software.aws.toolkits.eclipse.amazonq.telemetry.service.TelemetryService;
-import software.aws.toolkits.eclipse.amazonq.util.PluginLogger;
-import software.aws.toolkits.eclipse.amazonq.chat.ChatStateManager;
 import software.aws.toolkits.eclipse.amazonq.util.CodeReferenceLoggingService;
 import software.aws.toolkits.eclipse.amazonq.util.DefaultCodeReferenceLoggingService;
 import software.aws.toolkits.eclipse.amazonq.util.LoggingService;
+import software.aws.toolkits.eclipse.amazonq.util.PluginLogger;
 
 public class Activator extends AbstractUIPlugin {
 
@@ -30,6 +31,7 @@ public class Activator extends AbstractUIPlugin {
     private static LoginService loginService;
     private static CodeReferenceLoggingService codeReferenceLoggingService;
     private static PluginStore pluginStore;
+    private static EventBroker eventBroker = new EventBroker();
 
     public Activator() {
         super();
@@ -76,5 +78,8 @@ public static PluginStore getPluginStore() {
     public static CodeReferenceLoggingService getCodeReferenceLoggingService() {
         return codeReferenceLoggingService;
     }
+    public static EventBroker getEventBroker() {
+        return eventBroker;
+    }
 
 }

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,22 @@
 import org.eclipse.swt.widgets.Display;
 import org.eclipse.ui.part.ViewPart;
 
+import io.reactivex.rxjava3.disposables.Disposable;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
 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 Disposable authStateSubscription;
+
     protected AmazonQView() {
         this.viewController = new AmazonQViewController();
     }
@@ -81,10 +83,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 = Activator.getEventBroker().subscribe(AuthState.class, this);
+        Activator.getEventBroker().subscribe(AuthState.class, amazonQCommonActions.getSignoutAction());
+        Activator.getEventBroker().subscribe(AuthState.class, amazonQCommonActions.getFeedbackDialogContributionAction());
+        Activator.getEventBroker().subscribe(AuthState.class, amazonQCommonActions.getCustomizationDialogContributionAction());
     }
 
     @Override
@@ -123,7 +125,7 @@ function waitForFunction(functionName, timeout = 30000) {
      */
     @Override
     public void dispose() {
-        AuthStatusProvider.removeAuthStatusChangeListener(this);
+        authStateSubscription.dispose();
         super.dispose();
     }
 

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

@@ -18,18 +18,18 @@
 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 io.reactivex.rxjava3.disposables.Disposable;
+import software.aws.toolkits.eclipse.amazonq.broker.api.EventObserver;
 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 +38,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 Disposable 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 = Activator.getEventBroker().subscribe(AuthState.class, this);
     }
 
     @Override
@@ -98,7 +100,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 +140,7 @@ protected void updateButtonStyle(final Button button) {
 
     @Override
     public void dispose() {
-        AuthStatusProvider.removeAuthStatusChangeListener(this);
+        authStateSubscription.dispose();
     }
 
     @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());