[dashboard] add json rpc implement

Author: mustard-mh

components/dashboard/src/service/json-rpc-workspace-client.ts

@@ -4,12 +4,19 @@
  * See License.AGPL.txt in the project root for license information.
  */
 
-import { Code, ConnectError, PromiseClient } from "@connectrpc/connect";
+import { CallOptions, Code, ConnectError, PromiseClient } from "@connectrpc/connect";
 import { PartialMessage } from "@bufbuild/protobuf";
 import { WorkspaceService } from "@gitpod/public-api/lib/gitpod/experimental/v2/workspace_connect";
-import { GetWorkspaceRequest, GetWorkspaceResponse } from "@gitpod/public-api/lib/gitpod/experimental/v2/workspace_pb";
+import {
+    GetWorkspaceRequest,
+    GetWorkspaceResponse,
+    WatchWorkspaceRequest,
+    WatchWorkspaceResponse,
+} from "@gitpod/public-api/lib/gitpod/experimental/v2/workspace_pb";
 import { converter } from "./public-api";
 import { getGitpodService } from "./service";
+import { generateAsyncGenerator } from "@gitpod/gitpod-protocol/lib/generate-async-generator";
+import { WorkspaceInstance } from "@gitpod/gitpod-protocol";
 
 export class JsonRpcWorkspaceClient implements PromiseClient<typeof WorkspaceService> {
     async getWorkspace(request: PartialMessage<GetWorkspaceRequest>): Promise<GetWorkspaceResponse> {
@@ -22,4 +29,35 @@ export class JsonRpcWorkspaceClient implements PromiseClient<typeof WorkspaceSer
         result.item = workspace;
         return result;
     }
+
+    async *watchWorkspace(
+        request: PartialMessage<WatchWorkspaceRequest>,
+        options?: CallOptions,
+    ): AsyncIterable<WatchWorkspaceResponse> {
+        if (!options?.signal) {
+            throw new ConnectError("signal is required", Code.InvalidArgument);
+        }
+        const it = generateAsyncGenerator<WorkspaceInstance>(
+            (sink) => {
+                const dispose = getGitpodService().registerClient({
+                    onInstanceUpdate: (instance) => {
+                        sink.next(instance);
+                    },
+                });
+                return dispose.dispose;
+            },
+            { signal: options.signal },
+        );
+        for await (const item of it) {
+            if (!item) {
+                continue;
+            }
+            if (request.id && item.workspaceId !== request.id) {
+                continue;
+            }
+            const response = new WatchWorkspaceResponse();
+            response.item = converter.toWorkspace(item);
+            yield response;
+        }
+    }
 }

components/gitpod-protocol/src/generate-async-generator.ts

@@ -0,0 +1,60 @@
+/**
+ * Copyright (c) 2023 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.
+ */
+
+type Sink<T> = {
+    next: (value: T | void) => void;
+};
+
+/**
+ * Generates an asynchronous generator that yields values based on the provided setup function.
+ *
+ * the setup function that takes a sink and returns a cleanup function.
+ * setup sink object has a `next` method that accepts a value to be pushed to the generator.
+ */
+export async function* generateAsyncGenerator<T>(
+    setup: (sink: Sink<T>) => () => void,
+    opts: { signal: AbortSignal },
+): AsyncGenerator<T | void, void, unknown> {
+    const queue: T[] = [];
+
+    let resolveNext: ((value: T | void) => void) | null = null;
+
+    const sink: Sink<T> = {
+        next: (value: T | void) => {
+            if (resolveNext) {
+                resolveNext(value);
+                resolveNext = null;
+            } else {
+                if (value) {
+                    queue.push(value);
+                }
+            }
+        },
+    };
+
+    let isStopped = false;
+    opts.signal.addEventListener("abort", () => {
+        isStopped = true;
+        sink.next();
+    });
+
+    const cleanup = setup(sink);
+
+    try {
+        while (!isStopped) {
+            if (queue.length) {
+                yield queue.shift();
+            } else {
+                yield new Promise<T | void>((resolve) => {
+                    resolveNext = resolve;
+                });
+            }
+        }
+    } catch (e) {
+    } finally {
+        cleanup();
+    }
+}

components/server/src/api/workspace-service-api.ts

@@ -34,9 +34,6 @@ export class WorkspaceServiceAPI implements ServiceImpl<typeof WorkspaceServiceI
 
     async *watchWorkspace(req: WatchWorkspaceRequest, context: HandlerContext): AsyncIterable<WatchWorkspaceResponse> {
         const it = this.workspaceService.watchWorkspaces(context.user.id, { signal: context.signal });
-        context.signal.addEventListener("abort", async () => {
-            await it.throw(new Error("abort")).then().catch();
-        });
         for await (const info of it) {
             if (!info) {
                 continue;

components/server/src/workspace/workspace-service.ts

@@ -30,6 +30,7 @@ import {
     WorkspaceTimeoutDuration,
 } from "@gitpod/gitpod-protocol";
 import { ErrorCodes, ApplicationError } from "@gitpod/gitpod-protocol/lib/messaging/error";
+import { generateAsyncGenerator } from "@gitpod/gitpod-protocol/lib/generate-async-generator";
 import { Authorizer } from "../authorization/authorizer";
 import { TraceContext } from "@gitpod/gitpod-protocol/lib/util/tracing";
 import { WorkspaceFactory } from "./workspace-factory";
@@ -733,42 +734,13 @@ export class WorkspaceService {
         return urls;
     }
 
-    public async *watchWorkspaces(userId: string, opts: { signal: AbortSignal }) {
-        const queue: WorkspaceInstance[] = [];
-        let resolveNext: ((d: WorkspaceInstance | void) => void) | null;
-
-        const next = () => new Promise<WorkspaceInstance | void>((res) => (resolveNext = res));
-        const nextOK = (d: WorkspaceInstance | void) => {
-            if (resolveNext) {
-                resolveNext(d);
-                if (d) {
-                    queue.push(d);
-                }
-                resolveNext = null;
-            }
-        };
-
-        let isStopped = false;
-        opts.signal.addEventListener("abort", () => {
-            isStopped = true;
-            nextOK();
-        });
-        const dispose = this.subscriber.listenForWorkspaceInstanceUpdates(userId, (_ctx, instance) => {
-            nextOK(instance);
-        });
-
-        try {
-            while (!isStopped) {
-                if (queue.length) {
-                    yield queue.shift();
-                } else {
-                    yield next();
-                }
-            }
-        } catch (e) {
-        } finally {
-            dispose.dispose();
-        }
+    public watchWorkspaces(userId: string, opts: { signal: AbortSignal }) {
+        return generateAsyncGenerator<WorkspaceInstance>((sink) => {
+            const dispose = this.subscriber.listenForWorkspaceInstanceUpdates(userId, (_ctx, instance) => {
+                sink.next(instance);
+            });
+            return dispose.dispose;
+        }, opts);
     }
 
     public async watchWorkspaceImageBuildLogs(