Use base event emitter for Message API (#782)

* Use base event emitter for Message API

* emitter transform remove and changeset include

* ApplyEventListeners comment update

.changeset/dry-hairs-rhyme.md

@@ -0,0 +1,5 @@
+---
+'@signalwire/realtime-api': patch
+---
+
+Base event emitter for Messaging API

packages/realtime-api/src/ApplyEventListeners.ts

@@ -1,9 +1,7 @@
 import { BaseConsumer, EventEmitter } from '@signalwire/core'
 
 /**
- * This class is extended by Call and Voice since they use
- * `_on`, `_once`, and `_off` instead of
- * `on`, `once`, and `off`
+ * Override all old listeners with new listeners that uses BaseComponent's new even emitter
  */
 export class ApplyEventListeners<
   EventTypes extends EventEmitter.ValidEventTypes

packages/realtime-api/src/messaging/Messaging.ts

@@ -1,15 +1,14 @@
 import {
   DisconnectableClientContract,
   BaseComponentOptions,
-  EventTransform,
   toExternalJSON,
   ClientContextContract,
 } from '@signalwire/core'
-import { connect, BaseComponent } from '@signalwire/core'
+import { connect } from '@signalwire/core'
 import type { MessagingClientApiEvents } from '../types'
 import { RealtimeClient } from '../client/index'
 import { messagingWorker } from './workers'
-import { MessageContract, Message } from './Message'
+import { ApplyEventListeners } from '../ApplyEventListeners'
 
 interface MessagingSendParams {
   context?: string
@@ -85,7 +84,7 @@ export interface Messaging
 }
 
 /** @internal */
-class MessagingAPI extends BaseComponent<MessagingClientApiEvents> {
+class MessagingAPI extends ApplyEventListeners<MessagingClientApiEvents> {
   /** @internal */
 
   constructor(options: BaseComponentOptions<MessagingClientApiEvents>) {
@@ -94,46 +93,8 @@ class MessagingAPI extends BaseComponent<MessagingClientApiEvents> {
     this.runWorker('messagingWorker', {
       worker: messagingWorker,
     })
-    this._attachListeners('')
-  }
 
-  /** @internal */
-  protected getEmitterTransforms() {
-    return new Map<string | string[], EventTransform>([
-      [
-        [
-          'messaging.state',
-          'messaging.receive',
-          'message.updated',
-          'message.received',
-        ],
-        {
-          type: 'messagingMessage',
-          instanceFactory: (payload: any) => {
-            return new Message(payload)
-          },
-          payloadTransform: (payload: any): MessageContract => {
-            /** Building a MessageContract to conform with our Proxy API */
-            const {
-              message_id,
-              message_state,
-              from_number,
-              to_number,
-              tag,
-              ...rest
-            } = payload
-
-            return toExternalJSON<MessageContract>({
-              ...rest,
-              id: message_id,
-              state: message_state,
-              from: from_number,
-              to: to_number,
-            })
-          },
-        },
-      ],
-    ])
+    this._attachListeners('')
   }
 
   async send(params: MessagingSendParams): Promise<any> {

packages/realtime-api/src/messaging/workers/index.ts

@@ -0,0 +1 @@
+export * from './messagingWorker'

packages/realtime-api/src/messaging/workers/messagingWorker.ts

@@ -0,0 +1,53 @@
+import {
+  MessagingAction,
+  SDKActions,
+  SDKWorker,
+  SagaIterator,
+  getLogger,
+  sagaEffects,
+} from '@signalwire/core'
+import type { Client } from '../../client/index'
+import { Message } from '../Messaging'
+
+export const messagingWorker: SDKWorker<Client> = function* (
+  options
+): SagaIterator {
+  getLogger().trace('messagingWorker started')
+  const { channels, instance: client } = options
+  const { swEventChannel } = channels
+
+  function* worker(action: MessagingAction) {
+    const { payload, type } = action
+
+    // @ts-expect-error
+    const message = new Message(payload)
+
+    switch (type) {
+      case 'messaging.receive':
+        // @ts-expect-error
+        client.baseEmitter.emit('message.received', message)
+        break
+      case 'messaging.state':
+        // @ts-expect-error
+        client.baseEmitter.emit('message.updated', message)
+        break
+      default:
+        getLogger().warn(`Unknown message event: "${action.type}"`)
+        break
+    }
+  }
+
+  const isMessagingEvent = (action: SDKActions) =>
+    action.type.startsWith('messaging.')
+
+  while (true) {
+    const action: MessagingAction = yield sagaEffects.take(
+      swEventChannel,
+      isMessagingEvent
+    )
+
+    yield sagaEffects.fork(worker, action)
+  }
+
+  getLogger().trace('messagingWorker ended')
+}

packages/realtime-api/src/voice/workers/voiceCallingWorker.ts

@@ -4,8 +4,7 @@ import {
   getLogger,
   sagaEffects,
   SDKActions,
-  MapToPubSubShape,
-  SwEventParams,
+  VoiceCallAction,
 } from '@signalwire/core'
 import { fork } from '@redux-saga/core/effects'
 import type { Client } from '../../client/index'
@@ -30,7 +29,7 @@ export const voiceCallingWroker: SDKWorker<Client> = function* (
   const { channels, instance, instanceMap, initialState } = options
   const { swEventChannel } = channels
 
-  function* worker(action: MapToPubSubShape<SwEventParams>) {
+  function* worker(action: VoiceCallAction) {
     switch (action.type) {
       case 'calling.call.state':
         yield fork(voiceCallStateWorker, {
@@ -135,7 +134,7 @@ export const voiceCallingWroker: SDKWorker<Client> = function* (
     action.type.startsWith('calling.')
 
   while (true) {
-    const action: MapToPubSubShape<SwEventParams> = yield sagaEffects.take(
+    const action: VoiceCallAction = yield sagaEffects.take(
       swEventChannel,
       isCallingEvent
     )