[JetBrains] Improve ports listener leak

Author: mustard-mh

components/ide/jetbrains/backend-plugin/src/main/kotlin/io/gitpod/jetbrains/remote/AbstractGitpodPortForwardingService.kt

@@ -13,16 +13,20 @@ import com.intellij.util.application
 import com.jetbrains.rd.platform.codeWithMe.portForwarding.*
 import com.jetbrains.rd.util.URI
 import com.jetbrains.rd.util.lifetime.Lifetime
+import fleet.util.async.throttleLatest
 import io.gitpod.supervisor.api.Status
 import io.gitpod.supervisor.api.Status.PortsStatus
 import io.gitpod.supervisor.api.StatusServiceGrpc
 import io.grpc.stub.ClientCallStreamObserver
 import io.grpc.stub.ClientResponseObserver
 import kotlinx.coroutines.*
 import kotlinx.coroutines.future.asDeferred
+import kotlinx.coroutines.flow.MutableSharedFlow
 import org.apache.http.client.utils.URIBuilder
 import java.util.*
 import java.util.concurrent.CompletableFuture
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
 
 @Suppress("UnstableApiUsage")
 abstract class AbstractGitpodPortForwardingService : GitpodPortForwardingService {
@@ -34,8 +38,22 @@ abstract class AbstractGitpodPortForwardingService : GitpodPortForwardingService
     private val perClientPortForwardingManager = service<PerClientPortForwardingManager>()
     private val ignoredPortsForNotificationService = service<GitpodIgnoredPortsForNotificationService>()
     private val lifetime = Lifetime.Eternal.createNested()
+    private val portStatusFlow = MutableSharedFlow<Status.PortsStatusResponse>()
+
+    init {
+        // Start collecting port status updates with throttling
+        runJob(lifetime) {
+            portStatusFlow
+                .throttleLatest(1000) // Throttle to 1 second
+                .collect { response ->
+                    withContext(Dispatchers.IO) {
+                        syncPortsListWithClient(response)
+                    }
+                }
+        }
 
-    init { start() }
+        start()
+    }
 
     private fun start() {
         if (application.isHeadlessEnvironment) return
@@ -86,7 +104,11 @@ abstract class AbstractGitpodPortForwardingService : GitpodPortForwardingService
             }
 
             override fun onNext(response: Status.PortsStatusResponse) {
-                application.invokeLater { syncPortsListWithClient(response) }
+                application.invokeLater {
+                    runJob(lifetime) {
+                        portStatusFlow.emit(response)
+                    }
+                }
             }
 
             override fun onCompleted() {