Deliver a clean catalog at the pace your receiver accepts.

Browser implementation: runs the selected recipe in a module worker with a bounded local preview. The website adds the controls and progress display. Copied browser code uses the selected sample source; the downloaded app also accepts a local file. Both implementations share the same transformation and selected settings. Downloads include the complete project and preview dependency setup.

import { stream } from '@diakrio/converter/browser';

const config = {
  "recipe": "streaming-conversion",
  "inputFormat": "xml",
  "outputFormat": "csv",
  "count": 1000,
  "sinkDelayMs": 20
};

// Run this module inside a Web Worker.
async function runExample() {
  let input;
  if (config.sourceUrl) {
    const response = await fetch(config.sourceUrl);
    if (!response.ok || !response.body) throw new Error("Dataset could not be opened");
    input = response.body;
  } else input = createFixture(config.inputFormat, config.count, config.recipe);
  const job = startRecipe(stream, input, { ...config, expectedRows: config.count }, { progress: console.log });
  console.log(await job.result);
}

// Included shared recipe and fixture helpers.
export const recipeProfiles = {
  "streaming-conversion": {
    "name": "Turn a supplier catalog into store-ready CSV",
    "recordElement": "Product",
    "fields": [
      [
        "supplier_sku",
        "sku"
      ],
      [
        "product_name",
        "name"
      ],
      [
        "unit_price",
        "price",
        "f64"
      ],
      [
        "currency",
        "currency"
      ],
      [
        "available",
        "stock",
        "i64"
      ]
    ],
    "source": {
      "supplier_sku": "CAT-1",
      "product_name": "Desk lamp",
      "unit_price": "29.90",
      "currency": "EUR",
      "available": "12"
    },
    "expected": {
      "sku": "CAT-1",
      "name": "Desk lamp",
      "price": 29.9,
      "currency": "EUR",
      "stock": 12
    }
  }
};
export function getRecipe(id = 'large-xml') {
  const recipe = recipeProfiles[id];
  if (!recipe) throw new Error('Unknown recipe');
  return recipe;
}
export function fixtureRecord(id, index) {
  if (id === 'large-xml') return healthRecord(index);
  const recipe = getRecipe(id);
  const row = { ...recipe.source };
  const key = recipe.fields[0][0];
  row[key] = row[key].replace(/\d+$/, String(index + 1));
  // Wide event records make the cost of retaining irrelevant payloads visible.
  if (id === 'javascript-string-limits') row.payload = 'synthetic '.repeat(100);
  return row;
}
export function recipeTransform(id) {
  return { mode: 'replace', onMissingRequired: 'abort', onCoerceError: 'error', fields: getRecipe(id).fields.map(([originFieldName, targetFieldName, type]) => ({ originFieldName, targetFieldName, ...(id === 'large-xml' && targetFieldName === 'unit' ? { defaultValue: '' } : { required: true }), ...(type ? { coerce: { type } } : {}) })) };
}
// Original deterministic stress fixture, informed by the Apple Health XML shape.
// A 4,096-record cycle is repeated; this is not a person's longitudinal history.
export function healthRecord(index) {
  const n = index % 4096;
  const kinds = [['HKQuantityTypeIdentifierStepCount', 'count', String(50 + n % 200)], ['HKQuantityTypeIdentifierHeartRate', 'count/min', String(60 + n % 80)], ['HKQuantityTypeIdentifierBodyMass', 'kg', (65 + n % 200 / 10).toFixed(1)], ['HKCategoryTypeIdentifierSleepAnalysis', '', 'HKCategoryValueSleepAnalysisAsleepCore']];
  const [type, unit, value] = kinds[n % kinds.length];
  const stamp = minute => new Date(Date.UTC(2026, 0, 1, 0, minute)).toISOString().replace('T', ' ').replace('.000Z', ' +0000');
  return { type, sourceName: n % 2 ? 'Synthetic Phone' : 'Synthetic Watch', sourceVersion: '10.0', unit, creationDate: stamp(n + 1), startDate: stamp(n), endDate: stamp(n + 1), value };
}
export function healthXmlRecord(index) {
  const escape = value => String(value).replaceAll('&', '&amp;').replaceAll('"', '&quot;').replaceAll('<', '&lt;').replaceAll('>', '&gt;');
  const attrs = Object.entries(healthRecord(index)).map(([key, value]) => `${key}="${escape(value)}"`).join(' ');
  return index % 16 === 15 ? `  <Record ${attrs}><MetadataEntry key="HKTimeZone" value="Etc/UTC"/></Record>\n` : `  <Record ${attrs}/>\n`;
}
export function healthFixtureInfo(count = 1000) {
  if (!Number.isSafeInteger(count) || count < 1 || count > 100000000) throw new Error('Invalid health record count');
  const header = '<?xml version="1.0" encoding="UTF-8"?>\n<HealthData locale="en_US">\n  <ExportDate value="2026-01-04 00:00:00 +0000"/>\n';
  const footer = '</HealthData>\n';
  const encoder = new TextEncoder();
  const records = Array.from({ length: 4096 }, (_, index) => healthXmlRecord(index));
  const cycle = records.join('');
  const bytes = encoder.encode(header + footer).length + Math.floor(count / 4096) * encoder.encode(cycle).length + encoder.encode(records.slice(0, count % 4096).join('')).length;
  return { count, bytes, header, footer, records };
}
export function createHealthFixture(count = 1000) {
  const info = healthFixtureInfo(count);
  const encoder = new TextEncoder();
  const batches = Array.from({ length: 4 }, (_, batch) => encoder.encode(info.records.slice(batch * 1024, (batch + 1) * 1024).join('')));
  let index = -1;
  return new ReadableStream({ pull(controller) {
    if (index === -1) { index = 0; controller.enqueue(encoder.encode(info.header)); return; }
    if (index === count) { controller.enqueue(encoder.encode(info.footer)); controller.close(); return; }
    const size = Math.min(1024, count - index);
    controller.enqueue(size === 1024 ? batches[(index / 1024) % 4] : encoder.encode(info.records.slice(index % 4096, index % 4096 + size).join('')));
    index += size;
  } });
}

// Shared by the website worker and downloaded Node/browser examples.
// The preview is bounded; the complete output goes to the supplied incremental sink.

export function createFixture(format, count = 1000, recipeId = 'large-xml') {
  if (recipeId === 'large-xml' && format === 'xml') return createHealthFixture(count);
  const recipe = getRecipe(recipeId);
  let index = 0;
  let opened = false;
  const encoder = new TextEncoder();
  return new ReadableStream({
    pull(target) {
      let text = '';
      if (!opened) {
        opened = true;
        text = format === 'xml' ? '<records>' : format === 'json' ? '[' : Object.keys(recipe.source).join(',') + '\n';
      }
      const end = Math.min(index + 64, count);
      for (; index < end; index++) {
        const row = fixtureRecord(recipeId, index);
        text += format === 'xml'
          ? '<' + recipe.recordElement + '>' + Object.entries(row).map(([key, value]) => '<' + key + '>' + value + '</' + key + '>').join('') + '</' + recipe.recordElement + '>'
          : format === 'json' ? (index ? ',' : '') + JSON.stringify(row)
          : Object.values(row).map(value => '"' + String(value).replaceAll('"', '""') + '"').join(',') + '\n';
      }
      if (index === count) text += format === 'xml' ? '</records>' : format === 'json' ? ']' : '';
      target.enqueue(encoder.encode(text));
      if (index === count) target.close();
    }
  });
}
export function startRecipe(stream, input, config = {}, hooks = {}) {
  const recipeId = config.recipe ?? 'large-xml';
  const clock = () => performance.timeOrigin + performance.now();
  const startedAt = config.startedAt ?? clock();
  const decoder = new TextDecoder();
  let firstRowMs = null;
  let rows = 0;
  let preview = '';
  let lastReport = 0;
  let current = { bytesIn: 0, bytesOut: 0, recordsOut: 0 };
  const outputOnly = recipeId === 'large-xml' && config.outputFormat === 'csv';
  const samples = [];
  const snapshot = () => ({
    rows, bytesIn: current.bytesIn, bytesOut: current.bytesOut,
    firstRowMs, elapsedMs: clock() - startedAt, outputPreview: preview, samples,
    wasmMemoryBytes: current.wasmMemoryBytes ?? null,
  });
  const report = (force = false) => {
    if (force || clock() - lastReport >= 100) { lastReport = clock(); hooks.progress?.(snapshot()); }
  };
  const controller = stream(input, {
    backend: config.backend ?? 'auto',
    inputFormat: config.inputFormat ?? 'xml',
    outputFormat: config.outputFormat ?? 'json',
    ...(outputOnly ? { csvOutput: { columns: getRecipe(recipeId).fields.map(([, target]) => target) } } : {}),
    xmlConfig: { recordElement: getRecipe(recipeId).recordElement, trimText: true, includeAttributes: true },
    recordBatchSize: config.recordBatchSize ?? 256,
    ...(config.xmlParallel === undefined ? {} : { xmlParallel: config.xmlParallel }),
    ...(config.inputChunkBytes === undefined ? {} : { inputChunkBytes: config.inputChunkBytes }),
    ...(config.maxMemoryMB === undefined
      ? { limits: { maxRecordBytes: 1048576, maxOutputBytes: 4194304, maxRecordBatchBytes: 4194304, maxRecordsPerChunk: 4096 } }
      : { maxMemoryMB: config.maxMemoryMB }),
    transform: recipeTransform(recipeId),
    onRecords: outputOnly ? undefined : async (_job, records, stats) => {
      current = stats;
      if (records.length && firstRowMs === null) firstRowMs = clock() - startedAt;
      rows += records.length;
      for (const record of records) {
        if (samples.length === 5) break;
        if (JSON.stringify(record).length <= 1024) samples.push(record);
      }
      report();
    },
    async onBytes(bytes, stats) {
      current = stats;
      if (outputOnly) {
        rows = stats.recordsOut;
        if (rows && firstRowMs === null) firstRowMs = clock() - startedAt;
      }
      if (preview.length < 4096) preview += decoder.decode(bytes.subarray(0, Math.min(bytes.length, 4096 - preview.length)), { stream: true });
      if (config.sinkDelayMs) await new Promise(resolve => setTimeout(resolve, config.sinkDelayMs));
      await hooks.write?.(bytes);
      report();
    },
    onError() { /* result rejects; the host owns visible error handling */ }
  });
  const result = controller.done.then(async stats => {
    current = stats;
    await hooks.close?.();
    if (config.expectedRows !== undefined && rows !== config.expectedRows) throw new Error('Output record count does not match the fixture.');
    const final = snapshot();
    hooks.progress?.(final);
    return final;
  });
  return { controller, result };
}

await runExample();
// This browser example discards output after a bounded preview.