Adding more transforms and preparing for an initial release

Author: davedbase

.changeset/silver-clocks-sleep.md

@@ -0,0 +1,20 @@
+---
+"@solid-primitives/sse": minor
+---
+
+Initial release of `@solid-primitives/sse`.
+
+### Primitives
+
+- `makeSSE(url, options?)` — base non-reactive primitive. Creates an `EventSource`, attaches handlers, and returns `[source, cleanup]`. No Solid lifecycle dependency.
+- `createSSE(url, options?)` — reactive primitive. Accepts a static or signal URL, closes on owner disposal, and exposes `data`, `error`, `readyState`, `close`, and `reconnect`.
+- `makeSSEWorker(target, options?)` — runs the `EventSource` connection inside a Web Worker or SharedWorker, keeping network I/O off the main thread. The reactive API is identical to `createSSE`.
+
+### Built-in transformers
+
+- `json` — parse message data as a single JSON value
+- `ndjson` — parse newline-delimited JSON into an array
+- `lines` — split message data into a `string[]` by newline
+- `number` — parse message data as a number via `Number()`
+- `safe(transform, fallback?)` — fault-tolerant wrapper; returns `fallback` instead of throwing on bad input
+- `pipe(a, b)` — compose two transforms into one

packages/sse/README.md

@@ -14,6 +14,7 @@ Primitives for [Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web
 - [`makeSSE`](#makesse) — Base non-reactive primitive. Creates an `EventSource` and returns a cleanup function. No Solid lifecycle.
 - [`createSSE`](#createsse) — Reactive primitive. Accepts a reactive URL, integrates with Solid's owner lifecycle, and returns signals for `data`, `error`, and `readyState`.
 - [`makeSSEWorker`](./WORKERS.md) — Runs the SSE connection inside a Web Worker or SharedWorker. See [WORKERS.md](./WORKERS.md).
+- [Built-in transformers](./TRANSFORMS.md) — `json`, `ndjson`, `lines`, `number`, `safe`, `pipe`. See [TRANSFORMS.md](./TRANSFORMS.md).
 
 ## Installation
 
@@ -156,48 +157,16 @@ SSEReadyState.CLOSED     // 2
 
 ## Built-in transformers
 
-Three ready-made `transform` functions are exported for the most common SSE data formats.
+Ready-made `transform` functions for the most common SSE data formats. See [TRANSFORMS.md](./TRANSFORMS.md) for full documentation and examples.
 
-### `json`
-
-Parse the message data as a single JSON value. Equivalent to `JSON.parse` but named for consistency with the other transformers.
-
-```ts
-import { createSSE, json } from "@solid-primitives/sse";
-
-const { data } = createSSE<{ status: string; ts: number }>(url, { transform: json });
-// data() === { status: "ok", ts: 1718000000 }
-```
-
-### `ndjson`
-
-Parse the message data as [newline-delimited JSON](https://ndjson.org/) (NDJSON / JSON Lines). Each non-empty line is parsed as a separate JSON value and the transformer returns an array.
-
-Use this when the server batches multiple objects into one SSE event:
-
-```
-data: {"id":1,"type":"tick"}
-data: {"id":2,"type":"tick"}
-
-```
-
-```ts
-import { createSSE, ndjson } from "@solid-primitives/sse";
-
-const { data } = createSSE<TickEvent[]>(url, { transform: ndjson });
-// data() === [{ id: 1, type: "tick" }, { id: 2, type: "tick" }]
-```
-
-### `lines`
-
-Split the message data into individual lines, returning a `string[]`. Empty lines are filtered out. Useful for multi-line text events that are not JSON.
-
-```ts
-import { createSSE, lines } from "@solid-primitives/sse";
-
-const { data } = createSSE<string[]>(url, { transform: lines });
-// data() === ["line one", "line two"]
-```
+| Transformer | Description |
+|---|---|
+| [`json`](./TRANSFORMS.md#json) | Parse data as a single JSON value |
+| [`ndjson`](./TRANSFORMS.md#ndjson) | Parse newline-delimited JSON into an array |
+| [`lines`](./TRANSFORMS.md#lines) | Split data into a `string[]` by newline |
+| [`number`](./TRANSFORMS.md#number) | Parse data as a number via `Number()` |
+| [`safe(transform, fallback?)`](./TRANSFORMS.md#safetransform-fallback) | Fault-tolerant wrapper — returns `fallback` instead of throwing |
+| [`pipe(a, b)`](./TRANSFORMS.md#pipea-b) | Compose two transforms into one |
 
 ## Integration with `@solid-primitives/event-bus`
 

packages/sse/TRANSFORMS.md

@@ -0,0 +1,108 @@
+# Built-in transformers
+
+Ready-made `transform` functions for the most common SSE data formats. Pass one as the `transform` option to `createSSE`:
+
+```ts
+import { createSSE, json } from "@solid-primitives/sse";
+
+const { data } = createSSE<{ status: string }>(url, { transform: json });
+```
+
+---
+
+## `json`
+
+Parse the message data as a single JSON value. Equivalent to `JSON.parse` but named for consistency with the other transformers.
+
+```ts
+import { createSSE, json } from "@solid-primitives/sse";
+
+const { data } = createSSE<{ status: string; ts: number }>(url, { transform: json });
+// data() === { status: "ok", ts: 1718000000 }
+```
+
+---
+
+## `ndjson`
+
+Parse the message data as [newline-delimited JSON](https://ndjson.org/) (NDJSON / JSON Lines). Each non-empty line is parsed as a separate JSON value and the transformer returns an array.
+
+Use this when the server batches multiple objects into one SSE event:
+
+```
+data: {"id":1,"type":"tick"}
+data: {"id":2,"type":"tick"}
+
+```
+
+```ts
+import { createSSE, ndjson } from "@solid-primitives/sse";
+
+const { data } = createSSE<TickEvent[]>(url, { transform: ndjson });
+// data() === [{ id: 1, type: "tick" }, { id: 2, type: "tick" }]
+```
+
+---
+
+## `lines`
+
+Split the message data into individual lines, returning a `string[]`. Empty lines are filtered out. Useful for multi-line text events that are not JSON.
+
+```ts
+import { createSSE, lines } from "@solid-primitives/sse";
+
+const { data } = createSSE<string[]>(url, { transform: lines });
+// data() === ["line one", "line two"]
+```
+
+---
+
+## `number`
+
+Parse the message data as a number using `Number()` semantics. Handy for streams that emit counters, progress percentages, sensor readings, or prices.
+
+```ts
+import { createSSE, number } from "@solid-primitives/sse";
+
+const { data } = createSSE<number>(url, { transform: number });
+// data() === 42
+```
+
+Note: follows `Number()` coercion — an empty string becomes `0` and non-numeric strings become `NaN`.
+
+---
+
+## `safe(transform, fallback?)`
+
+Wraps any transform in a `try/catch`. When the inner transform throws, `safe` returns `fallback` instead of propagating the error. This keeps the stream alive across malformed events.
+
+```ts
+import { createSSE, json, number, safe } from "@solid-primitives/sse";
+
+// Returns undefined on a bad event instead of throwing
+const { data } = createSSE<MyEvent>(url, { transform: safe(json) });
+
+// With an explicit fallback value
+const { data } = createSSE<number>(url, { transform: safe(number, 0) });
+```
+
+---
+
+## `pipe(a, b)`
+
+Composes two transforms into one: the output of `a` is passed as the input of `b`. Useful for building custom transforms from existing primitives without writing anonymous functions.
+
+```ts
+import { createSSE, ndjson, json, safe, pipe } from "@solid-primitives/sse";
+
+// Parse NDJSON then keep only "tick" rows
+type RawEvent = { type: string };
+const { data } = createSSE<RawEvent[]>(url, {
+  transform: pipe(ndjson<RawEvent>, rows => rows.filter(r => r.type === "tick")),
+});
+
+// Safe JSON with a post-processing step
+const { data } = createSSE<string>(url, {
+  transform: pipe(safe(json<{ label: string }>), ev => ev?.label ?? ""),
+});
+```

packages/sse/package.json

@@ -22,7 +22,10 @@
       "makeSSEWorker",
       "json",
       "ndjson",
-      "lines"
+      "lines",
+      "number",
+      "safe",
+      "pipe"
     ],
     "category": "Network"
   },

packages/sse/src/index.ts

@@ -11,4 +11,4 @@ export {
   type SSEReturn,
 } from "./sse.js";
 
-export { json, ndjson, lines } from "./transform.js";
+export { json, ndjson, lines, number, safe, pipe } from "./transform.js";

packages/sse/src/transform.ts

@@ -57,3 +57,65 @@ export const ndjson = <T>(raw: string): T[] =>
  * ```
  */
 export const lines = (raw: string): string[] => raw.split("\n").filter(line => line !== "");
+
+/**
+ * Parse SSE message data as a number using `Number()` semantics.
+ *
+ * Use this for streams that emit plain numeric values: counters, progress
+ * percentages, sensor readings, prices, etc.
+ *
+ * ```ts
+ * const { data } = createSSE<number>(url, { transform: number });
+ * // data() === 42
+ * ```
+ *
+ * Note: follows `Number()` coercion — `""` → `0`, non-numeric strings → `NaN`.
+ */
+export const number = (raw: string): number => Number(raw);
+
+/**
+ * Wrap any transform in a `try/catch` so that a malformed event does not
+ * throw; instead it returns `fallback` (default `undefined`).
+ *
+ * ```ts
+ * // Returns undefined on bad input instead of throwing
+ * const { data } = createSSE<MyEvent>(url, { transform: safe(json) });
+ *
+ * // With an explicit fallback value
+ * const { data } = createSSE<number>(url, { transform: safe(number, 0) });
+ * ```
+ */
+export function safe<T>(transform: (raw: string) => T): (raw: string) => T | undefined;
+export function safe<T>(transform: (raw: string) => T, fallback: T): (raw: string) => T;
+export function safe<T>(
+  transform: (raw: string) => T,
+  fallback?: T,
+): (raw: string) => T | undefined {
+  return (raw: string): T | undefined => {
+    try {
+      return transform(raw);
+    } catch {
+      return fallback;
+    }
+  };
+}
+
+/**
+ * Compose two transforms into one: the output of `a` is passed as the input
+ * of `b`.
+ *
+ * ```ts
+ * // Parse NDJSON then keep only "tick" events
+ * const { data } = createSSE<TickEvent[]>(url, {
+ *   transform: pipe(ndjson<RawEvent>, rows => rows.filter(r => r.type === "tick")),
+ * });
+ *
+ * // Safe JSON followed by a post-processing step
+ * const { data } = createSSE<string>(url, {
+ *   transform: pipe(safe(json<{ label: string }>), ev => ev?.label ?? ""),
+ * });
+ * ```
+ */
+export function pipe<A, B>(a: (raw: string) => A, b: (a: A) => B): (raw: string) => B {
+  return (raw: string): B => b(a(raw));
+}

packages/sse/test/transform.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest";
-import { json, ndjson, lines } from "../src/transform.js";
+import { json, ndjson, lines, number, safe, pipe } from "../src/transform.js";
 
 // ── json ──────────────────────────────────────────────────────────────────────
 
@@ -80,3 +80,107 @@ describe("lines", () => {
     expect(lines("")).toEqual([]);
   });
 });
+
+// ── number ────────────────────────────────────────────────────────────────────
+
+describe("number", () => {
+  it("parses an integer string", () => {
+    expect(number("42")).toBe(42);
+  });
+
+  it("parses a float string", () => {
+    expect(number("3.14")).toBe(3.14);
+  });
+
+  it("parses a negative number", () => {
+    expect(number("-7")).toBe(-7);
+  });
+
+  it("converts an empty string to 0", () => {
+    expect(number("")).toBe(0);
+  });
+
+  it("converts a non-numeric string to NaN", () => {
+    expect(number("not a number")).toBeNaN();
+  });
+});
+
+// ── safe ──────────────────────────────────────────────────────────────────────
+
+describe("safe", () => {
+  it("returns the transform result when successful", () => {
+    expect(safe(json)('{"a":1}')).toEqual({ a: 1 });
+  });
+
+  it("returns undefined when the inner transform throws (no fallback)", () => {
+    expect(safe(json)("bad json")).toBeUndefined();
+  });
+
+  it("returns the fallback when the inner transform throws", () => {
+    expect(safe(json, null)("bad json")).toBeNull();
+  });
+
+  it("returns a numeric fallback on error", () => {
+    expect(safe(number, 0)("NaN")).toBe(NaN); // Number("NaN") === NaN, not throwing
+    // Demonstrate fallback with a throwing transform:
+    const throwing = (_: string): number => {
+      throw new Error("fail");
+    };
+    expect(safe(throwing, -1)("any")).toBe(-1);
+  });
+
+  it("is composable: safe(ndjson) returns undefined on invalid line", () => {
+    expect(safe(ndjson)('{"a":1}\nbad')).toBeUndefined();
+  });
+
+  it("overload without fallback infers T | undefined return type", () => {
+    const result: { a: number } | undefined = safe(json<{ a: number }>)('{"a":1}');
+    expect(result).toEqual({ a: 1 });
+  });
+
+  it("overload with fallback infers T return type", () => {
+    const result: { a: number } = safe(json<{ a: number }>, { a: 0 })("bad");
+    expect(result).toEqual({ a: 0 });
+  });
+});
+
+// ── pipe ──────────────────────────────────────────────────────────────────────
+
+describe("pipe", () => {
+  it("passes the string through both transforms in order", () => {
+    const upper = (s: string) => s.toUpperCase();
+    const exclaim = (s: string) => `${s}!`;
+    expect(pipe(upper, exclaim)("hello")).toBe("HELLO!");
+  });
+
+  it("composes json with a post-processing step", () => {
+    const getLabel = pipe(json<{ label: string }>, ev => ev.label);
+    expect(getLabel('{"label":"tick"}')).toBe("tick");
+  });
+
+  it("composes ndjson with a filter step", () => {
+    type Row = { type: string };
+    const ticks = pipe(
+      ndjson<Row>,
+      rows => rows.filter(r => r.type === "tick"),
+    );
+    expect(ticks('{"type":"tick"}\n{"type":"other"}\n{"type":"tick"}')).toEqual([
+      { type: "tick" },
+      { type: "tick" },
+    ]);
+  });
+
+  it("composes safe(json) with a fallback mapping", () => {
+    const getLabel = pipe(safe(json<{ label: string }>), ev => ev?.label ?? "");
+    expect(getLabel('{"label":"ok"}')).toBe("ok");
+    expect(getLabel("bad")).toBe("");
+  });
+
+  it("infers the correct return type", () => {
+    const toLength: (raw: string) => number = pipe(
+      (s: string) => s.split(","),
+      arr => arr.length,
+    );
+    expect(toLength("a,b,c")).toBe(3);
+  });
+});