chore: event publisher

Author: karenc-bq

.prettierrc

@@ -2,5 +2,5 @@
   "semi": true,
   "trailingComma": "none",
   "printWidth": 150,
-  "endOfLine": "lf"
+  "endOfLine": "auto"
 }

common/lib/aws_client.ts

@@ -37,6 +37,7 @@ import { StorageService } from "./utils/storage/storage_service";
 import { MonitorService } from "./utils/monitoring/monitor_service";
 import { CoreServicesContainer } from "./utils/core_services_container";
 import { FullServicesContainer } from "./utils/full_services_container";
+import { EventPublisher } from "./utils/events/event";
 
 const { EventEmitter } = pkgStream;
 
@@ -45,6 +46,7 @@ export abstract class AwsClient extends EventEmitter implements SessionStateClie
   private readonly fullServiceContainer: FullServicesContainer;
   private readonly storageService: StorageService;
   private readonly monitorService: MonitorService;
+  private readonly eventPublisher: EventPublisher;
   protected telemetryFactory: TelemetryFactory;
   protected pluginManager: PluginManager;
   protected pluginService: PluginService;
@@ -108,11 +110,13 @@ export abstract class AwsClient extends EventEmitter implements SessionStateClie
     const coreServicesContainer: CoreServicesContainer = CoreServicesContainer.getInstance();
     this.storageService = coreServicesContainer.getStorageService();
     this.monitorService = coreServicesContainer.getMonitorService();
+    this.eventPublisher = coreServicesContainer.getEventPublisher();
     this.telemetryFactory = new DefaultTelemetryFactory(this.properties);
 
     this.fullServiceContainer = ServiceUtils.instance.createStandardServiceContainer(
       this.storageService,
       this.monitorService,
+      this.eventPublisher,
       this,
       this.properties,
       dbType,

common/lib/types.ts

@@ -31,3 +31,9 @@ export type ShouldDisposeFunc<V> = (item: V) => boolean;
  * @param item The item to dispose
  */
 export type ItemDisposalFunc<V> = (item: V) => void;
+
+/**
+ * Type representing an event class constructor.
+ */
+export type EventClass<T extends import("./utils/events/event").Event = import("./utils/events/event").Event> =
+  new (...args: any[]) => T;

common/lib/utils/core_services_container.ts

@@ -16,6 +16,8 @@
 
 import { StorageService, StorageServiceImpl } from "./storage/storage_service";
 import { MonitorService, MonitorServiceImpl } from "./monitoring/monitor_service";
+import { EventPublisher } from "./events/event";
+import { BatchingEventPublisher } from "./events/batching_event_publisher";
 
 /**
  * A singleton container object used to instantiate and access core universal services. This class should be used
@@ -27,13 +29,14 @@ import { MonitorService, MonitorServiceImpl } from "./monitoring/monitor_service
 export class CoreServicesContainer {
   private static readonly INSTANCE = new CoreServicesContainer();
 
-  // TODO: implement monitor service
   private readonly monitorService: MonitorService;
   private readonly storageService: StorageService;
+  private readonly eventPublisher: EventPublisher;
 
   private constructor() {
-    this.storageService = new StorageServiceImpl();
-    this.monitorService = new MonitorServiceImpl();
+    this.eventPublisher = new BatchingEventPublisher();
+    this.storageService = new StorageServiceImpl(this.eventPublisher);
+    this.monitorService = new MonitorServiceImpl(this.eventPublisher);
   }
 
   static getInstance(): CoreServicesContainer {
@@ -48,6 +51,10 @@ export class CoreServicesContainer {
     return this.monitorService;
   }
 
+  getEventPublisher(): EventPublisher {
+    return this.eventPublisher;
+  }
+
   static releaseResources(): void {
     CoreServicesContainer.INSTANCE.storageService.releaseResources();
   }

common/lib/utils/events/batching_event_publisher.ts

@@ -0,0 +1,96 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  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.
+*/
+
+import { Event, EventPublisher, EventSubscriber } from "./event";
+import { EventClass } from "../../types";
+
+const DEFAULT_MESSAGE_INTERVAL_MS = 30_000; // 30 seconds
+
+/**
+ * An event publisher that periodically publishes a batch of all unique events
+ * encountered during the latest time interval.
+ */
+export class BatchingEventPublisher implements EventPublisher {
+  protected readonly subscribersMap = new Map<EventClass, Set<EventSubscriber>>();
+  protected readonly pendingEvents = new Set<Event>();
+  protected publishingInterval?: ReturnType<typeof setInterval>;
+
+  constructor(messageIntervalMs: number = DEFAULT_MESSAGE_INTERVAL_MS) {
+    this.initPublishingInterval(messageIntervalMs);
+  }
+
+  protected initPublishingInterval(messageIntervalMs: number): void {
+    this.publishingInterval = setInterval(() => this.sendMessages(), messageIntervalMs);
+    // Allow the process to exit even if the interval is still running
+    this.publishingInterval.unref();
+  }
+
+  protected sendMessages(): void {
+    for (const event of this.pendingEvents) {
+      this.pendingEvents.delete(event);
+      this.deliverEvent(event);
+    }
+  }
+
+  protected deliverEvent(event: Event): void {
+    const subscribers = this.subscribersMap.get(event.constructor as EventClass);
+    if (!subscribers) {
+      return;
+    }
+
+    for (const subscriber of subscribers) {
+      subscriber.processEvent(event);
+    }
+  }
+
+  subscribe(subscriber: EventSubscriber, eventClasses: Set<EventClass>): void {
+    for (const eventClass of eventClasses) {
+      let subscribers = this.subscribersMap.get(eventClass);
+      if (!subscribers) {
+        subscribers = new Set();
+        this.subscribersMap.set(eventClass, subscribers);
+      }
+      subscribers.add(subscriber);
+    }
+  }
+
+  unsubscribe(subscriber: EventSubscriber, eventClasses: Set<EventClass>): void {
+    for (const eventClass of eventClasses) {
+      const subscribers = this.subscribersMap.get(eventClass);
+      if (subscribers) {
+        subscribers.delete(subscriber);
+        if (subscribers.size === 0) {
+          this.subscribersMap.delete(eventClass);
+        }
+      }
+    }
+  }
+
+  publish(event: Event): void {
+    if (event.isImmediateDelivery) {
+      this.deliverEvent(event);
+    } else {
+      this.pendingEvents.add(event);
+    }
+  }
+
+  releaseResources(): void {
+    if (this.publishingInterval) {
+      clearInterval(this.publishingInterval);
+      this.publishingInterval = undefined;
+    }
+  }
+}

common/lib/utils/events/data_access_event.ts

@@ -0,0 +1,35 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  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.
+*/
+
+import { Event } from "./event";
+
+/**
+ * A class defining a data access event. The class specifies the class of the data
+ * that was accessed and the key for the data.
+ *
+ * Used by StorageService to notify MonitorService when data is accessed,
+ * allowing monitors to extend their expiration time.
+ */
+export class DataAccessEvent implements Event {
+  readonly isImmediateDelivery = false;
+  readonly dataClass: new (...args: any[]) => any;
+  readonly key: unknown;
+
+  constructor(dataClass: new (...args: any[]) => any, key: unknown) {
+    this.dataClass = dataClass;
+    this.key = key;
+  }
+}

common/lib/utils/events/event.ts

@@ -0,0 +1,62 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  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.
+*/
+
+import { EventClass } from "../../types";
+
+/**
+ * An interface for events that need to be communicated between different components.
+ */
+export interface Event {
+  readonly isImmediateDelivery: boolean;
+}
+
+/**
+ * An event subscriber. Subscribers can subscribe to a publisher's events.
+ */
+export interface EventSubscriber {
+  /**
+   * Processes an event. This method will only be called on this subscriber
+   * if it has subscribed to the event class.
+   * @param event the event to process.
+   */
+  processEvent(event: Event): void;
+}
+
+/**
+ * An event publisher that publishes events to subscribers.
+ * Subscribers can specify which types of events they would like to receive.
+ */
+export interface EventPublisher {
+  /**
+   * Registers the given subscriber for the given event classes.
+   * @param subscriber the subscriber to be notified when the given event classes occur.
+   * @param eventClasses the classes of events that the subscriber should be notified of.
+   */
+  subscribe(subscriber: EventSubscriber, eventClasses: Set<EventClass>): void;
+
+  /**
+   * Unsubscribes the given subscriber from the given event classes.
+   * @param subscriber the subscriber to unsubscribe from the given event classes.
+   * @param eventClasses the classes of events that the subscriber wants to unsubscribe from.
+   */
+  unsubscribe(subscriber: EventSubscriber, eventClasses: Set<EventClass>): void;
+
+  /**
+   * Publishes an event. All subscribers to the given event class will be notified of the event.
+   * @param event the event to publish.
+   */
+  publish(event: Event): void;
+}

common/lib/utils/events/monitor_reset_event.ts

@@ -0,0 +1,32 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  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.
+*/
+
+import { Event } from "./event";
+
+/**
+ * Event indicating that a monitor should be reset with new endpoints.
+ * Used by ClusterTopologyMonitorImpl to reset monitoring when cluster topology changes.
+ */
+export class MonitorResetEvent implements Event {
+  readonly isImmediateDelivery = true;
+  readonly clusterId: string;
+  readonly endpoints: Set<string>;
+
+  constructor(clusterId: string, endpoints: Set<string>) {
+    this.clusterId = clusterId;
+    this.endpoints = endpoints;
+  }
+}

common/lib/utils/events/monitor_stop_event.ts

@@ -0,0 +1,33 @@
+/*
+  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ 
+  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.
+*/
+
+import { Event } from "./event";
+import { Monitor } from "../monitoring/monitor";
+
+/**
+ * Event indicating that a monitor should be stopped.
+ * Used by MonitorService to stop and remove monitors.
+ */
+export class MonitorStopEvent implements Event {
+  readonly isImmediateDelivery = true;
+  readonly monitorClass: new (...args: any[]) => Monitor;
+  readonly key: unknown;
+
+  constructor(monitorClass: new (...args: any[]) => Monitor, key: unknown) {
+    this.monitorClass = monitorClass;
+    this.key = key;
+  }
+}