lobe-chat

1
/**
2
 * file copy from https://github.com/Azure/fetch-event-source/blob/45ac3cfffd30b05b79fbf95c21e67d4ef59aa56a/src/fetch.ts
3
 * and remove some code
4
 */
5
import { EventSourceMessage, getBytes, getLines, getMessages } from './parse';
6

7
export const EventStreamContentType = 'text/event-stream';
8

9
const LastEventId = 'last-event-id';
10

11
// eslint-disable-next-line no-undef
12
export interface FetchEventSourceInit extends RequestInit {
13
  /** The Fetch function to use. Defaults to window.fetch */
14
  fetch?: typeof fetch;
15

16
  /**
17
   * The request headers. FetchEventSource only supports the Record<string,string> format.
18
   */
19
  headers?: Record<string, string>;
20

21
  /**
22
   * Called when a response finishes. If you don't expect the server to kill
23
   * the connection, you can throw an exception here and retry using onerror.
24
   */
25
  onclose?: () => void;
26

27
  /**
28
   * Called when there is any error making the request / processing messages /
29
   * handling callbacks etc. Use this to control the retry strategy: if the
30
   * error is fatal, rethrow the error inside the callback to stop the entire
31
   * operation. Otherwise, you can return an interval (in milliseconds) after
32
   * which the request will automatically retry (with the last-event-id).
33
   * If this callback is not specified, or it returns undefined, fetchEventSource
34
   * will treat every error as retriable and will try again after 1 second.
35
   */
36
  onerror?: (err: any) => number | null | undefined | void;
37

38
  /**
39
   * Called when a message is received. NOTE: Unlike the default browser
40
   * EventSource.onmessage, this callback is called for _all_ events,
41
   * even ones with a custom `event` field.
42
   */
43
  onmessage?: (ev: EventSourceMessage) => void;
44

45
  /**
46
   * Called when a response is received. Use this to validate that the response
47
   * actually matches what you expect (and throw if it doesn't.) If not provided,
48
   * will default to a basic validation to ensure the content-type is text/event-stream.
49
   */
50
  onopen: (response: Response) => Promise<void>;
51
}
52

53
export function fetchEventSource(
54
  // eslint-disable-next-line no-undef
55
  input: RequestInfo,
56
  {
57
    signal: inputSignal,
58
    headers: inputHeaders,
59
    onopen: inputOnOpen,
60
    onmessage,
61
    onclose,
62
    onerror,
63
    fetch: inputFetch,
64
    ...rest
65
  }: FetchEventSourceInit,
66
) {
67
  return new Promise<void>((resolve) => {
68
    // make a copy of the input headers since we may modify it below:
69
    const headers = { ...inputHeaders };
70
    if (!headers.accept) {
71
      headers.accept = EventStreamContentType;
72
    }
73

74
    const fetch = inputFetch ?? window.fetch;
75
    async function create() {
76
      try {
77
        const response = await fetch(input, {
78
          ...rest,
79
          headers,
80
          signal: inputSignal,
81
        });
82

83
        await inputOnOpen(response);
84

85
        await getBytes(
86
          response.body!,
87
          getLines(
88
            getMessages((id) => {
89
              if (id) {
90
                // store the id and send it back on the next retry:
91
                headers[LastEventId] = id;
92
              } else {
93
                // don't send the last-event-id header anymore:
94
                delete headers[LastEventId];
95
              }
96
            }, onmessage),
97
          ),
98
        );
99

100
        onclose?.();
101
        resolve();
102
      } catch (err) {
103
        onerror?.(err);
104
        resolve();
105
      }
106
    }
107

108
    create();
109
  });
110
}
111