Skip to content
rachel/workerd
File

Blob: src/workerd/api/tests/eventsource-test.js

javascript614 lines
1// Copyright (c) 2017-2024 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4import { strictEqual, ok, throws } from 'node:assert';
5 
6export const acceptEventStreamTest = {
7  async test(ctrl, env) {
8    const eventsource = new EventSource(
9      'http://example.org/accept-event-stream',
10      { fetcher: env.subrequest }
11    );
12    ok(eventsource instanceof EventTarget);
13    strictEqual(eventsource.readyState, EventSource.CONNECTING);
14    const { promise, resolve } = Promise.withResolvers();
15    let opened = false;
16    eventsource.onopen = () => {
17      strictEqual(eventsource.readyState, EventSource.OPEN);
18      opened = true;
19    };
20    eventsource.onmessage = (event) => {
21      strictEqual(event.data, 'text/event-stream');
22      strictEqual(event.origin, 'http://example.org');
23      eventsource.close();
24      strictEqual(eventsource.readyState, EventSource.CLOSED);
25      resolve();
26    };
27    await promise;
28    ok(opened);
29  },
30};
31 
32export const cacheControlEventStreamTest = {
33  async test(ctrl, env) {
34    const eventsource = new EventSource(
35      'http://example.org/cache-control-event-stream',
36      { fetcher: env.subrequest }
37    );
38    const { promise, resolve } = Promise.withResolvers();
39    eventsource.onmessage = (event) => {
40      strictEqual(event.data, 'no-cache');
41      eventsource.close();
42      resolve();
43    };
44    await promise;
45  },
46};
47 
48export const lastEventIdTest = {
49  async test(ctrl, env) {
50    const eventsource = new EventSource('http://example.org/last-event-id', {
51      fetcher: env.subrequest,
52    });
53    const { promise, resolve } = Promise.withResolvers();
54    let first = true;
55    eventsource.onmessage = (event) => {
56      if (first) {
57        strictEqual(event.data, 'first');
58        first = false;
59      } else {
60        strictEqual(event.data, '1');
61        eventsource.close();
62        resolve();
63      }
64    };
65    await promise;
66  },
67};
68 
69export const eventIdPersistsTest = {
70  async test(ctrl, env) {
71    const eventsource = new EventSource(
72      'http://example.org/event-id-persists',
73      { fetcher: env.subrequest }
74    );
75    const { promise, resolve } = Promise.withResolvers();
76    eventsource.onmessage = (event) => {
77      switch (event.data) {
78        case 'first':
79          strictEqual(event.lastEventId, '1');
80          break;
81        case 'second':
82          strictEqual(event.lastEventId, '1');
83          break;
84        case 'third':
85          strictEqual(event.lastEventId, '2');
86          break;
87        case 'fourth':
88          strictEqual(event.lastEventId, '2');
89          eventsource.close();
90          resolve();
91          break;
92        default:
93          throw new Error(`Unexpected message: ${event.data}`);
94      }
95    };
96    await promise;
97  },
98};
99 
100export const eventIdResetsTest = {
101  async test(ctrl, env) {
102    const eventsource = new EventSource('http://example.org/event-id-resets', {
103      fetcher: env.subrequest,
104    });
105    const { promise, resolve } = Promise.withResolvers();
106    eventsource.onmessage = (event) => {
107      switch (event.data) {
108        case 'first':
109          strictEqual(event.lastEventId, '1');
110          break;
111        case 'second':
112          strictEqual(event.lastEventId, '');
113          eventsource.close();
114          resolve();
115          break;
116        default:
117          throw new Error(`Unexpected message: ${event.data}`);
118      }
119    };
120    await promise;
121  },
122};
123 
124export const eventIdResets2Test = {
125  async test(ctrl, env) {
126    const eventsource = new EventSource(
127      'http://example.org/event-id-resets-2',
128      { fetcher: env.subrequest }
129    );
130    const { promise, resolve } = Promise.withResolvers();
131    eventsource.onmessage = (event) => {
132      switch (event.data) {
133        case 'first':
134          strictEqual(event.lastEventId, '1');
135          break;
136        case 'second':
137          strictEqual(event.lastEventId, '');
138          eventsource.close();
139          resolve();
140          break;
141        default:
142          throw new Error(`Unexpected message: ${event.data}`);
143      }
144    };
145    await promise;
146  },
147};
148 
149export const messageTest = {
150  async test(ctrl, env) {
151    const eventsource = new EventSource('http://example.org/message', {
152      fetcher: env.subrequest,
153    });
154    const { promise, resolve } = Promise.withResolvers();
155    // We should get two messages...
156    let count = 0;
157    eventsource.onmessage = (event) => {
158      switch (count++) {
159        case 0: {
160          strictEqual(event.data, 'one\ntwo');
161          break;
162        }
163        case 1: {
164          strictEqual(event.data, 'end');
165          eventsource.close();
166          resolve();
167          break;
168        }
169      }
170    };
171    await promise;
172  },
173};
174 
175export const reconnectFailTest = {
176  async test(ctrl, env) {
177    const eventsource = new EventSource('http://example.org/reconnect-fail', {
178      fetcher: env.subrequest,
179    });
180    const { promise, resolve } = Promise.withResolvers();
181    let count = 0;
182    eventsource.onmessage = (event) => {
183      switch (count++) {
184        case 0: {
185          strictEqual(event.data, 'opened');
186          break;
187        }
188        case 1: {
189          strictEqual(event.data, 'reconnected');
190          break;
191        }
192      }
193    };
194    // Should be called four times.
195    let errorCount = 0;
196    eventsource.onerror = (event) => {
197      if (errorCount < 3) {
198        strictEqual(eventsource.readyState, EventSource.CONNECTING);
199      }
200      if (++errorCount === 4) {
201        strictEqual(eventsource.readyState, EventSource.CLOSED);
202        resolve();
203      }
204    };
205    await promise;
206  },
207};
208 
209export const statusErrorTest = {
210  async test(ctrl, env) {
211    const eventsource = new EventSource('http://example.org/status-error', {
212      fetcher: env.subrequest,
213    });
214    const { promise, resolve } = Promise.withResolvers();
215    eventsource.onopen = () => {
216      throw new Error('should not be called');
217    };
218    eventsource.onerror = (event) => {
219      strictEqual(eventsource.readyState, EventSource.CLOSED);
220      resolve();
221    };
222    await promise;
223  },
224};
225 
226export const eventTest = {
227  async test(ctrl, env) {
228    const eventsource = new EventSource('http://example.org/event', {
229      fetcher: env.subrequest,
230    });
231    const { promise, resolve } = Promise.withResolvers();
232    let count = 0;
233    eventsource.ontest = (event) => {
234      switch (count++) {
235        case 0: {
236          strictEqual(event.data, 'first');
237          break;
238        }
239        case 1: {
240          strictEqual(event.data, 'second');
241          eventsource.close();
242          resolve();
243          break;
244        }
245      }
246    };
247    await promise;
248  },
249};
250 
251export const retryTest = {
252  async test(ctrl, env) {
253    const eventsource = new EventSource('http://example.org/retry', {
254      fetcher: env.subrequest,
255    });
256    const { promise, resolve } = Promise.withResolvers();
257    let count = 0;
258    eventsource.onmessage = (event) => {
259      switch (count++) {
260        case 0: {
261          strictEqual(event.data, 'first');
262          break;
263        }
264        case 1: {
265          strictEqual(event.data, 'second');
266          break;
267        }
268        case 2: {
269          strictEqual(event.data, 'first');
270          break;
271        }
272        case 3: {
273          strictEqual(event.data, 'second');
274          eventsource.close();
275          resolve();
276          break;
277        }
278      }
279    };
280    await promise;
281  },
282};
283 
284export const constructorTest = {
285  test(ctrl, env) {
286    throws(() => new EventSource('not a valid url'), {
287      name: 'SyntaxError',
288      message:
289        "Cannot open an EventSource to 'not a valid url'. The URL is invalid.",
290    });
291 
292    throws(() => new EventSource(123), {
293      name: 'SyntaxError',
294      message: "Cannot open an EventSource to '123'. The URL is invalid.",
295    });
296 
297    throws(
298      () => new EventSource('http://example.org', { withCredentials: true }),
299      {
300        name: 'NotSupportedError',
301        message:
302          'The init.withCredentials option is not supported. It must be false or undefined.',
303      }
304    );
305 
306    // Doesn't throw
307    new EventSource('http://example.org/message', {
308      fetcher: env.subrequest,
309    }).close();
310    new EventSource('http://example.org/message', {
311      fetcher: env.subrequest,
312      withCredentials: false,
313    }).close();
314    new EventSource('http://example.org/message', {
315      fetcher: env.subrequest,
316      withCredentials: undefined,
317    }).close();
318  },
319};
320 
321export const eventSourceFromTest = {
322  async test() {
323    const enc = new TextEncoder();
324    const chunks = ['data: first\n\n', 'data: second\n\n', 'data: third\n\n'];
325    const rs = new ReadableStream({
326      async pull(c) {
327        await scheduler.wait(10);
328        c.enqueue(enc.encode(chunks.shift()));
329        if (chunks.length === 0) {
330          c.close();
331        }
332      },
333    });
334    const { promise, resolve } = Promise.withResolvers();
335    const eventsource = EventSource.from(rs);
336    // Should happen three times
337    let count = 0;
338    eventsource.onmessage = (event) => {
339      switch (count++) {
340        case 0: {
341          strictEqual(event.data, 'first');
342          break;
343        }
344        case 1: {
345          strictEqual(event.data, 'second');
346          break;
347        }
348        case 2: {
349          strictEqual(event.data, 'third');
350          eventsource.close();
351          resolve();
352          break;
353        }
354      }
355    };
356    await promise;
357  },
358};
359 
360export const eventSourceFromWithBOMTest = {
361  async test() {
362    const enc = new TextEncoder();
363    // The first chunk is going to include the UTF-8 BOM, which should
364    // be ignored and filtered out by the parser.
365    const chunks = [
366      '\uFEFFdata: first\n\n',
367      'data: second\n\n',
368      'data: third\n\n',
369    ];
370    const rs = new ReadableStream({
371      async pull(c) {
372        await scheduler.wait(10);
373        c.enqueue(enc.encode(chunks.shift()));
374        if (chunks.length === 0) {
375          c.close();
376        }
377      },
378    });
379    const { promise, resolve } = Promise.withResolvers();
380    const eventsource = EventSource.from(rs);
381    // Should happen three times
382    let count = 0;
383    eventsource.onmessage = (event) => {
384      switch (count++) {
385        case 0: {
386          strictEqual(event.data, 'first');
387          break;
388        }
389        case 1: {
390          strictEqual(event.data, 'second');
391          break;
392        }
393        case 2: {
394          strictEqual(event.data, 'third');
395          eventsource.close();
396          resolve();
397          break;
398        }
399      }
400    };
401    await promise;
402  },
403};
404 
405export const prototypePropertyTest = {
406  test() {
407    strictEqual(EventSource.prototype.constructor, EventSource);
408    strictEqual(EventSource.prototype.CLOSED, 2);
409    strictEqual(EventSource.prototype.CONNECTING, 0);
410    strictEqual(EventSource.prototype.OPEN, 1);
411    ok('onopen' in EventSource.prototype);
412    ok('onmessage' in EventSource.prototype);
413    ok('onerror' in EventSource.prototype);
414    ok('close' in EventSource.prototype);
415    ok('readyState' in EventSource.prototype);
416    ok('url' in EventSource.prototype);
417    ok('withCredentials' in EventSource.prototype);
418  },
419};
420 
421export const disposable = {
422  test() {
423    // EventSource is not defined by the spec as being disposable using ERM, but
424    // it makes sense to do so. The dispose operation simply defers to close()
425    const rs = new ReadableStream();
426    const eventsource = EventSource.from(rs);
427    strictEqual(eventsource.readyState, EventSource.OPEN);
428    eventsource[Symbol.dispose]();
429    strictEqual(eventsource.readyState, EventSource.CLOSED);
430  },
431};
432 
433// ======================================================================================
434 
435const handlers = {
436  '/accept-event-stream': acceptEventStream,
437  '/cache-control-event-stream': cacheControlEventStream,
438  '/last-event-id': lastEventId,
439  '/event-id-persists': eventIdPersists,
440  '/event-id-resets': eventIdResets,
441  '/event-id-resets-2': eventIdResets2,
442  '/message': message,
443  '/reconnect-fail': reconnectFail,
444  '/status-error': statusError,
445  '/event': event,
446  '/retry': retry,
447};
448 
449async function acceptEventStream(request) {
450  return new Response(`data: ${request.headers.get('accept')}\n\n`, {
451    headers: {
452      'Content-Type': 'text/event-stream',
453      'Cache-Control': 'no-cache',
454    },
455  });
456}
457 
458async function cacheControlEventStream(request) {
459  return new Response(`data: ${request.headers.get('cache-control')}\n\n`, {
460    headers: {
461      'Content-Type': 'text/event-stream',
462      'Cache-Control': 'no-cache',
463    },
464  });
465}
466 
467async function lastEventId(request) {
468  const lastEventId = request.headers.get('last-event-id');
469  if (lastEventId == null) {
470    return new Response('id: 1\ndata: first\n\n', {
471      headers: {
472        'Content-Type': 'text/event-stream',
473        'Cache-Control': 'no-cache',
474      },
475    });
476  } else {
477    return new Response(`data: ${lastEventId}\n\n`, {
478      headers: {
479        'Content-Type': 'text/event-stream',
480        'Cache-Control': 'no-cache',
481      },
482    });
483  }
484}
485 
486async function eventIdPersists(request) {
487  return new Response(
488    'id: 1\ndata: first\n\n' +
489      'data: second\n\n' +
490      'id: 2\ndata: third\n\n' +
491      'data: fourth\n\n',
492    {
493      headers: {
494        'Content-Type': 'text/event-stream',
495        'Cache-Control': 'no-cache',
496      },
497    }
498  );
499}
500 
501async function eventIdResets(request) {
502  return new Response('id: 1\ndata: first\n\n' + 'id: \ndata: second\n\n', {
503    headers: {
504      'Content-Type': 'text/event-stream',
505      'Cache-Control': 'no-cache',
506    },
507  });
508}
509 
510async function eventIdResets2(request) {
511  return new Response('id: 1\ndata: first\n\n' + 'id\ndata: second\n\n', {
512    headers: {
513      'Content-Type': 'text/event-stream',
514      'Cache-Control': 'no-cache',
515    },
516  });
517}
518 
519async function message(request) {
520  // The response payload contains a couple of messages with different structures.
521  // including good messages, comments, and bad fields.
522  return new Response(
523    'data: one\n' +
524      'data: two\n\n' +
525      ': comment' +
526      'falsefield:msg\n\n' +
527      'falsefield:msg\n' +
528      'Data: data\n\n' +
529      'data\n\n' +
530      'data:end\n\n',
531    {
532      headers: {
533        'Content-Type': 'text/event-stream',
534        'Cache-Control': 'no-cache',
535      },
536    }
537  );
538}
539 
540let reconnectTestCount = 0;
541async function reconnectFail(request) {
542  switch (reconnectTestCount++) {
543    case 0: {
544      return new Response('data: opened\n\n', {
545        headers: {
546          'Content-Type': 'text/event-stream',
547          'Cache-Control': 'no-cache',
548        },
549      });
550    }
551    case 1: {
552      return new Response('data: reconnected\n\n', {
553        headers: {
554          'Content-Type': 'text/event-stream',
555          'Cache-Control': 'no-cache',
556        },
557      });
558    }
559    // falls through
560    case 2:
561    // falls through
562    case 3: {
563      return new Response(null, {
564        headers: {
565          'Content-Type': 'text/event-stream',
566          'Cache-Control': 'no-cache',
567        },
568        status: 204,
569      });
570    }
571  }
572}
573 
574async function statusError(request) {
575  return new Response(null, {
576    status: 500,
577  });
578}
579 
580async function event(request) {
581  return new Response(
582    'event: test\n' + 'data: first\n\n' + 'event: test\n' + 'data: second\n\n',
583    {
584      headers: {
585        'Content-Type': 'text/event-stream',
586        'Cache-Control': 'no-cache',
587      },
588    }
589  );
590}
591 
592async function retry(request) {
593  return new Response(
594    'retry: 3000\n\n' + 'data: first\n\n' + 'data: second\n\n',
595    {
596      headers: {
597        'Content-Type': 'text/event-stream',
598        'Cache-Control': 'no-cache',
599      },
600    }
601  );
602}
603 
604export default {
605  async fetch(request) {
606    const url = new URL(request.url, 'http://example.org/');
607    const handler = handlers[url.pathname];
608    if (handler === undefined) {
609      throw new Error('Not found');
610    }
611    return await handler(request);
612  },
613};