Implement websocket server

src/support/bus/RedisBus.ts

@@ -101,9 +101,7 @@ export class RedisBus implements EventBus, AwareOfContainerLifecycle {
 
         await this.subscriptions
             .where('eventName', '=', name)
-            .pluck('handler')
-            .map(handler => handler(event))
-            .awaitAll()
+            .awaitMapCall('handler', event)
     }
 
     isConnected(): boolean {

src/support/bus/WebSocketBus.ts

@@ -0,0 +1,123 @@
+import {AwareOfContainerLifecycle, Container, Inject, Injectable, StaticInstantiable} from '../../di'
+import {
+    EventBus,
+    Event,
+    EventHandler,
+    EventHandlerSubscription,
+    BusSubscriber,
+    EventHandlerReturn,
+} from './types'
+import * as WebSocket from 'ws'
+import {NoSerializerError, Serialization} from './serial/Serialization'
+import {Logging} from '../../service/Logging'
+import {Awaitable, Collection, Pipeline, uuid4} from '../../util'
+import {getEventName} from './getEventName'
+import {Bus} from './Bus'
+import {WebSocketCloseEvent} from '../../http/lifecycle/WebSocketCloseEvent'
+import {apiEvent, error} from '../../http/response/api'
+import {AsyncResource, executionAsyncId} from 'async_hooks'
+
+@Injectable()
+export class WebSocketBus implements EventBus, AwareOfContainerLifecycle {
+    awareOfContainerLifecycle: true = true
+
+    @Inject()
+    protected readonly ws!: WebSocket.WebSocket
+
+    @Inject()
+    protected readonly bus!: Bus
+
+    @Inject()
+    protected readonly serial!: Serialization
+
+    @Inject()
+    protected readonly logging!: Logging
+
+    public readonly uuid = uuid4()
+
+    private connected = false
+
+    /** List of local subscriptions on this bus. */
+    protected subscriptions: Collection<BusSubscriber<Event>> = new Collection()
+
+    /** Get a Promise that resolves then the socket closes. */
+    onClose(): Promise<WebSocketCloseEvent> {
+        return new Promise<WebSocketCloseEvent>(res => {
+            this.bus.subscribe(WebSocketCloseEvent, event => res(event))
+        })
+    }
+
+    pipe<T extends Event>(eventKey: StaticInstantiable<T>, line: Pipeline<T, EventHandlerReturn>): Awaitable<EventHandlerSubscription> {
+        return this.subscribe(eventKey, event => line.apply(event))
+    }
+
+    async push(event: Event): Promise<void> {
+        this.logging.verbose(`Pushing event to WebSocket: ${event.eventName}`)
+        this.logging.verbose(event)
+
+        await this.ws.send(await this.serial.encodeJSON(event))
+    }
+
+    async subscribe<T extends Event>(eventKey: StaticInstantiable<T>, handler: EventHandler<T>): Promise<EventHandlerSubscription> {
+        const uuid = uuid4()
+        const subscriber: BusSubscriber<Event> = {
+            eventName: getEventName(eventKey),
+            eventKey,
+            handler,
+            uuid,
+        } as unknown as BusSubscriber<Event>
+
+        this.logging.verbose(`Creating WebSocket subscriber ${uuid}...`)
+        this.logging.verbose(subscriber)
+
+        this.subscriptions.push(subscriber)
+
+        return {
+            unsubscribe: () => {
+                this.subscriptions = this.subscriptions.where('uuid', '!=', uuid)
+            },
+        }
+    }
+
+    protected async onMessage(message: string): Promise<void> {
+        const payload = await this.serial.decodeJSON<Event>(message)  // FIXME validation
+
+        await this.subscriptions
+            .where('eventName', '=', payload.eventName)
+            .awaitMapCall('handler', payload)
+    }
+
+    up(): void {
+        const resource = new AsyncResource('WebSocketBus', {
+            triggerAsyncId: executionAsyncId(),
+            requireManualDestroy: false,
+        })
+
+        this.ws.on('message', async data => {
+            await resource.runInAsyncScope(async () => {
+                this.logging.verbose(`Got data from websocket: ${data}`)
+                try {
+                    Container.getContainer()
+                    await this.onMessage(`${data}`)
+                } catch (e: unknown) {
+                    if ( e instanceof NoSerializerError ) {
+                        this.logging.warn(`Discarding message as no validator could be found to deserialize it: ${data}`)
+                        this.push(apiEvent(error('Invalid message format or serializer.')))
+                    } else {
+                        throw e
+                    }
+                }
+            })
+        })
+
+        this.connected = true
+    }
+
+    down(): void {
+        this.connected = false
+    }
+
+    isConnected(): boolean {
+        return this.connected  // FIXME: monitor for bad connections
+    }
+}

src/support/bus/index.ts

@@ -4,10 +4,13 @@ export * from './serial/BaseSerializer'
 export * from './serial/SimpleCanonicalItemSerializer'
 export * from './serial/Serialization'
 export * from './serial/decorators'
+export * from './serial/NamedEventPayload'
+export * from './serial/JSONMessageEvent'
 
 export * from './Bus'
 export * from './LocalBus'
 export * from './RedisBus'
+export * from './WebSocketBus'
 
 export * from './queue/event/PushingToQueue'
 export * from './queue/event/PushedToQueue'

src/support/bus/serial/JSONMessageEvent.ts

@@ -0,0 +1,35 @@
+import {BaseSerializer} from './BaseSerializer'
+import {Awaitable, JSONState, uuid4} from '../../../util'
+import {ObjectSerializer} from './decorators'
+import {Event} from '../types'
+
+export class JSONMessageEvent<T extends JSONState> implements Event {
+    constructor(
+        public readonly value: T,
+    ) {}
+
+    eventName = '@extollo/lib:JSONMessageEvent'
+
+    eventUuid = uuid4()
+}
+
+@ObjectSerializer()
+export class JSONMessageEventSerializer extends BaseSerializer<JSONMessageEvent<JSONState>, { value: JSONState }> {
+    protected decodeSerial(serial: { value: JSONState }): Awaitable<JSONMessageEvent<JSONState>> {
+        return new JSONMessageEvent(serial.value)
+    }
+
+    protected encodeActual(actual: JSONMessageEvent<JSONState>): Awaitable<{ value: JSONState }> {
+        return {
+            value: actual.value,
+        }
+    }
+
+    protected getName(): string {
+        return '@extollo/lib:JSONMessageEventSerializer'
+    }
+
+    matchActual(some: JSONMessageEvent<JSONState>): boolean {
+        return some instanceof JSONMessageEvent
+    }
+}

src/support/bus/serial/NamedEventPayload.ts

@@ -0,0 +1,36 @@
+import {Event, SerialPayload} from '../types'
+import {ObjectSerializer} from './decorators'
+import {BaseSerializer} from './BaseSerializer'
+import {JSONState} from '../../../util'
+
+export class NamedEventPayload {
+    constructor(
+        public readonly eventName: string,
+        public readonly event: Event,
+    ) {}
+}
+
+@ObjectSerializer()
+export class NamedEventPayloadSerializer extends BaseSerializer<NamedEventPayload, { eventName: string, event: SerialPayload<Event, JSONState> }> {
+    protected async decodeSerial(serial: { eventName: string; event: SerialPayload<Event, JSONState> }): Promise<NamedEventPayload> {
+        return new NamedEventPayload(
+            serial.eventName,
+            await this.getSerialization().decode(serial.event),
+        )
+    }
+
+    protected async encodeActual(actual: NamedEventPayload): Promise<{ eventName: string; event: SerialPayload<Event, JSONState> }> {
+        return {
+            eventName: actual.eventName,
+            event: await this.getSerialization().encode(actual.event),
+        }
+    }
+
+    protected getName(): string {
+        return '@extollo/lib:NamedEventPayloadSerializer'
+    }
+
+    matchActual(some: NamedEventPayload): boolean {
+        return some instanceof NamedEventPayload
+    }
+}