feat(jb): observe ports status and send notification

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

@@ -4,6 +4,7 @@
 
 package io.gitpod.jetbrains.remote
 
+import com.intellij.ide.BrowserUtil
 import com.intellij.ide.plugins.PluginManagerCore
 import com.intellij.notification.NotificationAction
 import com.intellij.notification.NotificationGroupManager
@@ -23,6 +24,8 @@ import io.gitpod.jetbrains.remote.utils.Retrier.retry
 import io.gitpod.supervisor.api.*
 import io.gitpod.supervisor.api.Info.WorkspaceInfoResponse
 import io.gitpod.supervisor.api.Notification.*
+import io.gitpod.supervisor.api.Status.PortsStatusRequest
+import io.gitpod.supervisor.api.Status.PortsStatusResponse
 import io.grpc.ManagedChannel
 import io.grpc.ManagedChannelBuilder
 import io.grpc.stub.ClientCallStreamObserver
@@ -46,6 +49,7 @@ import java.util.concurrent.CancellationException
 import java.util.concurrent.CompletableFuture
 import javax.websocket.DeploymentException
 
+
 @Service
 class GitpodManager : Disposable {
 
@@ -57,6 +61,7 @@ class GitpodManager : Disposable {
     val devMode = System.getenv("JB_DEV").toBoolean()
     private val backendKind = System.getenv("JETBRAINS_GITPOD_BACKEND_KIND") ?: "unknown"
     private val backendQualifier = System.getenv("JETBRAINS_BACKEND_QUALIFIER") ?: "unknown"
+    private val workspaceUrl = System.getenv("GITPOD_WORKSPACE_URL")
 
     private val lifetime = Lifetime.Eternal.createNested()
 
@@ -199,6 +204,71 @@ class GitpodManager : Disposable {
         }
     }
 
+    private val portsObserveJob = GlobalScope.launch {
+        if (application.isHeadlessEnvironment) {
+            return@launch
+        }
+
+        val status = StatusServiceGrpc.newStub(supervisorChannel)
+//        while (isActive) {
+        try {
+            val f = CompletableFuture<Void>()
+            status.portsStatus(
+                    PortsStatusRequest.newBuilder().build(),
+                    object : ClientResponseObserver<PortsStatusRequest, PortsStatusResponse> {
+                        override fun beforeStart(requestStream: ClientCallStreamObserver<PortsStatusRequest>) {
+                            println("[Andrea]: beforeStart")
+                            // TODO(ak): actually should be bound to cancellation of notifications job
+                            lifetime.onTerminationOrNow {
+                                requestStream.cancel(null, null)
+                            }
+                        }
+
+                        override fun onNext(ps: PortsStatusResponse) {
+                            println("[Andrea]: onNext")
+                            for (port in ps.portsList) {
+                                println(port)
+                                if (port.exposed.onExposed.number == 0) {
+                                    return
+                                }
+
+                                println(port)
+
+                                if (port.served && port.exposed.url.isNullOrEmpty() == false) {
+                                    val message = "Your application running on port " + port.localPort + " is available."
+                                    val notification = notificationGroup.createNotification(message, NotificationType.INFORMATION)
+                                    // TODO(andreafalzetti): add analytics event
+                                    val lambda = { BrowserUtil.browse(port.exposed.url) }
+                                    val action = NotificationAction.createSimpleExpiring("Open in browser", lambda)
+                                    notification.addAction(action)
+                                    notification.notify(null)
+                                }
+                            }
+
+                        }
+
+                        override fun onError(t: Throwable) {
+                            println("[Andrea]: onError")
+                            f.completeExceptionally(t)
+                        }
+
+                        override fun onCompleted() {
+                            f.complete(null)
+                        }
+                    })
+        } catch (t: Throwable) {
+            if (t is CancellationException) {
+                throw t
+            }
+            thisLogger().error("gitpod: failed to stream ports status: ", t)
+        }
+    }
+    init {
+        lifetime.onTerminationOrNow {
+            portsObserveJob.cancel()
+        }
+    }
+
     val pendingInfo = CompletableFuture<WorkspaceInfoResponse>()
     private val infoJob = GlobalScope.launch {
         if (application.isHeadlessEnvironment) {