feat: usefetch support sse

Author: yiludege

package.json

@@ -64,7 +64,7 @@
     "husky": "^9.1.6",
     "jsdom": "^25.0.1",
     "lint-staged": "^15.2.10",
-    "msw": "1.0.1",
+    "msw": "^2.10.5",
     "node-fetch": "^3.3.2",
     "prettier": "^3.3.3",
     "sass": "^1.81.0",

packages/core/useFetch/index.ts

@@ -285,7 +285,34 @@ export function useFetch<T>(
         response.value = fetchResponse;
         statusCode.value = fetchResponse.status;
 
-        responseData = await fetchResponse.clone()[config.type]();
+        const isStream =
+          fetchResponse.headers.get("Content-Type") === "text/event-stream" &&
+          fetchResponse.body instanceof ReadableStream;
+
+        // handle stream response
+        if (isStream) {
+          const reader = fetchResponse
+            .clone()
+            .body?.getReader() as ReadableStreamDefaultReader<
+            Uint8Array<ArrayBuffer>
+          >;
+          const decoder = new TextDecoder();
+          let received = "";
+
+          while (true) {
+            const { done, value } = await reader.read();
+
+            if (done) break;
+
+            const chunk = decoder.decode(value, { stream: true });
+            promise$.next(chunk as T);
+            received += chunk;
+          }
+
+          responseData = received;
+        } else {
+          responseData = await fetchResponse.clone()[config.type]();
+        }
 
         // see: https://www.tjvantoll.com/2015/09/13/fetch-and-errors/
         if (!fetchResponse.ok) {
@@ -310,7 +337,9 @@ export function useFetch<T>(
               cacheSetting.expiration,
             );
         }
-        promise$.next(Promise.resolve(responseData));
+        if (!isStream) {
+          promise$.next(Promise.resolve(responseData));
+        }
 
         responseEvent.trigger(fetchResponse);
         return responseData;

packages/core/useFetch/test/mockServer.ts

@@ -4,28 +4,127 @@
  */
 
 import { setupServer } from "msw/node";
-import type { RestContext, RestRequest } from "msw";
-import { rest } from "msw";
+import { http, HttpResponse } from "msw";
 import { afterAll, afterEach, beforeAll } from "vitest";
 
 const defaultJsonMessage = { hello: "world" };
 const defaultTextMessage = "Hello World";
 const baseUrl = "https://example.com";
 
-function commonTransformers(req: RestRequest, _: any, ctx: RestContext) {
-  const t: any[] = [];
-  const qs = req.url.searchParams;
+/**
+ * Create a Server-Sent Events (SSE) stream response
+ * @param count Number of events to send (default: 3)
+ * @param interval Interval between events in milliseconds (default: 1000)
+ * @param customData Custom data to send in events
+ */
+function createSSEStream(count = 3, interval = 1000, customData?: string) {
+  const encoder = new TextEncoder();
+
+  const stream = new ReadableStream({
+    async start(controller) {
+      try {
+        for (let eventCount = 1; eventCount <= count; eventCount++) {
+          let message;
+          if (customData) {
+            if (count <= customData.length) {
+              // Split customData into count parts
+              const partSize = Math.ceil(customData.length / count);
+              const startIndex = (eventCount - 1) * partSize;
+              const endIndex = Math.min(
+                startIndex + partSize,
+                customData.length,
+              );
+              message = customData.slice(startIndex, endIndex);
+            } else {
+              // Send one character at a time if count > customData length
+              if (eventCount <= customData.length) {
+                message = customData[eventCount - 1];
+              } else {
+                // If we've sent all characters, send empty or repeat last char
+                message = "";
+              }
+            }
+          } else {
+            // Default structure
+            message = `Event ${eventCount}`;
+          }
+
+          const data = `{"type":"data","message":"${message}","timestamp":"${new Date().toISOString()}","count":${eventCount},"total":${count}}`;
+          const event = `data: ${data}\n\n`;
+
+          controller.enqueue(encoder.encode(event));
+
+          // Wait for interval before next event (except for last event)
+          if (eventCount < count) {
+            await new Promise((resolve) => setTimeout(resolve, interval));
+          }
+        }
+
+        // Send final completion event
+        const finalEvent = `data: {"type":"complete","message":"Stream completed","count":${count}}\n\n`;
+        controller.enqueue(encoder.encode(finalEvent));
+        controller.close();
+      } catch (error) {
+        controller.error(error);
+      }
+    },
+
+    cancel() {
+      // Cleanup if stream is cancelled
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+      "Access-Control-Allow-Origin": "*",
+      "Access-Control-Allow-Headers": "Cache-Control",
+    },
+  });
+}
+
+function createResponse(request: Request) {
+  const url = new URL(request.url);
+  const qs = url.searchParams;
+
+  let response: Response;
 
-  if (qs.get("delay")) t.push(ctx.delay(Number(qs.get("delay"))));
-  if (qs.get("status")) t.push(ctx.status(Number(qs.get("status"))));
-  if (qs.get("text") != null) {
-    t.push(ctx.text(qs.get("text") ?? defaultTextMessage));
+  if (qs.get("stream") != null) {
+    // Handle SSE stream response
+    const count = qs.get("count") ? Number(qs.get("count")) : 3;
+    const interval = qs.get("interval") ? Number(qs.get("interval")) : 1000;
+    const customData = qs.get("data") || undefined;
+    response = createSSEStream(count, interval, customData);
+  } else if (qs.get("text") != null) {
+    response = HttpResponse.text(qs.get("text") ?? defaultTextMessage);
   } else if (qs.get("json") != null) {
     const jsonVal = qs.get("json");
-    const jsonTransformer = ctx.json(jsonVal?.length ? JSON.parse(jsonVal) : defaultJsonMessage);
-    t.push(jsonTransformer);
+    const jsonData = jsonVal?.length ? JSON.parse(jsonVal) : defaultJsonMessage;
+    response = HttpResponse.json(jsonData);
+  } else {
+    response = HttpResponse.json(defaultJsonMessage);
+  }
+
+  // Apply status code if specified
+  if (qs.get("status")) {
+    const status = Number(qs.get("status"));
+    response = new HttpResponse(response.body, {
+      status,
+      headers: response.headers,
+    });
   }
-  return t;
+
+  // Apply delay if specified
+  if (qs.get("delay")) {
+    const delay = Number(qs.get("delay"));
+    return new Promise<Response>((resolve) => {
+      setTimeout(() => resolve(response), delay);
+    });
+  }
+
+  return response;
 }
 
 /**
@@ -36,31 +135,57 @@ function commonTransformers(req: RestRequest, _: any, ctx: RestContext) {
  * @example https://example.com?delay=1000 will respond in 1000ms.
  * @example https://example.com?status=301&text=thanks&delay=1000
  *          will respond in 1000ms with statusCode 300 and the response body "thanks" as a string
+ * @example https://example.com?stream will respond with a Server-Sent Events (SSE) stream
+ * @example https://example.com?stream&count=5&interval=500
+ *          will send 5 SSE events with 500ms interval between each event
+ * @example https://example.com?stream&data=hello
+ *          will send SSE events with custom message "hello"
+ * @example https://example.com?stream&count=5&data=hello
+ *          will send each character "h", "e", "l", "l", "o" in separate events
+ * @example https://example.com?stream&count=2&data=hello
+ *          will split "hello" into 2 parts: "hel" and "lo"
  */
 const server = setupServer(
-  rest.post(baseUrl, (req, res, ctx) => {
-    // Support all the normal examples (delay, status, text, and json)
-    const t = commonTransformers(req, res, ctx);
+  http.post(baseUrl, async ({ request }) => {
+    const url = new URL(request.url);
+    const qs = url.searchParams;
 
-    // Echo back the request payload
-    if (typeof req.body === "number" || typeof req.body === "string") t.push(ctx.text(String(req.body)));
-    else t.push(ctx.json(req.body));
+    // Check if query parameters specify response format
+    if (
+      qs.get("text") != null ||
+      qs.get("json") != null ||
+      qs.get("stream") != null ||
+      qs.get("status") ||
+      qs.get("delay")
+    ) {
+      const response = createResponse(request);
+      return response instanceof Promise ? await response : response;
+    }
 
-    return res(...t);
+    // Default behavior: Echo back the request payload for POST requests
+    const body = await request.text();
+    try {
+      const jsonBody = JSON.parse(body);
+      return HttpResponse.json(jsonBody);
+    } catch {
+      return HttpResponse.text(body || defaultTextMessage);
+    }
   }),
 
-  rest.get(baseUrl, (req, res, ctx) => {
-    return res(...commonTransformers(req, res, ctx));
+  http.get(baseUrl, async ({ request }) => {
+    const response = createResponse(request);
+    return response instanceof Promise ? await response : response;
   }),
 
   // Another duplicate route for the sole purpose of re-triggering requests on url change.
-  rest.get(`${baseUrl}/test`, (req, res, ctx) => {
-    return res(...commonTransformers(req, res, ctx));
+  http.get(`${baseUrl}/test`, async ({ request }) => {
+    const response = createResponse(request);
+    return response instanceof Promise ? await response : response;
   }),
 
-  rest.get(`${baseUrl}/url`, (req, res, ctx) => {
-    return res(ctx.text(req.url.toString()));
-  })
+  http.get(`${baseUrl}/url`, ({ request }) => {
+    return HttpResponse.text(request.url.toString());
+  }),
 );
 
 beforeAll(() => server.listen());