feat(ceb-messaging-core): received messages should be observable

packages/ceb-messaging-core/src/command.ts

@@ -127,6 +127,10 @@ export interface CommandBus extends Disposable {
  * The map defines the internal events of an {@link CommandBus}.
  */
 export type CommandBusNotificationMap = {
+  command_received: {
+    bus: CommandBus
+    command: Command
+  }
   command_handler_failed: {
     bus: CommandBus
     command: Command

packages/ceb-messaging-core/src/event.ts

@@ -78,6 +78,10 @@ export interface EventBus extends Disposable {
  * The map defines the internal events of an {@link EventBus}.
  */
 export type EventBusNotificationMap = {
+  event_received: {
+    bus: EventBus
+    event: Event
+  }
   event_listener_failed: {
     bus: EventBus
     event: Event

packages/ceb-messaging-core/src/message.ts

@@ -16,7 +16,7 @@ export type MessageType = string
 /**
  * The value of a header.
  */
-export type MessageHeaderValue = string | number | Array<string | number>
+export type MessageHeaderValue = string | number | boolean | Array<string | number | boolean>
 
 /**
  * The headers of the a message.

packages/ceb-messaging-core/src/query.ts

@@ -78,6 +78,10 @@ export interface QueryBus extends Disposable {
  * The map defines the internal events of an {@link QueryBus}.
  */
 export type QueryBusNotificationMap = {
+  query_received: {
+    bus: QueryBus
+    query: Query
+  }
   query_handler_failed: {
     bus: QueryBus
     query: Query

packages/ceb-messaging-simple/src/command.ts

@@ -36,6 +36,11 @@ export class SimpleCommandBus implements CommandBus, Disposable {
     command: C,
     options?: Partial<ExecuteActionOptions>
   ): Promise<CommandResult<R>> {
+    this.emitter.emit("command_received", {
+      bus: this,
+      command,
+    })
+
     const handler = this.resolveHandler(command)
 
     const opts: ExecuteActionOptions = {
@@ -58,6 +63,11 @@ export class SimpleCommandBus implements CommandBus, Disposable {
   }
 
   executeAndForget<C extends Command = Command>(command: C): void {
+    this.emitter.emit("command_received", {
+      bus: this,
+      command,
+    })
+
     const handler = this.resolveHandler(command)
     Promise.resolve((async () => handler(command))())
       .then((output) => this.processHandlerOutput(output))

packages/ceb-messaging-simple/src/event.ts

@@ -26,6 +26,10 @@ export class SimpleEventBus implements EventBus, Disposable {
 
   publish<E extends Event = Event>(...events: Array<E>): void {
     events.forEach((event) => {
+      this.emitter.emit("event_received", {
+        bus: this,
+        event,
+      })
       this.listeners.get(event.headers.messageType)?.forEach((listener) => {
         Promise.resolve((async () => listener(event))()).catch((error: Error) =>
           this.emitter.emit("event_listener_failed", {

packages/ceb-messaging-simple/src/query.ts

@@ -31,6 +31,11 @@ export class SimpleQueryBus implements QueryBus, Disposable {
     query: Q,
     options?: Partial<ExecuteActionOptions>
   ): Promise<QueryResult<R>> {
+    this.emitter.emit("query_received", {
+      bus: this,
+      query,
+    })
+
     const handler = this.resolveHandler<Q, R>(query)
 
     const opts: ExecuteActionOptions = {