前言

終于遇到一個(gè)簡(jiǎn)單的庫(kù)來(lái)學(xué)習(xí)它的源碼了。這個(gè)項(xiàng)目只有2個(gè)主要文件，代碼加起來(lái)不到500行，是真的很mini了。

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

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

目錄結(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'其實(shí)是import + export的縮寫(xiě)。

上面的代碼其實(shí)就是下面代碼的縮寫(xiě)：

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

發(fā)起請(qǐng)求

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

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

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

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

onerror連接發(fā)送錯(cuò)誤時(shí)的回調(diào)函數(shù)，如果沒(méi)有指定這個(gè)回調(diào)或返回undefined就會(huì)發(fā)起重新連接請(qǐng)求。

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

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

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

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

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

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

再說(shuō)一下主動(dòng)打斷連接這里，fetchEventSource的第2個(gè)參數(shù)可以傳入一個(gè)信號(hào)signal，這個(gè)屬性在FetchEventSourceInit中沒(méi)有定義。借助AbortController中斷連接，具體信息可以看AbortController-MDN。

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

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

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

解析消息

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

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

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

處理過(guò)程分析

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)建一個(gè)讀取器，用while循環(huán)讀取流數(shù)據(jù)，每讀取一段就執(zhí)行onChunk解析流數(shù)據(jù)，onChunk就是在fetch.ts中getLines的返回值。

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

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

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

問(wèn)題

在調(diào)試這個(gè)庫(kù)的時(shí)候，在html中引入打包后的esm文件會(huì)報(bào)錯(cuò)文件找不到，因?yàn)槲募麤](méi)有添加后綴。

接口返回值的類(lèi)型必須是text/event-stream類(lèi)型的，就算是流數(shù)據(jù)也不行。

總結(jié)

getBytes的第2個(gè)參數(shù)是getLines的返回值，getLines的參數(shù)又是getMessages的返回值，嵌套的比較深。
onChunk將字節(jié)塊切割成一行一行的字節(jié)，涉及字節(jié)數(shù)據(jù)的知識(shí)。

以上就是fetch-event-source庫(kù)源碼學(xué)習(xí)的詳細(xì)內(nèi)容，更多關(guān)于fetch-event-source的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論