Add server-sent event (for client-side) supports

[kingcean]

Purpose

Extends following fromFetch API to add supports for server-sent event (a.k.a SSE).

import { fromFetch } from 'rxjs/fetch'

Usages

Currently, the API defines following arguments. It contains 2 arguments as same as the ones of fetch API.

export function fromFetch(input: string | Request, init?: RequestInit): Observable<Response>;

The current API also has an overload function to extends the second argument which contains a new property selector.

export function fromFetch<T>(
  input: string | Request,
  init: RequestInit & {
    selector: (response: Response) => ObservableInput<T>;
  }
): Observable<T>;

The new property is a handler to get and convert result, e.g. following example (copied from the sample in source code).

 const data = fromFetch('https://api.github.com/users?per_page=5', {
   selector: response => response.json() // new property here
 });

My opinion to add SSE supports is to extend the current API. Specifically, we add more overloads of this function to support further scenarios.

export function fromFetch(
  input: string | Request,
  init: RequestInit & {
    selector: 'sse' | 'server-sent-event';
  }
): Observable<ServerSentEventItem>;

export class ServerSentEventItem {
  private source: Record<string, string>;
  private dataParsedInJson: Record<string, unknown> | undefined = undefined;
  constructor(source: Record<string, string>) {
    this.source = source || {};
  }
  get event() {
    return this.source.event || "message";
  }
  get data() {
    return this.source.data;
  }
  get id() {
    return this.source.id;
  }
  get comment() {
    return this.source[''];
  }
  get retry() {
    return this.source.retry ? parseInt(this.source.retry, 10) : undefined;
  }
  dataJson<T = Record<string, unknown>>() {
    const data = this.source.data;
    if (!data) return undefined;
    if (!this.dataParsedInJson) this.dataParsedInJson = JSON.parse(data);
    return data as T;
  }
  get(key: string) {
    return this.source[key];
  }
}

So it will get data by SSE mode if the value of property selector is string sse. The Observable object returned may occur next for multiple times that depends on the SSE item messages received during networking streaming.

 const data = fromFetch(UrlWithStreamMode, {
   selector: 'sse'
 });

Further more, the property selector can be other string value for simplifying common scenarios.

String value of selector	Equivalent
json	response => response.json()
text	response => response.text()
form-data	response => response.formData()
blob	response => response.blob()
array-buffer	response => response.arrayBuffer()

export function fromFetch<T = Record<string, unknown>>(
  input: string | Request,
  init: RequestInit & {
    selector: 'json';
  }
): Observable<T>;
export function fromFetch(
  input: string | Request,
  init: RequestInit & {
    selector: 'text';
  }
): Observable<string>;
// Others are omitted here

Implementation

In ~\packages\rxjs\src\internal\observable\dom\fetch.ts file.

export function fromFetch<T>(
  input: string | Request,
  initWithSelector: RequestInit & {
    selector?: (response: Response) => ObservableInput<T>;
  } = {}
): Observable<Response | T> {
+  const { selector: originalSelector, ...init } = initWithSelector;
+  let selector = originalSelector;
-  const { selector, ...init } = initWithSelector;
  return new Observable<Response | T>((destination) => {
    // … some code is omitted here
    fetch(input, perSubscriberInit)
      .then((response) => {
        if (selector) {
+        if (typeof selector === 'string') {
+          // add switch-case code here to extend
+          // the implementation will introduce later
+        }
          from(selector(response)).subscribe(
            operate({
              destination,
              complete: () => {
                abortable = false;
                destination.complete();
              },
              error: handleError,
            })
          );
        } else {
          abortable = false;
          destination.next(response);
          destination.complete();
        }
      })
      .catch(handleError);
    // … some code is omitted here
  });
}

The inserted code above is following. It switches the supported selector to the corresponding cases.

if (typeof selector === 'string') {
  switch (selector.toLowerCase()) {
    case 'json':
      selector = response => response.json();
      break;
    case 'text':
      selector = response => response.text();
      break;
    case 'formdata':
    case 'form-data':
      selector = response => response.formData();
      break;
    case 'blob':
      selector = response => response.blob();
      break;
    case 'arraybuffer':
    case 'array-buffer':
      selector = response => response.arrayBuffer();
      break;
    case 'sse':
    case 'server-sent-event':
      const decoder = new TextDecoder('utf-8');
      const reader = response.body!.getReader();
      handleSse(destination, reader, decoder, () => {
        abortable = false;
      });
      return;
    default:
      abortable = false;
      destination.error('selector does not support');
      return;
  }
}

The implementation calls fetch and response.body.getReader() to parse SSE instead of EventSource, because the latter only support GET method and can't set headers.

async function handleSse(destination: Subscriber<ServerSentEventItem>, reader: ReadableStreamDefaultReader<Uint8Array<ArrayBuffer>>, decoder: TextDecoder, callback: () => void) {
  let buffer = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) {
      callback();
      destination.complete();
      break;
    }
    buffer += decoder.decode(value, { stream: true });
    const messages = buffer.split('\n\n');
    if (messages.length < 2) continue;
    buffer = messages.pop() || '';
    messages.forEach(msg => {
      if (!msg) return;
      const item: Record<string, string> = {};
      msg.split('\n').forEach(line => {
        const pos = line.indexOf(':');
        if (pos < 0) return;
        const key = line.substring(0, pos);
        const value = line.substring(pos + 1);
        if (!item[key] || (key !== 'data' && key !== '')) item[key] = value;
        else item[key] += value;
      });
      destination.next(new ServerSentEventItem(item));
    });
  }
}