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

dev/index.html

@@ -154,7 +154,8 @@
   }
 
   function setQueryLog() {
-    vg.coordinator().manager.logQueries(qlogToggle.checked);
+    // logQueries is no longer in use
+    // vg.coordinator().manager.logQueries(qlogToggle.checked);
   }
 
   function setCache() {

packages/mosaic/core/src/Coordinator.ts

@@ -11,6 +11,7 @@ import { type MosaicClient } from './MosaicClient.js';
 import { type SelectionClause } from './SelectionClause.js';
 import { MaybeArray } from '@uwdata/mosaic-sql';
 import { Table } from '@uwdata/flechette';
+import { EventBus, EventType } from './EventBus.js';
 
 interface FilterGroupEntry {
   selection: Selection;
@@ -50,6 +51,7 @@ export class Coordinator {
   public clients = new Set<MosaicClient>;
   public filterGroups = new Map<Selection, FilterGroupEntry>;
   protected _logger: Logger = voidLogger();
+  public eventBus: EventBus;
 
   /**
    * @param db Database connector. Defaults to a web socket connection.
@@ -77,7 +79,9 @@ export class Coordinator {
       consolidate = true,
       preagg = {}
     } = options;
+    this.eventBus = new EventBus();
     this.manager = manager;
+    this.manager.eventBus = this.eventBus;
     this.manager.cache(cache);
     this.manager.consolidate(consolidate);
     this.databaseConnector(db);
@@ -126,6 +130,11 @@ 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!;
   }
@@ -195,16 +204,18 @@ export class Coordinator {
       cache?: boolean;
       persist?: boolean;
       priority?: number;
+      clientId?: string;
       [key: string]: unknown;
     } = {}
   ): QueryResult<any> {
     const {
       type = 'arrow',
       cache = true,
       priority = Priority.Normal,
+      clientId,
       ...otherOptions
     } = options;
-    return this.manager.request({ type, query, cache, options: otherOptions }, priority);
+    return this.manager.request({ type, query, cache, options: otherOptions, clientId }, priority);
   }
 
   /**
@@ -246,12 +257,12 @@ export class Coordinator {
     priority: number = Priority.Normal
   ): Promise<unknown> {
     client.queryPending();
-    return client._pending = this.query(query, { priority })
+    return client._pending = this.query(query, { priority, clientId: (client as any).id })
       .then(
         data => client.queryResult(data).update(),
-        err => { this._logger?.error(err); client.queryError(err); }
+        err => { this.eventBus.emit(EventType.Error, { message: err, timestamp: Date.now() }); client.queryError(err); }
       )
-      .catch(err => this._logger?.error(err));
+      .catch(err => { this.eventBus.emit(EventType.Error, { message: err, timestamp: Date.now() }); });
   }
 
   /**
@@ -280,6 +291,17 @@ export class Coordinator {
       throw new Error('Client already connected.');
     }
 
+    // assign client id if not present
+    if (!(client as any).id) {
+      (client as any).id = `client-${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
+    }
+
+    // emit ClientConnect
+    this.eventBus.emit(EventType.ClientConnect, {
+      clientId: (client as any).id,
+      timestamp: Date.now()
+    });
+
     // add client to client set
     clients?.add(client);
 
@@ -394,4 +416,4 @@ function updateSelection(
     const query = info?.query(active.predicate) ?? client.query(filter);
     return mc.updateClient(client, query);
   }));
-}
+}

packages/mosaic/core/src/EventBus.ts

@@ -0,0 +1,49 @@
+import { ObserveDispatch } from './util/ObserveDispatch.js';
+
+export enum EventType {
+  QueryStart = 'query-start',
+  QueryEnd = 'query-end',
+  ClientConnect = 'client-connect',
+  ClientStateChange = 'client-state-change',
+  Error = 'error'
+}
+
+export interface QueryStartEvent {
+  query: string;
+  materialized: boolean;
+  clientId?: string;
+  timestamp: number;
+}
+
+export interface QueryEndEvent {
+  query: string;
+  materialized: boolean;
+  clientId?: string;
+  timestamp: number;
+}
+
+export interface ClientConnectEvent {
+  clientId: string;
+  timestamp: number;
+}
+
+export interface ErrorEvent {
+  message: unknown;
+  timestamp: number;
+}
+
+export class EventBus {
+  private dispatch = new ObserveDispatch();
+
+  observe(type: EventType, callback: (value: any) => void): void {
+    this.dispatch.observe(type, callback);
+  }
+
+  unobserve(type: EventType, callback: (value: any) => void): void {
+    this.dispatch.unobserve(type, callback);
+  }
+
+  emit(type: EventType, value: any): void {
+    this.dispatch.emit(type, value);
+  }
+}

packages/mosaic/core/src/QueryManager.ts

@@ -5,6 +5,7 @@ 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 { EventBus, EventType } from './EventBus.js';
 
 export const Priority = Object.freeze({ High: 0, Normal: 1, Low: 2 });
 
@@ -13,23 +14,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?: EventBus;
 
-  constructor(maxConcurrentRequests: number = 32) {
+  constructor(maxConcurrentRequests: number = 32, eventBus?: EventBus) {
     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 {
@@ -77,23 +78,43 @@ export class QueryManager {
    */
   async submit(request: QueryRequest, result: QueryResult): Promise<void> {
     try {
-      const { query, type, cache = false, options } = request;
+      const { query, type, cache = false, options, clientId } = 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,
+          clientId,
+          timestamp: Date.now()
+        });
+      }
+
       // 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,
+              clientId,
+              timestamp: Date.now()
+            });
+          }
           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 +128,22 @@ 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,
+          clientId,
+          timestamp: Date.now()
+        });
+      }
     } catch (err) {
+      if (this.eventBus) {
+        this.eventBus.emit(EventType.Error, {
+          message: err,
+          timestamp: Date.now()
+        });
+      }
       result.reject(err);
     }
   }
@@ -136,17 +172,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 +242,4 @@ export class QueryManager {
     }
     this.pendingResults = [];
   }
-}
+}

packages/mosaic/core/src/types.ts

@@ -7,6 +7,7 @@ export interface QueryRequest {
   query: string | Query | DescribeQuery;
   cache?: boolean;
   options?: Record<string, unknown>;
+  clientId?: string;
 }
 
 /** Type for an entry within a query manager. */