[rd debugger] Add session events test

Fix a race between flow collection and addind a listener

GitOrigin-RevId: eab6cd7b487546d3b92f11c9a7df7a9f708947f8

java/testFramework/src/com/intellij/debugger/DebuggerTestCase.java

@@ -62,6 +62,7 @@ import javax.swing.*;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public abstract class DebuggerTestCase extends ExecutionWithDebuggerToolsTestCase {
   protected static final int DEFAULT_ADDRESS = 3456;
@@ -71,6 +72,7 @@ public abstract class DebuggerTestCase extends ExecutionWithDebuggerToolsTestCas
   private RunProfileState myRunnableState;
   private final List<ThrowableRunnable<Throwable>> myTearDownRunnables = new ArrayList<>();
   private CompilerManagerImpl myCompilerManager;
+  private final AtomicBoolean myProcessStarted = new AtomicBoolean();
 
   @Override
   protected void setUp() throws Exception {
@@ -134,7 +136,7 @@ public abstract class DebuggerTestCase extends ExecutionWithDebuggerToolsTestCas
   protected void runTestRunnable(@NotNull ThrowableRunnable<Throwable> testRunnable) throws Throwable {
     super.runTestRunnable(testRunnable);
     if (getDebugProcess() != null) {
-      getDebugProcess().getProcessHandler().startNotify();
+      runProcess();
       waitProcess(getDebugProcess().getProcessHandler());
       waitForCompleted();
       //disposeSession(myDebuggerSession);
@@ -146,6 +148,12 @@ public abstract class DebuggerTestCase extends ExecutionWithDebuggerToolsTestCas
     checkTestOutput();
   }
 
+  protected final void runProcess() {
+    if (getDebugProcess() != null && myProcessStarted.compareAndSet(false, true)) {
+      getDebugProcess().getProcessHandler().startNotify();
+    }
+  }
+
   /**
    * Ensures that the actual output from {@link #systemPrintln(String)} and the related methods
    * matches the expected output from the {@code .out} file.

platform/xdebugger-impl/backend/src/com/intellij/platform/debugger/impl/backend/BackendXDebuggerManagerApi.kt

@@ -61,7 +61,8 @@ internal class BackendXDebuggerManagerApi : XDebuggerManagerApi {
   override suspend fun sessions(projectId: ProjectId): XDebugSessionsList {
     val project = projectId.findProject()
     val sessions = XDebuggerManager.getInstance(project).debugSessions.map { createSessionDto(it as XDebugSessionImpl, it.debugProcess) }
-    return XDebugSessionsList(sessions, createSessionManagerEvents(projectId).toRpc())
+    val initialSessions = sessions.map { it.id }.toSet()
+    return XDebugSessionsList(sessions, createSessionManagerEvents(projectId, initialSessions).toRpc())
   }
 
   private suspend fun createSessionDto(currentSession: XDebugSessionImpl, debugProcess: XDebugProcess): XDebugSessionDto {
@@ -160,10 +161,10 @@ internal class BackendXDebuggerManagerApi : XDebuggerManagerApi {
   }.buffer(Channel.UNLIMITED)
 
   @OptIn(ExperimentalCoroutinesApi::class)
-  private fun createSessionManagerEvents(projectId: ProjectId): Flow<XDebuggerManagerSessionEvent> {
+  private fun createSessionManagerEvents(projectId: ProjectId, initialSessionIds: Set<XDebugSessionId>): Flow<XDebuggerManagerSessionEvent> {
     val project = projectId.findProject()
     return channelFlow {
-      project.messageBus.connect(this).subscribe(XDebuggerManager.TOPIC, object : XDebuggerManagerListener {
+      val listener = object : XDebuggerManagerListener {
         override fun processStarted(debugProcess: XDebugProcess) {
           val session = debugProcess.session as? XDebugSessionImpl ?: return
           launch {
@@ -182,7 +183,20 @@ internal class BackendXDebuggerManagerApi : XDebuggerManagerApi {
           val currentSessionId = (currentSession as? XDebugSessionImpl)?.id
           trySend(XDebuggerManagerSessionEvent.CurrentSessionChanged(previousSessionId, currentSessionId))
         }
-      })
+      }
+      project.messageBus.connect(this).subscribe(XDebuggerManager.TOPIC, listener)
+      val currentSessions = XDebuggerManager.getInstance(project).debugSessions.filterIsInstance<XDebugSessionImpl>().associateBy { it.id }
+      val newlyAddedSessions = currentSessions.keys - initialSessionIds
+      val completedSessions = initialSessionIds - currentSessions.keys
+
+      for (newlyAddedSession in newlyAddedSessions) {
+        val session = currentSessions[newlyAddedSession] ?: continue
+        listener.processStarted(session.debugProcess)
+      }
+      for (completedSession in completedSessions) {
+        send(XDebuggerManagerSessionEvent.ProcessStopped(completedSession))
+      }
+
       awaitClose()
     }.buffer(Channel.UNLIMITED)
   }