Fix taskWorker

.changeset/big-wolves-suffer.md

@@ -0,0 +1,6 @@
+---
+'@sw-internal/e2e-realtime-api': patch
+'@signalwire/realtime-api': patch
+---
+
+Fix `task.received` handler on the Task namespace.

internal/e2e-realtime-api/src/task.test.ts

@@ -4,9 +4,13 @@ import { createTestRunner } from './utils'
 const handler = () => {
  return new Promise<number>(async (resolve, reject) => {
  const context = 'task-e2e'
- const jobPayload = {
+ const firstPayload = {
  id: Date.now(),
- item: 'foo',
+ item: 'first',
+ }
+ const lastPayload = {
+ id: Date.now(),
+ item: 'last',
  }
 
  const client = new Task.Client({
@@ -16,20 +20,37 @@ const handler = () => {
  contexts: [context],
  })
 
+ let counter = 0
+
  client.on('task.received', (payload) => {
- if (payload.id === jobPayload.id && payload.item === 'foo') {
+ if (payload.id === firstPayload.id && payload.item === 'first') {
+ counter++
+ } else if (payload.id === lastPayload.id && payload.item === 'last') {
+ counter++
+ } else {
+ console.error('Invalid payload on `task.received`', payload)
+ return reject(4)
+ }
+
+ if (counter === 2) {
  return resolve(0)
  }
- console.error('Invalid payload on `task.received`', payload)
- return reject(4)
  })
 
  await Task.send({
  host: process.env.RELAY_HOST as string,
  project: process.env.RELAY_PROJECT as string,
  token: process.env.RELAY_TOKEN as string,
  context,
- message: jobPayload,
+ message: firstPayload,
+ })
+
+ await Task.send({
+ host: process.env.RELAY_HOST as string,
+ project: process.env.RELAY_PROJECT as string,
+ token: process.env.RELAY_TOKEN as string,
+ context,
+ message: lastPayload,
  })
  })
 }

packages/realtime-api/src/task/workers.ts

@@ -1,4 +1,5 @@
 import {
+ getLogger,
  sagaEffects,
  SagaIterator,
  SDKWorker,
@@ -7,17 +8,23 @@ import {
 import type { Task } from './Task'
 
 export const taskWorker: SDKWorker<Task> = function* (options): SagaIterator {
+ getLogger().trace('taskWorker started')
  const { channels } = options
  const { swEventChannel, pubSubChannel } = channels
- const action = yield sagaEffects.take(
- swEventChannel,
- (action: SDKActions) => {
- return action.type === 'queuing.relay.tasks'
- }
- )
 
- yield sagaEffects.put(pubSubChannel, {
- type: 'task.received',
- payload: action.payload.message,
- })
+ while (true) {
+ const action = yield sagaEffects.take(
+ swEventChannel,
+ (action: SDKActions) => {
+ return action.type === 'queuing.relay.tasks'
+ }
+ )
+
+ yield sagaEffects.put(pubSubChannel, {
+ type: 'task.received',
+ payload: action.payload.message,
+ })
+ }
+
+ getLogger().trace('taskWorker ended')
 }