refactor: setup better testing infra

Author: John Doe

packages/utils/src/lib/performance-observer.int.test.ts

@@ -0,0 +1,117 @@
+import { type PerformanceEntry, performance } from 'node:perf_hooks';
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  type PerformanceObserverOptions,
+  PerformanceObserverSink,
+} from './performance-observer.js';
+import type { Sink } from './sink-source.types';
+
+class MockSink implements Sink<string, string> {
+  private writtenItems: string[] = [];
+  private closed = false;
+
+  open(): void {
+    this.closed = false;
+  }
+
+  write(input: string): void {
+    this.writtenItems.push(input);
+  }
+
+  close(): void {
+    this.closed = true;
+  }
+
+  isClosed(): boolean {
+    return this.closed;
+  }
+
+  encode(input: string): string {
+    return `${input}-${this.constructor.name}-encoded`;
+  }
+
+  recover(): string[] {
+    return [...this.writtenItems];
+  }
+}
+
+describe('PerformanceObserverSink', () => {
+  let sink: MockSink;
+  let options: PerformanceObserverOptions<string>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    performance.clearMeasures();
+    performance.clearMarks();
+    sink = new MockSink();
+
+    options = {
+      sink,
+      encode: vi.fn((entry: PerformanceEntry) => [
+        `${entry.name}:${entry.entryType}`,
+      ]),
+    };
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it('creates instance with default options', () => {
+    expect(() => new PerformanceObserverSink(options)).not.toThrow();
+  });
+
+  it('creates instance with custom options', () => {
+    expect(
+      () =>
+        new PerformanceObserverSink({
+          ...options,
+          buffered: true,
+          flushThreshold: 10,
+        }),
+    ).not.toThrow();
+  });
+
+  it('should observe performance entries and write them to the sink on flush', () => {
+    const observer = new PerformanceObserverSink(options);
+
+    observer.subscribe();
+    performance.mark('test-mark');
+    observer.flush();
+    expect(sink.recover()).toHaveLength(1);
+  });
+
+  it('should observe buffered performance entries when buffered is enabled', async () => {
+    const observer = new PerformanceObserverSink({
+      ...options,
+      buffered: true,
+    });
+
+    performance.mark('test-mark-1');
+    performance.mark('test-mark-2');
+    await new Promise(resolve => setTimeout(resolve, 10));
+    observer.subscribe();
+    await new Promise(resolve => setTimeout(resolve, 10));
+    expect(performance.getEntries()).toHaveLength(2);
+    observer.flush();
+    expect(sink.recover()).toHaveLength(2);
+  });
+
+  it('handles multiple encoded items per performance entry', () => {
+    const multiEncodeFn = vi.fn(e => [
+      `${e.entryType}-item1`,
+      `${e.entryType}item2`,
+    ]);
+    const observer = new PerformanceObserverSink({
+      ...options,
+      encode: multiEncodeFn,
+    });
+
+    observer.subscribe();
+
+    performance.mark('test-mark');
+    observer.flush();
+
+    expect(sink.recover()).toHaveLength(2);
+  });
+});

packages/utils/src/lib/performance-observer.ts

@@ -1,107 +1,88 @@
 import {
+  type EntryType,
   type PerformanceEntry,
   PerformanceObserver,
+  type PerformanceObserverEntryList,
   performance,
 } from 'node:perf_hooks';
-import type { Buffered, Encoder, Sink } from './sink-source.types.js';
+import type { Buffered, Encoder, Observer, Sink } from './sink-source.types.js';
 
 export const DEFAULT_FLUSH_THRESHOLD = 20;
 
 export type PerformanceObserverOptions<T> = {
   sink: Sink<T, unknown>;
   encode: (entry: PerformanceEntry) => T[];
-  captureBuffered?: boolean;
+  buffered?: boolean;
   flushThreshold?: number;
 };
 
-export class PerformanceObserverHandle<T>
-  implements Buffered, Encoder<PerformanceEntry, T[]>
+export class PerformanceObserverSink<T>
+  implements Observer, Buffered, Encoder<PerformanceEntry, T[]>
 {
   #encode: (entry: PerformanceEntry) => T[];
-  #captureBuffered: boolean;
-  #observedEntryCount: number;
+  #buffered: boolean;
   #flushThreshold: number;
   #sink: Sink<T, unknown>;
   #observer: PerformanceObserver | undefined;
-  #closed = false;
+  #observedTypes: EntryType[] = ['mark', 'measure'];
+  #getEntries = (list: PerformanceObserverEntryList) =>
+    this.#observedTypes.flatMap(t => list.getEntriesByType(t));
+  #observedCount: number = 0;
 
   constructor(options: PerformanceObserverOptions<T>) {
     this.#encode = options.encode;
     this.#sink = options.sink;
-    this.#captureBuffered = options.captureBuffered ?? false;
+    this.#buffered = options.buffered ?? false;
     this.#flushThreshold = options.flushThreshold ?? DEFAULT_FLUSH_THRESHOLD;
-    this.#observedEntryCount = 0;
   }
 
   encode(entry: PerformanceEntry): T[] {
     return this.#encode(entry);
   }
 
-  connect(): void {
-    if (this.#observer || this.#closed) {
+  subscribe(): void {
+    if (this.#observer) {
       return;
     }
-    this.#observer = new PerformanceObserver(() => {
-      this.#observedEntryCount++;
-      if (this.#observedEntryCount >= this.#flushThreshold) {
+
+    this.#observer = new PerformanceObserver(list => {
+      const entries = this.#getEntries(list);
+      this.#observedCount += entries.length;
+      if (this.#observedCount >= this.#flushThreshold) {
         this.flush();
-        this.#observedEntryCount = 0;
       }
     });
 
     this.#observer.observe({
-      entryTypes: ['mark', 'measure'],
-      buffered: this.#captureBuffered,
+      entryTypes: this.#observedTypes,
+      buffered: this.#buffered,
     });
   }
 
-  flush(clear = false): void {
-    if (this.#closed || !this.#sink) {
+  flush(): void {
+    if (!this.#observer) {
       return;
     }
-    const entries = [
-      ...performance.getEntriesByType('mark'),
-      ...performance.getEntriesByType('measure'),
-    ];
 
-    // Process all entries
-    entries
-      .filter(e => e.entryType === 'mark' || e.entryType === 'measure')
-      .forEach(e => {
-        const encoded = this.encode(e);
-        encoded.forEach(item => {
-          this.#sink.write(item);
-        });
-
-        if (clear) {
-          if (e.entryType === 'mark') {
-            performance.clearMarks(e.name);
-          }
-          if (e.entryType === 'measure') {
-            performance.clearMeasures(e.name);
-          }
-        }
+    const entries = this.#getEntries(performance);
+    entries.forEach(entry => {
+      const encoded = this.encode(entry);
+      encoded.forEach(item => {
+        this.#sink.write(item);
       });
+    });
+    this.#observedCount = 0;
   }
 
-  disconnect(): void {
+  unsubscribe(): void {
     if (!this.#observer) {
       return;
     }
     this.#observer?.disconnect();
     this.#observer = undefined;
   }
 
-  close(): void {
-    if (this.#closed) {
-      return;
-    }
-    this.flush();
-    this.#closed = true;
-    this.disconnect();
-  }
-
-  isConnected(): boolean {
-    return this.#observer !== undefined && !this.#closed;
+  isSubscribed(): boolean {
+    return this.#observer !== undefined;
   }
 }