Use SSE to track CRC status change and logs

Signed-off-by: Yevhen Vydolob <yvydolob@redhat.com>

Author: evidolob

src/constants.ts

@@ -18,6 +18,8 @@
 
 import type { Preset } from './types';
 
+export const PRE_SSE_VERSION = '2.30.0';
+
 // copied from https://github.com/crc-org/crc/blob/632676d7c9ba0c030736c3d914984c4e140c1bf5/pkg/crc/constants/constants.go#L198
 
 export function getDefaultCPUs(preset: Preset): number {

src/crc-delete.ts

@@ -33,7 +33,7 @@ export function registerDeleteCommand(): void {
 }
 
 export async function deleteCrc(suppressNotification = false): Promise<boolean> {
-  if (crcStatus.status.CrcStatus === 'No Cluster') {
+  if (crcStatus.status.CrcStatus === 'NoVM') {
     if (!suppressNotification) {
       await extensionApi.window.showNotification({
         silent: false,

src/crc-status.ts

@@ -18,17 +18,23 @@
 
 import * as extensionApi from '@podman-desktop/api';
 import type { Status, CrcStatus as CrcStatusApi } from './types';
-import { commander } from './daemon-commander';
+import { commander, getCrcApiUrl } from './daemon-commander';
 import { isNeedSetup } from './crc-setup';
+import { EventSource } from './events/eventsource';
+import type { CrcVersion } from './crc-cli';
+import { compare } from 'compare-versions';
+import { PRE_SSE_VERSION } from './constants';
 
 const defaultStatus: Status = { CrcStatus: 'Unknown', Preset: 'openshift' };
 const setupStatus: Status = { CrcStatus: 'Need Setup', Preset: 'Unknown' };
 const errorStatus: Status = { CrcStatus: 'Error', Preset: 'Unknown' };
 
 export class CrcStatus {
-  private updateTimer: NodeJS.Timer;
   private _status: Status;
   private isSetupGoing: boolean;
+  private statusEventSource: EventSource;
+  private crcVersion: CrcVersion;
+  private updateTimer: NodeJS.Timer;
   private statusChangeEventEmitter = new extensionApi.EventEmitter<Status>();
   public readonly onStatusChange = this.statusChangeEventEmitter.event;
 
@@ -37,30 +43,61 @@ export class CrcStatus {
   }
 
   startStatusUpdate(): void {
-    if (this.updateTimer) {
-      return; // we already set timer
-    }
-    this.updateTimer = setInterval(async () => {
-      try {
-        // we don't need to update status while setup is going
-        if (this.isSetupGoing) {
-          this._status = createStatus('Starting', this._status.Preset);
-          return;
+    if (compare(this.crcVersion.version, PRE_SSE_VERSION, '<=')) {
+      if (this.updateTimer) {
+        return; // we already set timer
+      }
+      this.updateTimer = setInterval(async () => {
+        try {
+          // we don't need to update status while setup is going
+          if (this.isSetupGoing) {
+            this._status = createStatus('Starting', this._status.Preset);
+            return;
+          }
+          const oldStatus = this._status;
+          this._status = await commander.status();
+          // notify listeners when status changed
+          if (oldStatus.CrcStatus !== this._status.CrcStatus) {
+            this.statusChangeEventEmitter.fire(this._status);
+          }
+        } catch (e) {
+          console.error('CRC Status tick: ' + e);
+          this._status = defaultStatus;
         }
-        const oldStatus = this._status;
-        this._status = await commander.status();
-        // notify listeners when status changed
-        if (oldStatus.CrcStatus !== this._status.CrcStatus) {
+      }, 2000);
+    } else {
+      if (this.statusEventSource) {
+        return;
+      }
+
+      this.statusEventSource = new EventSource(getCrcApiUrl() + '/events?stream=status_change');
+      this.statusEventSource.on('status_change', (e: MessageEvent) => {
+        const data = e.data;
+        try {
+          if (this.isSetupGoing) {
+            this._status = createStatus('Starting', this._status.Preset);
+            return;
+          }
+          console.error(`On Status: ${data}`);
+          this._status = JSON.parse(data).status;
           this.statusChangeEventEmitter.fire(this._status);
+        } catch (err) {
+          console.error(err);
+          this._status = defaultStatus;
         }
-      } catch (e) {
-        console.error('CRC Status tick: ' + e);
+      });
+      this.statusEventSource.on('error', e => {
+        console.error(e);
         this._status = defaultStatus;
-      }
-    }, 2000);
+      });
+    }
   }
 
   stopStatusUpdate(): void {
+    if (this.statusEventSource) {
+      this.statusEventSource.close();
+      this.statusEventSource = undefined;
+    }
     if (this.updateTimer) {
       clearInterval(this.updateTimer);
     }
@@ -74,7 +111,8 @@ export class CrcStatus {
     this._status = errorStatus;
   }
 
-  async initialize(): Promise<void> {
+  async initialize(crcVersion: CrcVersion): Promise<void> {
+    this.crcVersion = crcVersion;
     if (isNeedSetup) {
       this._status = setupStatus;
       return;
@@ -107,7 +145,7 @@ export class CrcStatus {
       case 'Stopping':
         return 'stopping';
       case 'Stopped':
-      case 'No Cluster':
+      case 'NoVM':
         return 'stopped';
       default:
         return 'unknown';
@@ -124,7 +162,7 @@ export class CrcStatus {
         return 'stopping';
       case 'Stopped':
         return 'configured';
-      case 'No Cluster':
+      case 'NoVM':
         return 'installed';
       case 'Error':
         return 'error';

src/daemon-commander.ts

@@ -21,15 +21,18 @@ import got from 'got';
 import { isWindows } from './util';
 import type { ConfigKeys, Configuration, StartInfo, Status } from './types';
 
+export function getCrcApiUrl(): string {
+  if (isWindows()) {
+    return 'http://unix://?/pipe/crc-http:';
+  }
+  return `http://unix:${process.env.HOME}/.crc/crc-http.sock:`;
+}
+
 export class DaemonCommander {
   private apiPath: string;
 
   constructor() {
-    this.apiPath = `http://unix:${process.env.HOME}/.crc/crc-http.sock:/api`;
-
-    if (isWindows()) {
-      this.apiPath = 'http://unix://?/pipe/crc-http:/api';
-    }
+    this.apiPath = getCrcApiUrl() + '/api';
   }
 
   async status(): Promise<Status> {
@@ -41,7 +44,7 @@ export class DaemonCommander {
     } catch (error) {
       // ignore status error, as it may happen when no cluster created
       return {
-        CrcStatus: 'No Cluster',
+        CrcStatus: 'NoVM',
       };
     }
   }

src/events/buffered-reader.ts

@@ -0,0 +1,119 @@
+/**********************************************************************
+ * Copyright (C) 2023 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ***********************************************************************/
+
+export interface LineConsumer {
+  (buf: Buffer, pos: number, fieldLength: number, lineLength: number): void;
+}
+
+const maxBufferAheadAllocation = 1024 * 256;
+const bom = [239, 187, 191];
+const colon = 58;
+const lineFeed = 10;
+const carriageReturn = 13;
+
+export class BufferedReader {
+  private buffer: Buffer;
+  private bytesUsed = 0;
+  private discardTrailingNewline = false;
+  private startingFieldLength = -1;
+  private startingPos = 0;
+
+  constructor(private readonly lineConsumer: LineConsumer) {}
+
+  onData(chunk: Buffer): void {
+    let newBuffer: Buffer;
+    let newBufferSize = 0;
+
+    if (!this.buffer) {
+      this.buffer = chunk;
+      if (hasBom(this.buffer)) {
+        this.buffer = this.buffer.subarray(bom.length);
+      }
+      this.bytesUsed = this.buffer.length;
+    } else {
+      if (chunk.length > this.buffer.length - this.bytesUsed) {
+        newBufferSize = this.buffer.length * 2 + chunk.length;
+        if (newBufferSize > maxBufferAheadAllocation) {
+          newBufferSize = this.buffer.length + chunk.length + maxBufferAheadAllocation;
+        }
+        newBuffer = Buffer.alloc(newBufferSize);
+        this.buffer.copy(newBuffer, 0, 0, this.bytesUsed);
+        this.buffer = newBuffer;
+      }
+      chunk.copy(this.buffer, this.bytesUsed);
+      this.bytesUsed += chunk.length;
+    }
+
+    let pos = 0;
+    const length = this.bytesUsed;
+
+    while (pos < length) {
+      if (this.discardTrailingNewline) {
+        if (this.buffer[pos] === lineFeed) {
+          ++pos;
+        }
+        this.discardTrailingNewline = false;
+      }
+
+      let lineLength = -1;
+      let fieldLength = this.startingFieldLength;
+      let c: number;
+
+      for (let i = this.startingPos; lineLength < 0 && i < length; ++i) {
+        c = this.buffer[i];
+        if (c === colon) {
+          if (fieldLength < 0) {
+            fieldLength = i - pos;
+          }
+        } else if (c === carriageReturn) {
+          this.discardTrailingNewline = true;
+          lineLength = i - pos;
+        } else if (c === lineFeed) {
+          lineLength = i - pos;
+        }
+      }
+
+      if (lineLength < 0) {
+        this.startingPos = length - pos;
+        this.startingFieldLength = fieldLength;
+        break;
+      } else {
+        this.startingPos = 0;
+        this.startingFieldLength = -1;
+      }
+
+      this.lineConsumer(this.buffer, pos, fieldLength, lineLength);
+
+      pos += lineLength + 1;
+    }
+
+    if (pos === length) {
+      this.buffer = void 0;
+      this.bytesUsed = 0;
+    } else if (pos > 0) {
+      this.buffer = this.buffer.subarray(pos, this.bytesUsed);
+      this.bytesUsed = this.buffer.length;
+    }
+  }
+}
+
+function hasBom(buffer: Buffer): boolean {
+  return bom.every((charCode, index) => {
+    return buffer[index] === charCode;
+  });
+}