快捷導(dǎo)航

fetch-event-source庫使用源碼學(xué)習(xí)

更新時間：2023年09月21日 08:45:25 作者：河豚學(xué)前端

這篇文章主要為大家介紹了fetch-event-source庫源碼學(xué)習(xí),有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

前言

終于遇到一個簡單的庫來學(xué)習(xí)它的源碼了。這個項目只有2個主要文件，代碼加起來不到500行，是真的很mini了。

客戶端向服務(wù)端發(fā)起請求用xhr或fetch，客戶端與服務(wù)端雙向通信用websocket，而服務(wù)端主動發(fā)起請求用sse。chatGPT就是用sse回復(fù)提問的。

window中有一個叫EventSource的構(gòu)造函數(shù)。一個EventSource實例會對服務(wù)器開啟一個持久化的連接，以text/event-stream格式發(fā)送事件，此連接會一直保持開啟直到通過調(diào)用EventSource.close()關(guān)閉。但使用EventSource時只能把參數(shù)加到url后面，而且也不能像fetch請求那樣設(shè)置header等參數(shù)。借助fetch-event-source這個庫就可以像發(fā)起fetch請求一樣發(fā)起服務(wù)器單向通信請求。

目錄結(jié)構(gòu)

入口文件

index.ts是入口文件，里面只有2行代碼，導(dǎo)出了fetch.ts和parse.ts中部分變量和方法。

export { fetchEventSource, FetchEventSourceInit, EventStreamContentType } from './fetch';
export { EventSourceMessage } from './parse';

export {...} from 'xx'其實是import + export的縮寫。

上面的代碼其實就是下面代碼的縮寫：

import { fetchEventSource, FetchEventSourceInit, EventStreamContentType } from './fetch';
import { EventSourceMessage } from './parse';
export {
    fetchEventSource,
    FetchEventSourceInit,
    EventStreamContentType,
    EventSourceMessage,
}

發(fā)起請求

首先定義了變量EventStreamContentType，它的值是sse的MIME Type。它在2個地方使用。第一處是發(fā)起請求時設(shè)置headers.accept，告訴服務(wù)器只接受text/event-stream格式的數(shù)據(jù)。第2處是在連接建立時判斷response.headers.get('content-type')是否等于EventStreamContentType的值，如果不是的話就拋出一個錯誤，聲明期待的類型是text/event-stream。
接下來定義了變量DefaultRetryInterval，sse有自動重連機(jī)制，這里定義了每次重連的默認(rèn)間隔為1s。然后定義了變量LastEventId，表示上一次事件的id，添加在headers中發(fā)送到服務(wù)端。

接下來定義了一個類型FetchEventSourceInit，它聲明了fetchEventSource的第2個參數(shù)的類型。參數(shù)一共有7個。
headers 請求頭。

onopen 連接建立時的回調(diào)函數(shù)，如果沒有設(shè)置會調(diào)用默認(rèn)的defaultOnOpen，這個默認(rèn)回調(diào)里進(jìn)行了返回值類型判斷。
onmessage每次收到消息時的回調(diào)函數(shù)，參數(shù)是消息對象，它的類型就是parse.ts中定義的EventSourceMessage。

onclose 連接關(guān)閉時的回調(diào)函數(shù)。

onerror連接發(fā)送錯誤時的回調(diào)函數(shù)，如果沒有指定這個回調(diào)或返回undefined就會發(fā)起重新連接請求。

openWhenHidden 默認(rèn)為false，監(jiān)聽visibilitychange，當(dāng)頁面不可見時關(guān)閉連接，當(dāng)頁面重新可見時重新打開連接。

fetch發(fā)起請求的方法，默認(rèn)為window.fetch。

Record<string, string>等價于{[key: string]: string}Promise<void>定義了一個異步函數(shù)，返回值是voidtypeof fetch 獲取fetch的類型，typeof后面跟的是變量，表示類型定義

接下來就是最重要的fetchEventSource，它是一個異步函數(shù)，接受2個參數(shù)：url和FetchEventSourceInit類型的對象。

在這個方法中，首先定義了接受的媒體類型。然后添加監(jiān)聽visibilitychange事件，然后添加監(jiān)聽abort事件供使用者可以手動打斷連接，然后發(fā)起連接，拿到返回值后將返回值傳遞給onopen，然后調(diào)用getBytes解析返回值，解析之后關(guān)閉連接。

用try...catch包裹發(fā)起連接和解析返回值以及關(guān)閉連接的過程，如果捕獲到錯誤且不是主動打斷的就發(fā)起重連。

再說一下主動打斷連接這里，fetchEventSource的第2個參數(shù)可以傳入一個信號signal，這個屬性在FetchEventSourceInit中沒有定義。借助AbortController中斷連接，具體信息可以看AbortController-MDN。

??類似||，相同點(diǎn)在于根據(jù)前面的值判斷返回前面的還是后面的，不同點(diǎn)在于??的第一個值為null或undefined時返回第二個值，||會將第一個值先轉(zhuǎn)換為布爾值。比如

0 ?? 2 // 0
0 || 2 // 2

defaultOnOpen定義默認(rèn)onopen回調(diào)，主要是檢查返回值類型。

解析消息

首先使用response.body來獲取響應(yīng)的主體內(nèi)容，并通過getBytes函數(shù)將其轉(zhuǎn)換為字節(jié)數(shù)組。然后，使用getLines函數(shù)將字節(jié)數(shù)組拆分成行，并使用getMessages函數(shù)將每行解析為事件消息。

處理ReadableStream數(shù)據(jù)

// 創(chuàng)建了一個數(shù)據(jù)讀取器
const reader = response.getReader();
// 創(chuàng)建了一個文本解碼器
const decoder = new TextDecoder();
reader.read().then(function processText({ done, value }) {
  // Result 對象包含了兩個屬性：
  // done  - 當(dāng) stream 傳完所有數(shù)據(jù)時則變成 true
  // value - 數(shù)據(jù)片段。當(dāng) done 為 true 時始終為 undefined
  if (done) {
 return;
  }
  // 將字節(jié)流轉(zhuǎn)換為字符
  const text = decoder.decode(value)
  // 內(nèi)容
  console.log(text);
  // 再次調(diào)用這個函數(shù)以讀取更多數(shù)據(jù)
  return reader.read().then(processText);
});

處理過程分析

await getBytes(response.body!, getLines(getMessages(id => {
 if (id) {
     // store the id and send it back on the next retry:
     headers[LastEventId] = id;
 } else {
     // don't send the last-event-id header anymore:
     delete headers[LastEventId];
 }
  }, retry => {
 retryInterval = retry;
}, onmessage)));

首先執(zhí)行的是getBytes方法，它創(chuàng)建一個讀取器，用while循環(huán)讀取流數(shù)據(jù)，每讀取一段就執(zhí)行onChunk解析流數(shù)據(jù)，onChunk就是在fetch.ts中getLines的返回值。

onChunk將字節(jié)塊按行分割，并將每行的字節(jié)子數(shù)組和字段長度傳遞給onLine回調(diào)函數(shù)。onLine則是getMessages的返回值。
getMessages創(chuàng)建了一個解碼器，返回一個名為onLine的函數(shù)，用于處理每個傳入的行數(shù)據(jù)。它將行的字節(jié)子數(shù)組解碼為字符串，并根據(jù)字段的類型進(jìn)行相應(yīng)的處理。比如，如果字段是data，它會將值追加到message.data中，如果message.data已經(jīng)存在，則在原有值的基礎(chǔ)上添加新值，并使用換行符分隔。

將字節(jié)流先按行分割再解析是為了更好的處理數(shù)據(jù)，因為數(shù)據(jù)都是field:value格式的。

TextDecoder表示一個文本解碼器，可以將字節(jié)流數(shù)據(jù)轉(zhuǎn)換成指定碼位流，默認(rèn)是utf-8。