JSONL Streaming: Process Data in Real-Time

import { createReadStream } from 'node:fs';
import { createInterface } from 'node:readline';

async function streamJsonl(filePath, onRecord) {
  const rl = createInterface({
    input: createReadStream(filePath, 'utf-8'),
    crlfDelay: Infinity,
  });

  let count = 0;
  for await (const line of rl) {
    const trimmed = line.trim();
    if (!trimmed) continue;
    try {
      const record = JSON.parse(trimmed);
      await onRecord(record);
      count++;
    } catch (err) {
      console.error(`Skipping invalid line: ${err.message}`);
    }
  }
  return count;
}

// Usage: process each record as it arrives
const total = await streamJsonl('events.jsonl', async (record) => {
  console.log(`Event: ${record.type} at ${record.timestamp}`);
});
console.log(`Processed ${total} records`);

import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';

// Split incoming chunks into individual lines
class LineSplitter extends Transform {
  constructor() {
    super({ readableObjectMode: true });
    this.buffer = '';
  }
  _transform(chunk, encoding, cb) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop();
    for (const line of lines) {
      if (line.trim()) this.push(line.trim());
    }
    cb();
  }
  _flush(cb) {
    if (this.buffer.trim()) this.push(this.buffer.trim());
    cb();
  }
}

// Parse, transform, and re-serialize each record
class JsonlTransform extends Transform {
  constructor(transformFn) {
    super({ objectMode: true });
    this.transformFn = transformFn;
  }
  _transform(line, encoding, cb) {
    try {
      const record = JSON.parse(line);
      const result = this.transformFn(record);
      if (result) cb(null, JSON.stringify(result) + '\n');
      else cb(); // Filter out null results
    } catch (err) {
      cb(); // Skip invalid lines
    }
  }
}

// Build the pipeline: read -> split -> transform -> write
await pipeline(
  createReadStream('input.jsonl', 'utf-8'),
  new LineSplitter(),
  new JsonlTransform((record) => {
    if (record.level === 'error') {
      return { ...record, flagged: true, reviewedAt: new Date().toISOString() };
    }
    return null; // Filter out non-error records
  }),
  createWriteStream('errors.jsonl', 'utf-8')
);
console.log('Pipeline complete');

import express from 'express';

const app = express();

// Simulate a real-time data source
function* generateMetrics() {
  let id = 0;
  while (true) {
    yield {
      id: ++id,
      cpu: Math.random() * 100,
      memory: Math.random() * 16384,
      timestamp: new Date().toISOString(),
    };
  }
}

app.get('/api/stream/metrics', (req, res) => {
  // SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('Access-Control-Allow-Origin', '*');
  res.flushHeaders();

  const metrics = generateMetrics();

  const interval = setInterval(() => {
    const record = metrics.next().value;
    // Each SSE event contains one JSONL record
    res.write(`data: ${JSON.stringify(record)}\n\n`);
  }, 1000);

  req.on('close', () => {
    clearInterval(interval);
    res.end();
  });
});

app.listen(3000, () => console.log('SSE server on :3000'));

// Connect to the SSE endpoint
const source = new EventSource('/api/stream/metrics');

// Parse each JSONL record as it arrives
source.onmessage = (event) => {
  const record = JSON.parse(event.data);
  console.log(`[${record.timestamp}] CPU: ${record.cpu.toFixed(1)}%`);

  // Update your UI in real time
  updateDashboard(record);
};

source.onerror = (err) => {
  console.error('SSE connection error:', err);
  // EventSource automatically reconnects
};

// For Fetch-based SSE (more control over the stream)
async function fetchSSE(url, onRecord) {
  const response = await fetch(url);
  const reader = response.body
    .pipeThrough(new TextDecoderStream())
    .getReader();

  let buffer = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += value;
    const events = buffer.split('\n\n');
    buffer = events.pop();
    for (const event of events) {
      const dataLine = event.split('\n').find(l => l.startsWith('data: '));
      if (dataLine) {
        const record = JSON.parse(dataLine.slice(6));
        onRecord(record);
      }
    }
  }
}

await fetchSSE('/api/stream/metrics', (record) => {
  console.log('Metric:', record);
});