sse.ts
3.6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import type { AxiosRequestHeaders, InternalAxiosRequestConfig } from 'axios';
import type { RequestClient } from '../request-client';
import type { SseRequestOptions } from '../types';
/**
* SSE模块
*/
class SSE {
private client: RequestClient;
constructor(client: RequestClient) {
this.client = client;
}
public async postSSE(
url: string,
data?: any,
requestOptions?: SseRequestOptions,
) {
return this.requestSSE(url, data, {
...requestOptions,
method: 'POST',
});
}
/**
* SSE请求方法
* @param url - 请求URL
* @param data - 请求数据
* @param requestOptions - SSE请求选项
*/
public async requestSSE(
url: string,
data?: any,
requestOptions?: SseRequestOptions,
) {
const baseUrl = this.client.getBaseUrl() || '';
let axiosConfig: InternalAxiosRequestConfig<any> = {
url,
method: (requestOptions?.method as any) ?? 'GET',
headers: {} as AxiosRequestHeaders,
};
const requestInterceptors = this.client.instance.interceptors
.request as any;
if (
requestInterceptors.handlers &&
requestInterceptors.handlers.length > 0
) {
for (const handler of requestInterceptors.handlers) {
if (typeof handler?.fulfilled === 'function') {
const next = await handler.fulfilled(axiosConfig as any);
if (next) axiosConfig = next as InternalAxiosRequestConfig<any>;
}
}
}
const merged = new Headers();
Object.entries(
(axiosConfig.headers ?? {}) as Record<string, string>,
).forEach(([k, v]) => merged.set(k, String(v)));
if (requestOptions?.headers) {
new Headers(requestOptions.headers).forEach((v, k) => merged.set(k, v));
}
if (!merged.has('accept')) {
merged.set('accept', 'text/event-stream');
}
let bodyInit = requestOptions?.body ?? data;
const ct = (merged.get('content-type') || '').toLowerCase();
if (
bodyInit &&
typeof bodyInit === 'object' &&
!ArrayBuffer.isView(bodyInit as any) &&
!(bodyInit instanceof ArrayBuffer) &&
!(bodyInit instanceof Blob) &&
!(bodyInit instanceof FormData) &&
ct.includes('application/json')
) {
bodyInit = JSON.stringify(bodyInit);
}
const requestInit: RequestInit = {
...requestOptions,
method: axiosConfig.method,
headers: merged,
body: bodyInit,
};
const response = await fetch(safeJoinUrl(baseUrl, url), requestInit);
if (!response.ok) {
throw new Error(`HTTP error! status: ${response.status}`);
}
const reader = response.body?.getReader();
const decoder = new TextDecoder();
if (!reader) {
throw new Error('No reader');
}
let isEnd = false;
while (!isEnd) {
const { done, value } = await reader.read();
if (done) {
isEnd = true;
decoder.decode(new Uint8Array(0), { stream: false });
requestOptions?.onEnd?.();
reader.releaseLock?.();
break;
}
const content = decoder.decode(value, { stream: true });
requestOptions?.onMessage?.(content);
}
}
}
function safeJoinUrl(baseUrl: string | undefined, url: string): string {
if (!baseUrl) {
return url; // 没有 baseUrl，直接返回 url
}
// 如果 url 本身就是绝对地址，直接返回
if (/^https?:\/\//i.test(url)) {
return url;
}
// 如果 baseUrl 是完整 URL，就用 new URL
if (/^https?:\/\//i.test(baseUrl)) {
return new URL(url, baseUrl).toString();
}
// 否则，当作路径拼接
return `${baseUrl.replace(/\/+$/, '')}/${url.replace(/^\/+/, '')}`;
}
export { SSE };