WIP: [JetBrains] Improve ports listener leaking

components/ide/jetbrains/backend-plugin/build.gradle-stable.kts

@@ -38,6 +38,7 @@ project(":") {
         if (properties("platformType") == "RD") {
             print("Rider: exclude unnecessary files")
             sourceSets["main"].kotlin.exclude("**/GitpodForceUpdateMavenProjectsActivity.kt")
+            sourceSets["main"].kotlin.exclude("**/GitpodPortForwardingServiceImpl.kt")
             sourceSets["main"].kotlin.exclude("**/maven.xml")
         }
     }

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

@@ -20,9 +20,12 @@ 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 +37,26 @@ 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
+        runJob(lifetime) {
+            portStatusFlow
+                .let { flow -> applyThrottling(flow) }
+                .collect { response ->
+                    withContext(Dispatchers.IO) {
+                        syncPortsListWithClient(response)
+                    }
+                }
+        }
 
-    init { start() }
+        start()
+    }
+
+    protected abstract fun runJob(lifetime: Lifetime, block: suspend CoroutineScope.() -> Unit): Job;
+
+    protected abstract suspend fun <T> applyThrottling(flow: kotlinx.coroutines.flow.Flow<T>): kotlinx.coroutines.flow.Flow<T>
 
     private fun start() {
         if (application.isHeadlessEnvironment) return
@@ -47,8 +68,6 @@ abstract class AbstractGitpodPortForwardingService : GitpodPortForwardingService
         observePortsListWhileProjectIsOpen()
     }
 
-    protected abstract fun runJob(lifetime: Lifetime, block: suspend CoroutineScope.() -> Unit): Job;
-
     private fun observePortsListWhileProjectIsOpen() = runJob(lifetime) {
         while (isActive) {
             try {
@@ -86,7 +105,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() {

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

@@ -6,10 +6,13 @@ package io.gitpod.jetbrains.remote
 
 import com.jetbrains.rd.util.lifetime.Lifetime
 import com.jetbrains.rd.util.threading.coroutines.launch
-import io.gitpod.jetbrains.remote.AbstractGitpodPortForwardingService
 import kotlinx.coroutines.CoroutineScope
+import fleet.util.async.throttleLatest
+import kotlinx.coroutines.flow.Flow
 
 @Suppress("UnstableApiUsage")
 class GitpodPortForwardingServiceImpl : AbstractGitpodPortForwardingService() {
     override fun runJob(lifetime: Lifetime, block: suspend CoroutineScope.() -> Unit) = lifetime.launch { block() }
+
+    override suspend fun <T> applyThrottling(flow: Flow<T>): Flow<T> = flow.throttleLatest(1000)
 }

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

@@ -0,0 +1,18 @@
+// Copyright (c) 2025 Gitpod GmbH. All rights reserved.
+// Licensed under the GNU Affero General Public License (AGPL).
+// See License.AGPL.txt in the project root for license information.
+
+package io.gitpod.jetbrains.remote.optional
+
+import com.jetbrains.rd.util.lifetime.Lifetime
+import com.jetbrains.rd.util.threading.coroutines.launch
+import io.gitpod.jetbrains.remote.AbstractGitpodPortForwardingService
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
+
+@Suppress("UnstableApiUsage")
+class GitpodRiderPortForwardingService : AbstractGitpodPortForwardingService() {
+    override fun runJob(lifetime: Lifetime, block: suspend CoroutineScope.() -> Unit) = lifetime.launch { block() }
+
+    override suspend fun <T> applyThrottling(flow: Flow<T>): Flow<T> = flow
+}

components/ide/jetbrains/backend-plugin/src/main/resources-latest/META-INF/extensions.xml

@@ -6,5 +6,8 @@
 <!--suppress PluginXmlValidity -->
 <idea-plugin>
     <extensions defaultExtensionNs="com.intellij">
+        <applicationService serviceInterface="io.gitpod.jetbrains.remote.GitpodPortForwardingService"
+                            serviceImplementation="io.gitpod.jetbrains.remote.GitpodPortForwardingServiceImpl"
+                            client="controller" preload="true"/>
     </extensions>
 </idea-plugin>

components/ide/jetbrains/backend-plugin/src/main/resources-rider/META-INF/plugin.xml

@@ -50,6 +50,9 @@
                      description="Disable the forced update of Maven projects when IDE opens."
                      restartRequired="true"/>
 
+        <applicationService serviceInterface="io.gitpod.jetbrains.remote.GitpodPortForwardingService"
+                            serviceImplementation="io.gitpod.jetbrains.remote.optional.GitpodRiderPortForwardingService"
+                            client="controller" preload="true" />
 
         <applicationService serviceInterface="io.gitpod.jetbrains.remote.GitpodIgnoredPortsForNotificationService"
                             serviceImplementation="io.gitpod.jetbrains.remote.internal.GitpodIgnoredPortsForNotificationServiceImpl"

components/ide/jetbrains/backend-plugin/src/main/resources-stable/META-INF/extensions.xml

@@ -6,5 +6,8 @@
 <!--suppress PluginXmlValidity -->
 <idea-plugin>
     <extensions defaultExtensionNs="com.intellij">
+        <applicationService serviceInterface="io.gitpod.jetbrains.remote.GitpodPortForwardingService"
+                            serviceImplementation="io.gitpod.jetbrains.remote.GitpodPortForwardingServiceImpl"
+                            client="controller" preload="true"/>
     </extensions>
 </idea-plugin>

components/ide/jetbrains/backend-plugin/src/main/resources/META-INF/plugin.xml

@@ -55,9 +55,6 @@
                             preload="true"/>
 
 
-        <applicationService serviceInterface="io.gitpod.jetbrains.remote.GitpodPortForwardingService"
-                            serviceImplementation="io.gitpod.jetbrains.remote.GitpodPortForwardingServiceImpl"
-                            client="controller" preload="true"/>
         <gateway.customization.performance id="gitpodMetricsControl" order="before cpuControl"
                                            implementation="io.gitpod.jetbrains.remote.GitpodMetricControlProvider"/>
     </extensions>