| 1428 | } |
| 1429 | |
| 1430 | const onStream = async (streamingExec: StreamingExecution) => { |
| 1431 | const blockId = (streamingExec.execution as any).blockId |
| 1432 | |
| 1433 | const reader = streamingExec.stream.getReader() |
| 1434 | const decoder = new TextDecoder() |
| 1435 | const cancelReader = () => { |
| 1436 | void reader.cancel(timeoutController.signal.reason).catch(() => {}) |
| 1437 | } |
| 1438 | |
| 1439 | try { |
| 1440 | if (timeoutController.signal.aborted || isStreamClosed) return |
| 1441 | timeoutController.signal.addEventListener('abort', cancelReader, { once: true }) |
| 1442 | |
| 1443 | while (true) { |
| 1444 | if (timeoutController.signal.aborted || isStreamClosed) break |
| 1445 | const { done, value } = await reader.read() |
| 1446 | if (timeoutController.signal.aborted || isStreamClosed) break |
| 1447 | if (done) break |
| 1448 | |
| 1449 | const chunk = decoder.decode(value, { stream: true }) |
| 1450 | await sendEvent({ |
| 1451 | type: 'stream:chunk', |
| 1452 | timestamp: new Date().toISOString(), |
| 1453 | executionId, |
| 1454 | workflowId, |
| 1455 | data: { blockId, chunk }, |
| 1456 | }) |
| 1457 | } |
| 1458 | |
| 1459 | if (!timeoutController.signal.aborted && !isStreamClosed) { |
| 1460 | await sendEvent({ |
| 1461 | type: 'stream:done', |
| 1462 | timestamp: new Date().toISOString(), |
| 1463 | executionId, |
| 1464 | workflowId, |
| 1465 | data: { blockId }, |
| 1466 | }) |
| 1467 | } |
| 1468 | } catch (error) { |
| 1469 | if (!timeoutController.signal.aborted && !isStreamClosed) { |
| 1470 | reqLogger.error('Error streaming block content:', error) |
| 1471 | } |
| 1472 | } finally { |
| 1473 | timeoutController.signal.removeEventListener('abort', cancelReader) |
| 1474 | try { |
| 1475 | await reader.cancel().catch(() => {}) |
| 1476 | } catch {} |
| 1477 | } |
| 1478 | } |
| 1479 | |
| 1480 | const metadata: ExecutionMetadata = { |
| 1481 | requestId, |