Add basic concepts for event bus, and implement in request and model

src/event/Event.ts

@@ -0,0 +1,11 @@
+import {Dispatchable} from './types'
+import {JSONState} from '../util'
+
+/**
+ * Abstract class representing an event that may be fired.
+ */
+export abstract class Event implements Dispatchable {
+    abstract dehydrate(): Promise<JSONState>
+
+    abstract rehydrate(state: JSONState): void | Promise<void>
+}

src/event/EventBus.ts

@@ -0,0 +1,53 @@
+import {Singleton, StaticClass} from '../di'
+import {Bus, Dispatchable, EventSubscriber, EventSubscriberEntry, EventSubscription} from './types'
+import {Awaitable, Collection, uuid4} from '../util'
+
+/**
+ * A non-queued bus implementation that executes subscribers immediately in the main thread.
+ */
+@Singleton()
+export class EventBus implements Bus {
+    /**
+     * Collection of subscribers, by their events.
+     * @protected
+     */
+    protected subscribers: Collection<EventSubscriberEntry<any>> = new Collection<EventSubscriberEntry<any>>()
+
+    subscribe<T extends Dispatchable>(event: StaticClass<T, T>, subscriber: EventSubscriber<T>): Awaitable<EventSubscription> {
+        const entry: EventSubscriberEntry<T> = {
+            id: uuid4(),
+            event,
+            subscriber,
+        }
+
+        this.subscribers.push(entry)
+        return this.buildSubscription(entry.id)
+    }
+
+    unsubscribe<T extends Dispatchable>(subscriber: EventSubscriber<T>): Awaitable<void> {
+        this.subscribers = this.subscribers.where('subscriber', '!=', subscriber)
+    }
+
+    async dispatch(event: Dispatchable): Promise<void> {
+        const eventClass: StaticClass<typeof event, typeof event> = event.constructor as StaticClass<Dispatchable, Dispatchable>
+        await this.subscribers.where('event', '=', eventClass)
+            .promiseMap(entry => entry.subscriber(event))
+    }
+
+    /**
+     * Build an EventSubscription object for the subscriber of the given ID.
+     * @param id
+     * @protected
+     */
+    protected buildSubscription(id: string): EventSubscription {
+        let subscribed = true
+        return {
+            unsubscribe: (): Awaitable<void> => {
+                if ( subscribed ) {
+                    this.subscribers = this.subscribers.where('id', '!=', id)
+                    subscribed = false
+                }
+            },
+        }
+    }
+}

src/event/PropagatingEventBus.ts

@@ -0,0 +1,28 @@
+import {EventBus} from './EventBus'
+import {Collection} from '../util'
+import {Bus, Dispatchable} from './types'
+
+/**
+ * A non-queued bus implementation that executes subscribers immediately in the main thread.
+ * This bus also supports "propagating" events along to any other connected buses.
+ * Such behavior is useful, e.g., if we want to have a semi-isolated request-
+ * level bus whose events still reach the global EventBus instance.
+ */
+export class PropagatingEventBus extends EventBus {
+    protected recipients: Collection<Bus> = new Collection<Bus>()
+
+    async dispatch(event: Dispatchable): Promise<void> {
+        await super.dispatch(event)
+        await this.recipients.promiseMap(bus => bus.dispatch(event))
+    }
+
+    /**
+     * Register the given bus to receive events fired on this bus.
+     * @param recipient
+     */
+    connect(recipient: Bus): void {
+        if ( !this.recipients.includes(recipient) ) {
+            this.recipients.push(recipient)
+        }
+    }
+}

src/event/types.ts

@@ -0,0 +1,47 @@
+import {Awaitable, Rehydratable} from '../util'
+import {StaticClass} from '../di'
+
+/**
+ * A closure that should be executed with the given event is fired.
+ */
+export type EventSubscriber<T extends Dispatchable> = (event: T) => Awaitable<void>
+
+/**
+ * An object used to track event subscriptions internally.
+ */
+export interface EventSubscriberEntry<T extends Dispatchable> {
+    /** Globally unique ID of this subscription. */
+    id: string
+
+    /** The event class subscribed to. */
+    event: StaticClass<T, T>
+
+    /** The closure to execute when the event is fired. */
+    subscriber: EventSubscriber<T>
+}
+
+/**
+ * An object returned upon subscription, used to unsubscribe.
+ */
+export interface EventSubscription {
+    /**
+     * Unsubscribe the associated listener from the event bus.
+     */
+    unsubscribe(): Awaitable<void>
+}
+
+/**
+ * An instance of something that can be fired on an event bus.
+ */
+export interface Dispatchable extends Rehydratable {
+    shouldQueue?: boolean
+}
+
+/**
+ * An event-driven bus that manages subscribers and dispatched items.
+ */
+export interface Bus {
+    subscribe<T extends Dispatchable>(eventClass: StaticClass<T, T>, subscriber: EventSubscriber<T>): Awaitable<EventSubscription>
+    unsubscribe<T extends Dispatchable>(subscriber: EventSubscriber<T>): Awaitable<void>
+    dispatch(event: Dispatchable): Awaitable<void>
+}

src/http/kernel/module/InjectRequestEventBusHTTPModule.ts

@@ -0,0 +1,28 @@
+import {HTTPKernelModule} from '../HTTPKernelModule'
+import {Inject, Injectable} from '../../../di'
+import {HTTPKernel} from '../HTTPKernel'
+import {Request} from '../../lifecycle/Request'
+import {EventBus} from '../../../event/EventBus'
+import {PropagatingEventBus} from '../../../event/PropagatingEventBus'
+
+/**
+ * HTTP kernel module that creates a request-specific event bus
+ * and injects it into the request container.
+ */
+@Injectable()
+export class InjectRequestEventBusHTTPModule extends HTTPKernelModule {
+    @Inject()
+    protected bus!: EventBus
+
+    public static register(kernel: HTTPKernel): void {
+        kernel.register(this).first()
+    }
+
+    public async apply(request: Request): Promise<Request> {
+        const bus = <PropagatingEventBus> this.make(PropagatingEventBus)
+        bus.connect(this.bus)
+
+        request.purge(EventBus).registerProducer(EventBus, () => bus)
+        return request
+    }
+}

src/index.ts

@@ -1,6 +1,11 @@
 export * from './util'
 export * from './di'
 
+export * from './event/types'
+export * from './event/Event'
+export * from './event/EventBus'
+export * from './event/PropagatingEventBus'
+
 export * from './service/Logging'
 
 export * from './lifecycle/RunLevelErrorHandler'

src/orm/index.ts

@@ -15,6 +15,7 @@ export * from './model/Field'
 export * from './model/ModelBuilder'
 export * from './model/ModelBuilder'
 export * from './model/ModelResultIterable'
+export * from './model/events'
 export * from './model/Model'
 
 export * from './services/Database'

src/orm/model/Model.ts

@@ -1,18 +1,26 @@
 import {ModelKey, QueryRow, QuerySource} from '../types'
-import {Container, Inject} from '../../di'
+import {Container, Inject, StaticClass} from '../../di'
 import {DatabaseService} from '../DatabaseService'
 import {ModelBuilder} from './ModelBuilder'
 import {getFieldsMeta, ModelField} from './Field'
-import {deepCopy, BehaviorSubject, Pipe, Collection} from '../../util'
+import {deepCopy, Pipe, Collection, Awaitable, uuid4} from '../../util'
 import {EscapeValueObject} from '../dialect/SQLDialect'
 import {AppClass} from '../../lifecycle/AppClass'
 import {Logging} from '../../service/Logging'
 import {Connection} from '../connection/Connection'
+import {Bus, Dispatchable, EventSubscriber, EventSubscriberEntry, EventSubscription} from '../../event/types'
+import {ModelRetrievedEvent} from './events/ModelRetrievedEvent'
+import {ModelSavingEvent} from './events/ModelSavingEvent'
+import {ModelSavedEvent} from './events/ModelSavedEvent'
+import {ModelUpdatingEvent} from './events/ModelUpdatingEvent'
+import {ModelUpdatedEvent} from './events/ModelUpdatedEvent'
+import {ModelCreatingEvent} from './events/ModelCreatingEvent'
+import {ModelCreatedEvent} from './events/ModelCreatedEvent'
 
 /**
  * Base for classes that are mapped to tables in a database.
  */
-export abstract class Model<T extends Model<T>> extends AppClass {
+export abstract class Model<T extends Model<T>> extends AppClass implements Bus {
     @Inject()
     protected readonly logging!: Logging;
 
@@ -78,49 +86,10 @@ export abstract class Model<T extends Model<T>> extends AppClass {
     protected originalSourceRow?: QueryRow
 
     /**
-     * Behavior subject that fires after the model is populated.
+     * Collection of event subscribers, by their events.
+     * @protected
      */
-    protected retrieved$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right before the model is saved.
-     */
-    protected saving$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right after the model is saved.
-     */
-    protected saved$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right before the model is updated.
-     */
-    protected updating$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right after the model is updated.
-     */
-    protected updated$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right before the model is inserted.
-     */
-    protected creating$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right after the model is inserted.
-     */
-    protected created$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right before the model is deleted.
-     */
-    protected deleting$ = new BehaviorSubject<Model<T>>()
-
-    /**
-     * Behavior subject that fires right after the model is deleted.
-     */
-    protected deleted$ = new BehaviorSubject<Model<T>>()
+    protected modelEventBusSubscribers: Collection<EventSubscriberEntry<any>> = new Collection<EventSubscriberEntry<any>>()
 
     /**
      * Get the table name for this model.
@@ -193,9 +162,16 @@ export abstract class Model<T extends Model<T>> extends AppClass {
         values?: {[key: string]: any},
     ) {
         super()
+        this.initialize()
         this.boot(values)
     }
 
+    /**
+     * Called when the model is instantiated. Use for any setup of events, &c.
+     * @protected
+     */
+    protected initialize(): void {}  // eslint-disable-line @typescript-eslint/no-empty-function
+
     /**
      * Initialize the model's properties from the given values and do any other initial setup.
      *
@@ -228,7 +204,7 @@ export abstract class Model<T extends Model<T>> extends AppClass {
             this.setFieldFromObject(field.modelKey, field.databaseKey, row)
         })
 
-        await this.retrieved$.next(this)
+        await this.dispatch(new ModelRetrievedEvent<T>(this as any))
         return this
     }
 
@@ -592,11 +568,11 @@ export abstract class Model<T extends Model<T>> extends AppClass {
      * @param withoutTimestamps
      */
     public async save({ withoutTimestamps = false } = {}): Promise<Model<T>> {
-        await this.saving$.next(this)
+        await this.dispatch(new ModelSavingEvent<T>(this as any))
         const ctor = this.constructor as typeof Model
 
         if ( this.exists() && this.isDirty() ) {
-            await this.updating$.next(this)
+            await this.dispatch(new ModelUpdatingEvent<T>(this as any))
 
             if ( !withoutTimestamps && ctor.timestamps && ctor.UPDATED_AT ) {
                 (this as any)[ctor.UPDATED_AT] = new Date()
@@ -617,9 +593,9 @@ export abstract class Model<T extends Model<T>> extends AppClass {
                 await this.assumeFromSource(data)
             }
 
-            await this.updated$.next(this)
+            await this.dispatch(new ModelUpdatedEvent<T>(this as any))
         } else if ( !this.exists() ) {
-            await this.creating$.next(this)
+            await this.dispatch(new ModelCreatingEvent<T>(this as any))
 
             if ( !withoutTimestamps ) {
                 if ( ctor.timestamps && ctor.CREATED_AT ) {
@@ -647,10 +623,11 @@ export abstract class Model<T extends Model<T>> extends AppClass {
             if ( data ) {
                 await this.assumeFromSource(result)
             }
-            await this.created$.next(this)
+
+            await this.dispatch(new ModelCreatedEvent<T>(this as any))
         }
 
-        await this.saved$.next(this)
+        await this.dispatch(new ModelSavedEvent<T>(this as any))
         return this
     }
 
@@ -822,4 +799,42 @@ export abstract class Model<T extends Model<T>> extends AppClass {
     protected setFieldFromObject(thisFieldName: string | symbol, objectFieldName: string, object: QueryRow): void {
         (this as any)[thisFieldName] = object[objectFieldName]
     }
+
+    subscribe<EventT extends Dispatchable>(event: StaticClass<EventT, EventT>, subscriber: EventSubscriber<EventT>): Awaitable<EventSubscription> {
+        const entry: EventSubscriberEntry<EventT> = {
+            id: uuid4(),
+            event,
+            subscriber,
+        }
+
+        this.modelEventBusSubscribers.push(entry)
+        return this.buildSubscription(entry.id)
+    }
+
+    unsubscribe<EventT extends Dispatchable>(subscriber: EventSubscriber<EventT>): Awaitable<void> {
+        this.modelEventBusSubscribers = this.modelEventBusSubscribers.where('subscriber', '!=', subscriber)
+    }
+
+    async dispatch(event: Dispatchable): Promise<void> {
+        const eventClass: StaticClass<typeof event, typeof event> = event.constructor as StaticClass<Dispatchable, Dispatchable>
+        await this.modelEventBusSubscribers.where('event', '=', eventClass)
+            .promiseMap(entry => entry.subscriber(event))
+    }
+
+    /**
+     * Build an EventSubscription object for the subscriber of the given ID.
+     * @param id
+     * @protected
+     */
+    protected buildSubscription(id: string): EventSubscription {
+        let subscribed = true
+        return {
+            unsubscribe: (): Awaitable<void> => {
+                if ( subscribed ) {
+                    this.modelEventBusSubscribers = this.modelEventBusSubscribers.where('id', '!=', id)
+                    subscribed = false
+                }
+            },
+        }
+    }
 }

src/orm/model/events/ModelCreatedEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right after a model is inserted.
+ */
+export class ModelCreatedEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelCreatingEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right before a model is inserted.
+ */
+export class ModelCreatingEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelDeletedEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right after a model is deleted.
+ */
+export class ModelDeletedEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelDeletingEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right before a model is deleted.
+ */
+export class ModelDeletingEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelEvent.ts

@@ -0,0 +1,31 @@
+import {Model} from '../Model'
+import {Event} from '../../../event/Event'
+import {JSONState} from '../../../util'
+
+/**
+ * Base class for events that concern an instance of a model.
+ */
+export abstract class ModelEvent<T extends Model<T>> extends Event {
+    /**
+     * The instance of the model.
+     */
+    public instance!: T
+
+    constructor(
+        instance?: T,
+    ) {
+        super()
+        if ( instance ) {
+            this.instance = instance
+        }
+    }
+
+    // TODO implement serialization here
+    dehydrate(): Promise<JSONState> {
+        return Promise.resolve({})
+    }
+
+    rehydrate(/* state: JSONState */): void | Promise<void> {
+        return undefined
+    }
+}

src/orm/model/events/ModelRetrievedEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right after a model's data is loaded from the source.
+ */
+export class ModelRetrievedEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelSavedEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right after a model is persisted to the source.
+ */
+export class ModelSavedEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelSavingEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right before a model is persisted to the source.
+ */
+export class ModelSavingEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelUpdatedEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right after a model's data is updated.
+ */
+export class ModelUpdatedEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/ModelUpdatingEvent.ts

@@ -0,0 +1,9 @@
+import {Model} from '../Model'
+import {ModelEvent} from './ModelEvent'
+
+/**
+ * Event fired right before a model's data is updated.
+ */
+export class ModelUpdatingEvent<T extends Model<T>> extends ModelEvent<T> {
+
+}

src/orm/model/events/index.ts

@@ -0,0 +1,21 @@
+import {ModelCreatedEvent} from './ModelCreatedEvent'
+import {ModelUpdatingEvent} from './ModelUpdatingEvent'
+import {ModelCreatingEvent} from './ModelCreatingEvent'
+import {ModelSavedEvent} from './ModelSavedEvent'
+import {ModelDeletedEvent} from './ModelDeletedEvent'
+import {ModelDeletingEvent} from './ModelDeletingEvent'
+import {ModelRetrievedEvent} from './ModelRetrievedEvent'
+import {ModelUpdatedEvent} from './ModelUpdatedEvent'
+import {ModelEvent} from './ModelEvent'
+
+export const ModelEvents = {
+    ModelCreatedEvent,
+    ModelCreatingEvent,
+    ModelDeletedEvent,
+    ModelDeletingEvent,
+    ModelEvent,
+    ModelRetrievedEvent,
+    ModelSavedEvent,
+    ModelUpdatedEvent,
+    ModelUpdatingEvent,
+}

src/service/HTTPServer.ts

@@ -16,6 +16,7 @@ import {ExecuteResolvedRoutePreflightHTTPModule} from '../http/kernel/module/Exe
 import {ExecuteResolvedRoutePostflightHTTPModule} from '../http/kernel/module/ExecuteResolvedRoutePostflightHTTPModule'
 import {ParseIncomingBodyHTTPModule} from '../http/kernel/module/ParseIncomingBodyHTTPModule'
 import {Config} from './Config'
+import {InjectRequestEventBusHTTPModule} from '../http/kernel/module/InjectRequestEventBusHTTPModule'
 
 /**
  * Application unit that starts the HTTP/S server, creates Request and Response objects
@@ -48,6 +49,7 @@ export class HTTPServer extends Unit {
         ExecuteResolvedRoutePreflightHTTPModule.register(this.kernel)
         ExecuteResolvedRoutePostflightHTTPModule.register(this.kernel)
         ParseIncomingBodyHTTPModule.register(this.kernel)
+        InjectRequestEventBusHTTPModule.register(this.kernel)
 
         await new Promise<void>(res => {
             this.server = createServer(this.handler)