Add event bus

Author: taldekar

plugin/META-INF/MANIFEST.MF

@@ -52,7 +52,6 @@ Bundle-Classpath: target/classes/,
  target/dependency/jackson-databind-2.17.2.jar,
  target/dependency/jakarta.inject-api-2.0.1.jar,
  target/dependency/json-utils-2.28.26.jar,
- target/dependency/maven-artifact-3.9.9.jar,
  target/dependency/metrics-spi-2.28.26.jar,
  target/dependency/netty-nio-client-2.28.26.jar,
  target/dependency/nimbus-jose-jwt-9.41.2.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>

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

@@ -9,7 +9,6 @@
 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 {
 
@@ -22,25 +21,17 @@ public <T> void post(final T event) {
         eventBus.onNext(event);
     }
 
-    public <T> Subscription subscribe(final EventObserver<T> observer) {
+    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);
             }
         };
 
-        Disposable disposable = eventBus.ofType(observer.getEventType()).distinct()
+        return eventBus.ofType(eventType).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

@@ -3,31 +3,7 @@
 
 package software.aws.toolkits.eclipse.amazonq.broker.api;
 
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-
+@FunctionalInterface
 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

@@ -9,8 +9,8 @@
 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.broker.api.Subscription;
 import software.aws.toolkits.eclipse.amazonq.controllers.AmazonQViewController;
 import software.aws.toolkits.eclipse.amazonq.lsp.auth.model.AuthState;
 import software.aws.toolkits.eclipse.amazonq.plugin.Activator;
@@ -23,7 +23,7 @@ public abstract class AmazonQView extends ViewPart implements EventObserver<Auth
     private AmazonQCommonActions amazonQCommonActions;
     private static final ThemeDetector THEME_DETECTOR = new ThemeDetector();
 
-    private Subscription authStateSubscription;
+    private Disposable authStateSubscription;
 
     protected AmazonQView() {
         this.viewController = new AmazonQViewController();
@@ -83,10 +83,10 @@ private void setupActions(final AuthState authState) {
     }
 
     private void setupAuthStatusListeners() {
-        authStateSubscription = Activator.getEventBroker().subscribe(this);
-        Activator.getEventBroker().subscribe(amazonQCommonActions.getSignoutAction());
-        Activator.getEventBroker().subscribe(amazonQCommonActions.getFeedbackDialogContributionAction());
-        Activator.getEventBroker().subscribe(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
@@ -125,7 +125,7 @@ function waitForFunction(functionName, timeout = 30000) {
      */
     @Override
     public void dispose() {
-        authStateSubscription.cancel();
+        authStateSubscription.dispose();
         super.dispose();
     }
 

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

@@ -18,8 +18,8 @@
 import org.eclipse.swt.widgets.Display;
 import org.eclipse.swt.widgets.Link;
 
+import io.reactivex.rxjava3.disposables.Disposable;
 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;
@@ -38,12 +38,12 @@ public final class ReauthenticateView extends CallToActionView implements EventO
     private static final String BUTTON_LABEL = "Re-authenticate";
     private static final String LINK_LABEL = "Sign out";
 
-    private Subscription authStateSubscription;
+    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.
-         authStateSubscription = Activator.getEventBroker().subscribe(this);
+         authStateSubscription = Activator.getEventBroker().subscribe(AuthState.class, this);
     }
 
     @Override
@@ -140,7 +140,7 @@ protected void updateButtonStyle(final Button button) {
 
     @Override
     public void dispose() {
-        authStateSubscription.cancel();
+        authStateSubscription.dispose();
     }
 
     @Override