出典: JD Cloud Developers

2.1 主な特徴

• 一方向通信：SSEはサーバーからクライアントへの一方向通信用に設計されています。クライアントはSSEを介してサーバーに直接データを送信することはできませんが、AJAXなどの他の方法を介してサーバーとやり取りすることができます。
• HTTPベース：SSEはHTTPプロトコル上で動作し、新しいプロトコルやポートを必要としません。そのため、既存のWebアプリケーションアーキテクチャで簡単に使用でき、標準的なHTTPプロキシとミドルウェアを通じてサポートされます。
• 自動再接続: 接続が失われた場合、ブラウザは自動的にサーバーへの再接続を試みます。
• シンプルな形式: SSE はシンプルなテキスト形式を使用してメッセージを送信します。各メッセージは 2 つの連続する改行文字で区切られます。
• ネイティブ ブラウザ サポート: 多くの最新ブラウザ (Chrome、Firefox、Safari など) は SSE をネイティブにサポートしていますが、Internet Explorer や以前のバージョンの Edge など、一部のブラウザは SSE をサポートしていないことに注意してください。

2.2 SSEとWebSocket

3.1 SSEプロトコルのメカニズム

• Content-Type: text/event-stream: 返されるコンテンツがイベント ストリームであることを示します。
• Cache-Control: no-cache: これにより、サーバーによってプッシュされたメッセージがキャッシュされなくなり、メッセージのリアルタイム性が保証されます。
• 接続: キープアライブ: サーバーがいつでもメッセージを送信できるように、接続を常に開いたままにしておくように指示します。

3.2 メッセージの形式と構造

• `event`: イベントの種類を定義します。これは、クライアントが受信したメッセージの処理方法を決定するのに役立ちます。
• id: イベントの一意の識別子を提供します。接続が中断された場合、クライアントは最後に受信したイベントIDを使用して、サーバーに特定の時点からのメッセージの再送信を要求できます。
• `retry` は、接続が切断された際にクライアントが再接続を試みるまでの待機時間をミリ秒単位で指定します。これにより、接続の中断と再接続のためのメカニズムが提供されます。
• データ: これはメッセージのメインコンテンツです。UTF-8でエンコードされた任意のテキストで、複数行にまたがる場合があります。各データ行はクライアント側での解析時に連結され、改行文字で区切られます。

 id: 123event: updatedata: {"message": "これはテストメッセージです"}

4.1 接続を確立する方法

 const options = { withCredentials : true // 允许跨域请求携带凭证};
 // 创建一个EventSource 对象以开始监听const eventSource = new EventSource( 'your_server_url' , options); 

4.2 受信したイベントの処理方法

• 一般的なイベント処理:

 // 监听连接打开事件eventSource.onopen = function ( event ) { console .log( 'Connection to SSE server established!' ); };
 // 监听标准消息事件eventSource.onmessage = function ( event ) { console .log( 'Received data from server: ' , event.data); };
 // 监听错误事件eventSource.onerror = function ( event ) { console .error( 'An error occurred while receiving data:' , event); }; 

• カスタムイベント処理:

 // 「update」という名前のカスタム イベントをリッスンします。eventSource.addEventListener('update', function(event) { console.log('Received update event:', event.data);});

4.3 接続を閉じる

イベントソースを閉じます。

5.1 サーバー側

 const http = require ( 'http' ); const fs = require ( 'fs' );
 // 初始化HTTP 服务器http.createServer( ( req, res ) => {
 // 为了简洁，将响应方法抽离成函数function serveFile ( filePath, contentType ) {    fs.readFile(filePath, ( err, data ) => { if (err) {        res.writeHead( 500 );        res.end( 'Error loading the file' );      } else {        res.writeHead( 200 , { 'Content-Type' : contentType});        res.end(data);      }    });  }
 function handleSSEConnection ( ) {    res.writeHead( 200 , { 'Content-Type' : 'text/event-stream' , 'Cache-Control' : 'no-cache' , 'Connection' : 'keep-alive'    });
 let id = 0 ; const intervalId = setInterval( () => { const message = {        event: 'customEvent' ,        id: id++,        retry: 30000 ,        data: { id, time: new Date ().toISOString() }      }; for ( let key in message) { if (key !== 'data' ) {          res.write( ` ${key} : ${message[key]} \n` );        } else {          res.write( `data: ${JSON.stringify(message.data)} \n\n` );        }      }    }, 1000 );
    req.on( 'close' , () => {      clearInterval(intervalId);      res.end();    });  }
 switch (req.url) { case '/' :      serveFile( 'index.html' , 'text/html' ); break ; case '/events' :      handleSSEConnection(); break ; default :      res.writeHead( 404 );      res.end(); break ;  }
 }).listen( 3000 );
 console .log( 'Server listening on port 3000' ); 

5.2 クライアント

 < html lang = "en" > < head > < meta charset = "UTF-8" > < meta http-equiv = "X-UA-Compatible" content = "IE=edge" > < meta name = "viewport" content = "width=device-width, initial-scale=1.0" > < title > SSE Demo title > head > < body > < h1 > SSE Demo h1 > < button onclick = "connectSSE()" >建立SSE 连接button > < button onclick = "closeSSE()" >断开SSE 连接button > < br /> < br /> < div id = "message" > div >
 < script > const messageElement = document .getElementById( 'message' ); let eventSource;
 // 连接SSE function connectSSE ( ) {      eventSource = new EventSource( '/events' );
      eventSource.addEventListener( 'customEvent' , handleReceivedMessage);      eventSource.onopen = handleConnectionOpen;      eventSource.onerror = handleConnectionError;    }
 // 断开SSE 连接function closeSSE ( ) {      eventSource.close();      appendMessage( `SSE 连接关闭，状态${eventSource.readyState} ` );    }
 // 处理从服务端收到的消息function handleReceivedMessage ( event ) { const data = JSON .parse(event.data);      appendMessage( ` ${data.id} --- ${data.time} ` );    }
 // 连接建立成功的处理函数function handleConnectionOpen ( ) {      appendMessage( `SSE 连接成功，状态${eventSource.readyState} ` );    }
 // 连接发生错误的处理函数function handleConnectionError ( ) {      appendMessage( `SSE 连接错误，状态${eventSource.readyState} ` );    }
 // 将消息添加到页面上function appendMessage ( message ) {      messageElement.innerHTML += ` ${message}
 ` ;    } script > body > html > 

6.1 既存の問題

• デフォルトのリクエストはGETメソッドのみをサポートしています。フロントエンドからバックエンドにパラメータを渡す必要がある場合、パラメータはリクエストURLに追加することしかできず、複雑なビジネスシナリオでは非常に扱いにくくなります。
• サーバーによって返されるデータの形式には固定の要件があり、イベント、ID、再試行、およびデータの構造に従って返される必要があります。
• サーバーから送信されたデータはブラウザ コンソールで表示できるため、機密データが公開され、データ セキュリティの問題が発生する可能性があります。

6.2 最適化手法

 /** * Utf8ArrayToStr: 将Uint8Array的数据转为字符串 * @param {Uint8Array} array - Uint8Array数据 * @return {string} - 转换后的字符串 */ function Utf8ArrayToStr ( array ) { const decoder = new TextDecoder(); return decoder.decode(array); }
 /** * fetchStream: 建立一个SSE连接，并支持多种HTTP请求方式 * @param {string} url - 请求的URL地址 * @param {object} params - 请求的参数，包括HTTP方法、头部、主体内容等 * @return {Promise} - 返回一个Promise对象 */ const fetchStream = ( url, params ) => { const { onmessage, onclose, ...otherParams } = params;
 return fetch(url, otherParams)        .then( response => { let reader = response.body?.getReader();
 return new ReadableStream({                start(controller) { function push ( ) {                        reader?.read().then( ( { done, value } ) => { if (done) {                                controller.close();                                onclose?.(); return ;                            } const decodedData = Utf8ArrayToStr(value); console .log(decodedData);
                            onmessage?.(decodedData);
                            controller.enqueue(value);
                            push();                        });                    }                    push();                }            });        })        .then( stream => { return new Response(stream, { headers : { "Content-Type" : "text/html" }            }).text();        }); };
 // 示例：调用fetchStream函数fetchStream( "/events" , { method : "POST" , // 使用POST方法    headers: { "content-type" : "application/json"    }, credentials : "include" , body : JSON .stringify({ // 这里列出了一些示例数据，实际业务场景请替换为你的数据        boxId: "exampleBoxId" , sessionId : "exampleSessionId" , queryContent : "exampleQueryContent"    }), onmessage : res => { console .log(res); // 当接收到消息时的回调    }, onclose : () => { console .log( "Connection closed." ); // 当连接关闭时的回调    } });
 

6.3 プラグインのパッケージ化

 // 定义请求主体的接口，需要根据具体的应用场景定义具体的属性interface RequestBody { // 示例属性，具体属性需要根据实际需求定义    key?: string ; }
 // 错误响应的结构interface ErrorResponse {    error: string ;    detail: string ; }
 // 返回值类型定义type TextStream = ReadableStreamDefaultReader< Uint8Array >;
 // 获取数据并返回TextStream async function fetchData (    url: string ,    body: RequestBody,    accessToken: string ,    onError: (message: string ) => void ): Promise < TextStream | undefined > { try { // 尝试发起请求const response = await fetch(url, {            method: "POST" ,            cache: "no-cache" ,            keepalive: true ,            headers: { "Content-Type" : "application/json" ,                Accept: "text/event-stream" ,                Authorization: `Bearer ${accessToken} ` ,            },            body: JSON .stringify(body),        });
 // 检查是否有冲突，例如重复请求if (response.status === 409 ) { const error: ErrorResponse = await response.json();            onError(error.detail); return undefined ;        }
 return response.body?.getReader();    } catch (error) {        onError( `Failed to fetch: ${error.message} ` ); return undefined ;    } }
 // 读取流数据async function readStream ( reader: TextStream ): Promise < string | null > { const result = await reader.read(); return result.done ? null : new TextDecoder().decode(result.value); }
 // async function processStream ( // 处理文本流数据( async function processStream (    reader: TextStream,    onStart: () => void ,    onText: (text: string ) => void ,    onError: (error: string ) => void ,    shouldClose: () => boolean ): Promise < void > { try { // 开始处理数据        onStart();         while ( true ) { if (shouldClose()) { await reader.cancel(); return ;            } const text = await readStream(reader); if (text === null ) break ;
            onText(text);        }    } catch (error) {        onError( `Processing stream failed: ${error.message} ` );    } }
 /** * 主要的导出函数，用于处理流式文本数据。 * * @param url 请求的URL。 * @param body 请求主体内容。 * @param accessToken 访问令牌。 * @param onStart 开始处理数据时的回调。 * @param onText 接收到数据时的回调。 * @param onError 错误处理回调。 * @param shouldClose 判断是否需要关闭流的函数。 */ export async function streamText (    url: string ,    body: RequestBody,    accessToken: string ,    onStart: () => void ,    onText: (text: string ) => void ,    onError: (error: string ) => void ,    shouldClose: () => boolean ): Promise < void > { const reader = await fetchData(url, body, accessToken, onError);     if (!reader) { console .error( "Reader is undefined!" ); return ;    }
 await processStream(reader, onStart, onText, onError, shouldClose); } 

-終わり-