k6 version

// sse-load.js
import http from 'k6/http';
import { check } from 'k6';
import { ReadableStream } from 'k6/experimental/streams';
import { Counter, Trend } from 'k6/metrics';

// Custom metrics
const sseEvents = new Counter('sse_events_received');
const eventLag  = new Trend('sse_event_lag_ms');

export const options = {
  scenarios: {
    // constant-vus holds N virtual users open for the full duration;
    // each VU maps to one persistent SSE connection.
    // See https://grafana.com/docs/k6/latest/using-k6/scenarios/executors/constant-vus/
    sustained_streams: {
      executor: 'constant-vus',
      vus: 50,
      duration: '60s',
    },
  },
  thresholds: {
    // TTFB: first byte of the event stream arrives within 500 ms for 95% of VUs
    // Metric reference: https://grafana.com/docs/k6/latest/using-k6/metrics/reference/
    'http_req_waiting': ['p(95)<500'],
    // Connection-slot starvation signal: VUs blocked waiting for a free TCP
    // slot should be negligible
    'http_req_blocked': ['p(99)<100'],
    // Overall error rate
    'http_req_failed': ['rate<0.01'],
  },
};

export default async function () {
  const startMs = Date.now();

  // Long-lived GET; set timeout generously (k6 default is 60 s per
  // https://grafana.com/docs/k6/latest/javascript-api/k6-http/params/ ).
  // responseType 'text' streams body as string.
  const res = http.get('http://localhost:3000/api/events', {
    headers: { Accept: 'text/event-stream', 'Cache-Control': 'no-cache' },
    timeout: '70s',
  });

  check(res, {
    'status 200':            (r) => r.status === 200,
    'correct content-type':  (r) =>
      (r.headers['Content-Type'] || '').includes('text/event-stream'),
  });

  // Parse the body that arrived before k6 closed the response.
  // For a true streaming parse, replace this block with a ReadableStream
  // wrapping a streaming HTTP client (see Step 3).
  const lines = (res.body || '').split('\n');
  let count = 0;
  for (const line of lines) {
    if (line.startsWith('data:')) {
      count++;
      eventLag.add(Date.now() - startMs);
    }
  }
  sseEvents.add(count);
}

import { ReadableStream } from 'k6/experimental/streams';
import { Counter } from 'k6/metrics';

const sseEvents = new Counter('sse_events_received');
const TARGET_EVENTS = 10; // drain after this many, then move on

export default async function () {
  // Open the SSE connection
  const res = http.get('http://localhost:3000/api/events', {
    headers: { Accept: 'text/event-stream' },
    timeout: '120s',
    responseType: 'none', // discard buffered body; we read from the stream
  });

  let buffer = '';
  let received = 0;

  const stream = new ReadableStream({
    async pull(controller) {
      // In a real integration, wire this to the chunked HTTP response.
      // k6's http module does not expose a streaming body reader natively;
      // for production use, combine with an xk6 extension or poll a shared
      // channel between the VU and a background goroutine.
      if (received >= TARGET_EVENTS) {
        controller.close();
        return;
      }
      // Simulate processing: parse lines from accumulated buffer
      // Replace this with actual chunk reads from your transport layer.
      controller.enqueue(buffer);
    },
  });

  const reader = stream.getReader();
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const lines = (value || '').split('\n');
    for (const line of lines) {
      if (line.startsWith('data:')) {
        received++;
        sseEvents.add(1);
      }
    }
  }
}

export const options = {
  scenarios: {
    ramp_streams: {
      executor: 'ramping-vus',
      startVUs: 10,
      stages: [
        { duration: '30s', target: 100  },
        { duration: '60s', target: 500  },
        { duration: '60s', target: 1000 },
        { duration: '30s', target: 0    },
      ],
    },
  },
  thresholds: {
    'http_req_waiting':  ['p(95)<500'],
    'http_req_blocked':  ['p(99)<100'],
    'http_req_failed':   ['rate<0.02'],
    'data_received':     ['count>0'],
  },
};

export const options = {
  scenarios: {
    churn: {
      executor: 'constant-arrival-rate',
      rate: 20,       // 20 new SSE connections per second
      timeUnit: '1s',
      duration: '60s',
      preAllocatedVUs: 40,
    },
  },
};

export default function () {
  // Each iteration: connect, receive 3 events, disconnect.
  const res = http.get('http://localhost:3000/api/events', {
    headers: { Accept: 'text/event-stream' },
    timeout: '10s',
  });
  check(res, { 'status 200': (r) => r.status === 200 });
  // Rapid disconnect after partial read - stresses server close path
}

k6 run sse-load.js

# HTTP/1.1 target
k6 run -e TARGET=http://localhost:3000/api/events sse-load.js

# HTTP/2 target (same VU count)
k6 run -e TARGET=https://localhost:3000/api/events sse-load.js

scenarios: (100.00%) 1 scenario, 50 max VUs, 1m30s max duration
default: 50 looping VUs for 1m0s (gracefulStop: 30s)

http_req_blocked............: avg=1.2ms   p(99)=8ms
http_req_connecting.........: avg=3.1ms   p(95)=12ms
http_req_waiting............: avg=42ms    p(95)=180ms
http_req_failed.............: 0.00%
data_received...............: 14 MB  230 kB/s
sse_events_received.........: 4800

thresholds:
  http_req_waiting p(95)<500    - OK
  http_req_blocked p(99)<100    - OK
  http_req_failed  rate<0.01    - OK

def stream(request):
    last_id = int(request.headers.get("Last-Event-ID", "0"))
    for evt in fetch_events_since(last_id):
        yield f"id: {evt.id}\nevent: {evt.type}\ndata: {evt.json()}\n\n"

import requests

# parse_until_count(r, n): read the stream until n complete events parse, return them
def test_replay_via_last_event_id():
    # First connection - read 5 events, then close
    with requests.get("http://localhost:8080/stream", stream=True) as r:
        events = parse_until_count(r, 5)
        last_id = events[-1]["id"]

    # Reconnect with Last-Event-ID
    headers = {"Last-Event-ID": last_id}
    with requests.get("http://localhost:8080/stream", stream=True, headers=headers) as r:
        replay = parse_until_count(r, 1)
        assert int(replay[0]["id"]) > int(last_id)

retry: 10000

test('client honors retry: 10000 on disconnect', async ({ page }) => {
  // Server emits retry: 10000, then closes
  const reconnectMs = await page.evaluate(() => {
    return new Promise<number>((resolve) => {
      const es = new EventSource('/api/stream-with-retry');
      let openTime = 0;
      es.onopen = () => {
        if (openTime === 0) {
          openTime = performance.now();
        } else {
          es.close();
          resolve(performance.now() - openTime);
        }
      };
    });
  });
  // Allow +/-20% slack
  expect(reconnectMs).toBeGreaterThanOrEqual(8000);
  expect(reconnectMs).toBeLessThanOrEqual(12000);
});

def stream(request):
    if user_unsubscribed(request):
        return Response(status=204)
    # ... event stream ...

test('client stops reconnecting after server returns 204', async ({ page }) => {
  // Server returns 204 immediately
  const states = await page.evaluate(() => {
    return new Promise<number[]>((resolve) => {
      const es = new EventSource('/api/stream-204');
      const seen: number[] = [];
      const interval = setInterval(() => seen.push(es.readyState), 100);
      setTimeout(() => {
        clearInterval(interval);
        resolve(seen);
      }, 2000);
    });
  });
  expect(states[states.length - 1]).toBe(2); // CLOSED
});

test('app uses single EventSource for fan-out', async ({ page }) => {
  await page.goto('https://localhost:3000/dashboard');

  const eventSourceCount = await page.evaluate(() =>
    performance.getEntriesByType('resource')
      .filter((r) => r.name.includes('/api/stream'))
      .length
  );
  expect(eventSourceCount).toBe(1);
});