fix abort

Author: mustard-mh

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

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

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

@@ -733,29 +733,36 @@ export class WorkspaceService {
         return urls;
     }
 
-    public async *watchWorkspaces(userId: string) {
-        let resolveNext: (() => void) | null;
-        const next = () => new Promise<void>((res) => (resolveNext = res));
-        const nextOK = () => {
+    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();
+                resolveNext(d);
+                if (d) {
+                    queue.push(d);
+                }
                 resolveNext = null;
             }
         };
 
-        const queue: WorkspaceInstance[] = [];
-
-        const dispose = this.subscriber.listenForWorkspaceInstanceUpdates(userId, (_ctx, instance) => {
-            queue.push(instance);
+        let isStopped = false;
+        opts.signal.addEventListener("abort", () => {
+            isStopped = true;
             nextOK();
         });
+        const dispose = this.subscriber.listenForWorkspaceInstanceUpdates(userId, (_ctx, instance) => {
+            nextOK(instance);
+        });
 
         try {
-            while (true) {
+            while (!isStopped) {
                 if (queue.length) {
                     yield queue.shift();
                 } else {
-                    await next();
+                    yield next();
                 }
             }
         } catch (e) {