Add CacheSession implementation & make WebSocketBus re-load the session when an event is received

src/support/bus/WebSocketBus.ts

@@ -16,11 +16,25 @@ import {Bus} from './Bus'
 import {WebSocketCloseEvent} from '../../http/lifecycle/WebSocketCloseEvent'
 import {apiEvent, error} from '../../http/response/api'
 import {AsyncResource, executionAsyncId} from 'async_hooks'
+import {Session} from '../../http/session/Session'
 
 @Injectable()
 export class WebSocketBus implements EventBus, AwareOfContainerLifecycle {
     awareOfContainerLifecycle: true = true
 
+    /**
+     * If true, the session will be loaded when an event is received and
+     * persisted after the event's handlers have executed.
+     *
+     * If an event has no handlers, the session will NOT be loaded.
+     *
+     * Use `disableSessionLoad()` to disable.
+     *
+     * @see disableSessionLoad
+     * @protected
+     */
+    protected shouldLoadSessionOnEvent = true
+
     @Inject()
     protected readonly ws!: WebSocket.WebSocket
 
@@ -33,6 +47,9 @@ export class WebSocketBus implements EventBus, AwareOfContainerLifecycle {
     @Inject()
     protected readonly logging!: Logging
 
+    @Inject()
+    protected readonly session!: Session
+
     public readonly uuid = uuid4()
 
     private connected = false
@@ -40,6 +57,15 @@ export class WebSocketBus implements EventBus, AwareOfContainerLifecycle {
     /** List of local subscriptions on this bus. */
     protected subscriptions: Collection<BusSubscriber<Event>> = new Collection()
 
+    /**
+     * Disables re-loading & persisting the session when an event with listeners is received.
+     * @see shouldLoadSessionOnEvent
+     */
+    disableSessionLoad(): this {
+        this.shouldLoadSessionOnEvent = false
+        return this
+    }
+
     /** Get a Promise that resolves then the socket closes. */
     onClose(): Promise<WebSocketCloseEvent> {
         return new Promise<WebSocketCloseEvent>(res => {
@@ -82,9 +108,21 @@ export class WebSocketBus implements EventBus, AwareOfContainerLifecycle {
     protected async onMessage(message: string): Promise<void> {
         const payload = await this.serial.decodeJSON<Event>(message)  // FIXME validation
 
-        await this.subscriptions
+        const listeners = await this.subscriptions
             .where('eventName', '=', payload.eventName)
-            .awaitMapCall('handler', payload)
+
+        // If configured, re-load the session data since it may have changed outside the
+        // current socket's request.
+        if ( this.shouldLoadSessionOnEvent && listeners.isNotEmpty() ) {
+            await this.session.load()
+        }
+
+        await listeners.awaitMapCall('handler', payload)
+
+        // Persist any changes to the session for other requests.
+        if ( this.shouldLoadSessionOnEvent && listeners.isNotEmpty() ) {
+            await this.session.persist()
+        }
     }
 
     up(): void {