example for cloudflare worker server-sent events

sse-worker.js

/**
 * Welcome to Cloudflare Workers! This is your first worker.
 *
 * - Run "npm run dev" in your terminal to start a development server
 * - Open a browser tab at http://localhost:8787/ to see your worker in action
 * - Run "npm run deploy" to publish your worker
 *
 * Learn more at https://developers.cloudflare.com/workers/
 */

// Cloudflare Worker code

/**
 * listen for fetch event
 */
addEventListener('fetch', event => {
  event.respondWith(fetchAndApply(event.request));
});

/**
 * apply event-stream to response
 * @param {Request} request 
 * @returns Response
 */
async function fetchAndApply(request) {
  const { readable, writable } = new TransformStream();
  const headers = new Headers();
  headers.append('Content-Type', 'text/event-stream');
  headers.append('Cache-Control', 'no-cache');
  headers.append('Connection', 'keep-alive');
  headers.append('Access-Control-Allow-Origin', '*');
  headers.append('Access-Control-Allow-Headers', 'Origin, X-Requested-With, Content-Type, Accept');
  
  const init = { status: 200, statusText: "ok", headers };

  writeToStream(writable);
  return new Response(readable, init);
}

/**
 * write data to writer stream
 * @param {WritableStream} writable 
 */
async function writeToStream(writable) {
  const writer = writable.getWriter();
  let count = 0;
  let eventId = `id-${++count}`;
  let msgBody = JSON.stringify({ status: true, text: "Hello to SSE message", time: new Date().toISOString() });

  // send hello message
  await constructSSE(writer, eventId, "userConnected", msgBody);

  // send message every 5 seconds
  const intervalId = setInterval(async () => {
    eventId = `id-${++count}`;
    msgBody = JSON.stringify({ status: true, text: `Repeat message: ${count}`, time: new Date().toISOString() });
    await constructSSE(writer, eventId, "userMessage", msgBody);
  }, 5000);

  // Clean up on stream close
  writable.closed.then(() => {
    clearInterval(intervalId);
    writer.releaseLock();
  }).catch(err => {
    console.error("Stream closed with error:", err);
    clearInterval(intervalId);
  });
}

/**
 * construct message in SSE format
 * @param {WritableStreamDefaultWriter} writer 
 * @param {string|number} eventId 
 * @param {string} eventType 
 * @param {string} msgBody 
 */
async function constructSSE(writer, eventId, eventType, msgBody) {
  const encoder = new TextEncoder();

  await writer.write(encoder.encode(`id: ${eventId}\n`));
  if (eventType) {
    await writer.write(encoder.encode(`event: ${eventType}\n`));
  }
  await writer.write(encoder.encode(`data: ${msgBody}\n\n`));
}