Add watch stream structs/transforming

Generified so that the caller can supply their own MetaV1.Status
implementation.

Author: danopia

common.ts

@@ -32,8 +32,23 @@ export interface RestClient {
   defaultNamespace?: string;
 }
 
-// Things that JSON can encode directly
+// Structures that JSON can encode directly
 export type JSONPrimitive = string | number | boolean | null | undefined;
 export type JSONValue = JSONPrimitive | JSONObject | JSONArray;
 export type JSONObject = {[key: string]: JSONValue};
 export type JSONArray = JSONValue[];
+
+// Types seen in a resource watch stream
+export type WatchEvent<T,U> = WatchEventObject<T> | WatchEventError<U> | WatchEventBookmark;
+export type WatchEventObject<T> = {
+  'type': "ADDED" | "MODIFIED" | "DELETED";
+  'object': T;
+};
+export type WatchEventError<U> = {
+  'type': "ERROR";
+  'object': U;
+};
+export type WatchEventBookmark = {
+  'type': "BOOKMARK";
+  'object': { metadata: { resourceVersion: string }};
+};

mod.ts

@@ -18,13 +18,16 @@ export { InClusterRestClient, KubectlRawRestClient };
 
 import type { RestClient } from './common.ts';
 
-// Feeble attempt at automatically deciding how to talk to Kubernetes
-// You'll still need to set the correct permissions for where you are running.
-// You can probably be more specific and secure with app-specific Deno.args flags
+/**
+ * Trial-and-error approach for automatically deciding how to talk to Kubernetes.
+ * You'll still need to set the correct permissions for where you are running.
+ * You can probably be more specific and secure with app-specific Deno.args flags.
+ */
 export async function autoDetectClient(): Promise<RestClient> {
 
   // try reading the incluster service account files
   try {
+    // TODO: this should be async
     return new InClusterRestClient();
   } catch (err) {
     console.log('debug: InCluster client failed:', err.name);
@@ -33,5 +36,24 @@ export async function autoDetectClient(): Promise<RestClient> {
   // TODO: try hitting localhost:9001 (for KubectlProxyRestClient)
 
   // fall back to execing kubectl
+  // TODO: try execing first (probably to select default namespace)
   return new KubectlRawRestClient();
 }
+
+
+/** Paginates through an API request, yielding each successive page as a whole */
+export async function* readAllPages<T, U extends {continue?: string | null}>(pageFunc: (token?: string) => Promise<{metadata: U, items: T[]}>) {
+  let pageToken: string | undefined;
+  do {
+    const page = await pageFunc(pageToken ?? undefined);
+    yield page;
+    pageToken = page.metadata.continue ?? undefined;
+  } while (pageToken);
+}
+
+/** Paginates through an API request, yielding every individual item returned */
+export async function* readAllItems<T>(pageFunc: (token?: string) => Promise<{metadata: {continue?: string | null}, items: T[]}>) {
+  for await (const page of readAllPages(pageFunc)) {
+    yield* page.items;
+  }
+}

stream-transformers.ts

@@ -1,4 +1,4 @@
-import { JSONObject } from "./common.ts";
+import { JSONObject, WatchEvent } from "./common.ts";
 
 class TextLineReader {
   decoder: TextDecoder;
@@ -105,3 +105,59 @@ export function readableStreamFromReaderCloser(
     () => source.close(),
   );
 }
+
+
+class WatchEventReader<T,U> {
+  objValidator: (val: JSONObject) => T;
+  errValidator: (val: JSONObject) => U;
+  constructor(objValidator: (val: JSONObject) => T, errValidator: (val: JSONObject) => U) {
+    this.objValidator = objValidator;
+    this.errValidator = errValidator;
+  }
+  processObject(raw: JSONObject, controller: TransformStreamDefaultController<WatchEvent<T,U>>) {
+    const {type, object} = raw;
+    if (typeof type !== 'string') {
+      throw new Error(`BUG: watch record 'type' field was ${typeof type}`);
+    }
+    if (object == null) {
+      throw new Error(`BUG: watch record 'object' field was null`);
+    }
+    if (typeof object !== 'object') {
+      throw new Error(`BUG: watch record 'object' field was ${typeof object}`);
+    }
+    if (Array.isArray(object)) {
+      throw new Error(`BUG: watch record 'object' field was Array`);
+    }
+
+    switch (type) {
+      case 'ERROR':
+        controller.enqueue({type, object: this.errValidator(object)});
+        break;
+      case 'ADDED':
+      case 'MODIFIED':
+      case 'DELETED':
+        controller.enqueue({type, object: this.objValidator(object)});
+        break;
+      case 'BOOKMARK':
+        if (object.metadata && typeof object.metadata === 'object' && !Array.isArray(object.metadata)) {
+          if (typeof object.metadata.resourceVersion === 'string') {
+            controller.enqueue({type, object: {
+              metadata: { resourceVersion: object.metadata.resourceVersion },
+            }});
+            break;
+          }
+        }
+        throw new Error(`BUG: BOOKMARK event wasn't recognizable: ${JSON.stringify(object)}`);
+      default:
+        throw new Error(`BUG: watch record got unknown event type ${type}`);
+    }
+  }
+}
+
+/** Validates JSON objects belonging to a watch stream */
+export class WatchEventTransformer<T,U> extends TransformStream<JSONObject, WatchEvent<T,U>> {
+  constructor(objValidator: (val: JSONObject) => T, errValidator: (val: JSONObject) => U) {
+    const reader = new WatchEventReader(objValidator, errValidator);
+    super({ transform: reader.processObject.bind(reader) });
+  }
+}