feat: Devtools - A separate, lightweight event bus -- utilizing Dispatch (Sync)

dev/index.html

@@ -154,7 +154,11 @@
   }
 
   function setQueryLog() {
-    vg.coordinator().manager.logQueries(qlogToggle.checked);
+    // logQueries is no longer in use
+    // vg.coordinator().manager.logQueries(qlogToggle.checked);
+    // 
+    // 
+    // Add custom observer here that redirects logs to concolse
   }
 
   function setCache() {

packages/mosaic/core/src/Coordinator.ts

@@ -11,6 +11,8 @@ import { type MosaicClient } from './MosaicClient.js';
 import { type SelectionClause } from './SelectionClause.js';
 import { MaybeArray } from '@uwdata/mosaic-sql';
 import { Table } from '@uwdata/flechette';
+import { EventType } from './Events.js';
+import { ObserveDispatch } from './util/ObserveDispatch.js';
 
 interface FilterGroupEntry {
   selection: Selection;
@@ -50,6 +52,7 @@ export class Coordinator {
   public clients = new Set<MosaicClient>;
   public filterGroups = new Map<Selection, FilterGroupEntry>;
   protected _logger: Logger = voidLogger();
+  public eventBus: ObserveDispatch<any>; // temporary measures until we finalize event types
 
   /**
    * @param db Database connector. Defaults to a web socket connection.
@@ -77,7 +80,9 @@ export class Coordinator {
       consolidate = true,
       preagg = {}
     } = options;
+    this.eventBus = new ObserveDispatch();
     this.manager = manager;
+    this.manager.eventBus = this.eventBus;
     this.manager.cache(cache);
     this.manager.consolidate(consolidate);
     this.databaseConnector(db);
@@ -126,6 +131,12 @@ export class Coordinator {
     if (arguments.length) {
       this._logger = logger || voidLogger();
       this.manager.logger(this._logger);
+
+      // subscribe logger to events
+      this.eventBus.observe(EventType.QueryStart, (event) => this._logger.info('Query started:', event));
+      this.eventBus.observe(EventType.QueryEnd, (event) => this._logger.info('Query ended:', event));
+      this.eventBus.observe(EventType.ClientConnect, (event) => this._logger.info('Client connected:', event));
+      this.eventBus.observe(EventType.Error, (event) => this._logger.error('Error:', event.message));
     }
     return this._logger!;
   }
@@ -249,9 +260,9 @@ export class Coordinator {
     return client._pending = this.query(query, { priority })
       .then(
         data => client.queryResult(data).update(),
-        err => { this._logger?.error(err); client.queryError(err); }
+        err => { this.eventBus.emit(EventType.Error, { message: err }); client.queryError(err); }
       )
-      .catch(err => this._logger?.error(err));
+      .catch(err => { this.eventBus.emit(EventType.Error, { message: err }); });
   }
 
   /**
@@ -280,6 +291,11 @@ export class Coordinator {
       throw new Error('Client already connected.');
     }
 
+    // emit ClientConnect
+    this.eventBus.emit(EventType.ClientConnect, {
+      // additional arguments for later -- for comprehensive client data
+    });
+
     // add client to client set
     clients?.add(client);
 
@@ -394,4 +410,4 @@ function updateSelection(
     const query = info?.query(active.predicate) ?? client.query(filter);
     return mc.updateClient(client, query);
   }));
-}
+}

packages/mosaic/core/src/Events.ts

@@ -0,0 +1,30 @@
+export enum EventType {
+  QueryStart = 'query-start',
+  QueryEnd = 'query-end',
+  ClientConnect = 'client-connect',
+  ClientStateChange = 'client-state-change',
+  Error = 'error'
+}
+
+export interface MosaicEvent {
+  timestamp: number;
+  // Extend later with more fields
+}
+
+export interface QueryStartEvent extends MosaicEvent {
+  query: string;
+  materialized: boolean;
+}
+
+export interface QueryEndEvent extends MosaicEvent {
+  query: string;
+  materialized: boolean;
+}
+
+export interface ClientConnectEvent extends MosaicEvent {
+  clientId?: string;
+}
+
+export interface ErrorEvent extends MosaicEvent {
+  message: unknown;
+}

packages/mosaic/core/src/QueryManager.ts

@@ -5,6 +5,8 @@ import { lruCache, voidCache } from './util/cache.js';
 import { PriorityQueue } from './util/priority-queue.js';
 import { QueryResult, QueryState } from './util/query-result.js';
 import { voidLogger } from './util/void-logger.js';
+import { EventType } from './Events.js';
+import { ObserveDispatch } from './util/ObserveDispatch.js';
 
 export const Priority = Object.freeze({ High: 0, Normal: 1, Low: 2 });
 
@@ -13,23 +15,23 @@ export class QueryManager {
   private db: Connector | null;
   private clientCache: Cache | null;
   private _logger: Logger;
-  private _logQueries: boolean;
   private _consolidate: ReturnType<typeof consolidator> | null;
   /** Requests pending with the query manager. */
   public pendingResults: QueryResult[];
   private maxConcurrentRequests: number;
   private pendingExec: boolean;
+  public eventBus?;
 
-  constructor(maxConcurrentRequests: number = 32) {
+  constructor(maxConcurrentRequests: number = 32, eventBus?: ObserveDispatch<unknown>) {
     this.queue = new PriorityQueue(3);
     this.db = null;
     this.clientCache = null;
     this._logger = voidLogger();
-    this._logQueries = false;
     this._consolidate = null;
     this.pendingResults = [];
     this.maxConcurrentRequests = maxConcurrentRequests;
     this.pendingExec = false;
+    this.eventBus = eventBus;
   }
 
   next(): void {
@@ -80,20 +82,36 @@ export class QueryManager {
       const { query, type, cache = false, options } = request;
       const sql = query ? `${query}` : null;
 
+      // emit QueryStart
+      // this `if` check is our version of the `logQueries` flag
+      if (this.eventBus) {
+        this.eventBus.emit(EventType.QueryStart, {
+          query: sql || '',
+          materialized: cache,
+        });
+      }
+
       // check query cache
       if (cache) {
         const cached = this.clientCache!.get(sql!);
         if (cached) {
           const data = await cached;
           this._logger.debug('Cache');
           result.ready(data);
+          // emit QueryEnd for cached
+          if (this.eventBus) {
+            this.eventBus.emit(EventType.QueryEnd, {
+              query: sql || '',
+              materialized: cache,
+            });
+          }
           return;
         }
       }
 
       // issue query, potentially cache result
       const t0 = performance.now();
-      if (this._logQueries) {
+      if (this.eventBus) {
         this._logger.debug('Query', { type, sql, ...options });
       }
 
@@ -107,7 +125,19 @@ export class QueryManager {
 
       this._logger.debug(`Request: ${(performance.now() - t0).toFixed(1)}`);
       result.ready(type === 'exec' ? null : data);
+
+      if (this.eventBus) {
+        this.eventBus.emit(EventType.QueryEnd, {
+          query: sql || '',
+          materialized: cache,
+        });
+      }
     } catch (err) {
+      if (this.eventBus) {
+        this.eventBus.emit(EventType.Error, {
+          message: err,
+        });
+      }
       result.reject(err);
     }
   }
@@ -136,17 +166,6 @@ export class QueryManager {
     return value ? (this._logger = value) : this._logger;
   }
 
-  /**
-   * Get or set if queries should be logged.
-   * @param value Whether to log queries
-   * @returns Current logging state
-   */
-  logQueries(): boolean;
-  logQueries(value: boolean): boolean;
-  logQueries(value?: boolean): boolean {
-    return value !== undefined ? this._logQueries = !!value : this._logQueries;
-  }
-
   /**
    * Get or set the database connector.
    * @param connector Connector to set
@@ -217,4 +236,4 @@ export class QueryManager {
     }
     this.pendingResults = [];
   }
-}
+}