zoujing/YGChatCS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: zoujing:f1249e2f72986c7b6b695b84bce1e81dd619fdb6
Branches Tags
zoujing:main
zoujing:fix-812
..
compare: zoujing:b2f7f67e3929790a1ce520c46674562471f97f48
Branches Tags
zoujing:main
zoujing:fix-812

No commits in common. "f1249e2f72986c7b6b695b84bce1e81dd619fdb6" and "b2f7f67e3929790a1ce520c46674562471f97f48" have entirely different histories.
f1249e2f72 ... b2f7f67e39

1 changed files with 166 additions and 199 deletions
Show Stats Expand all files Collapse all files

359
request/api/AgentChatStream.js
View File

@ -12,187 +12,168 @@ const API = '/agent/assistant/chat';
let requestTask = null; let requestTask = null;
let isAborted = false; // 添加终止状态标志 let isAborted = false; // 添加终止状态标志
let currentPromiseReject = null; // 保存当前Promise的reject函数 let currentPromiseReject = null; // 保存当前Promise的reject函数
let currentMessageId = null; // 当前请求的messageId let requestId = 0; // 请求ID，用于区分不同的请求
/**
* 终止的请求
*/
const stopAbortTask = () => { const stopAbortTask = () => {
console.log("🛑 开始强制终止请求... "); console.log('🛑 开始强制终止请求...');
isAborted = true; // 立即设置终止标志 isAborted = true; // 立即设置终止标志
// 立即拒绝当前Promise（最强制的终止） // 立即拒绝当前Promise（最强制的终止）
if (currentPromiseReject) { if (currentPromiseReject) {
currentPromiseReject(new Error('请求已被用户终止')); console.log('🛑 立即拒绝Promise');
currentPromiseReject = null; currentPromiseReject(new Error('请求已被用户终止'));
} currentPromiseReject = null;
}
if (requestTask) { if (requestTask) {
// 先取消所有监听器 // 先取消所有监听器（关键：必须在abort之前）
try { try {
if (requestTask.offChunkReceived) { if (requestTask.offChunkReceived) {
requestTask.offChunkReceived(); requestTask.offChunkReceived();
} console.log('🛑 已取消 ChunkReceived 监听');
} catch (e) { }
console.log('🛑 取消 ChunkReceived 监听失败:', e); } catch (e) {
} console.log('🛑 取消 ChunkReceived 监听失败:', e);
}
try { try {
if (requestTask.offHeadersReceived) { if (requestTask.offHeadersReceived) {
requestTask.offHeadersReceived(); requestTask.offHeadersReceived();
} console.log('🛑 已取消 HeadersReceived 监听');
} catch (e) { }
console.log('🛑 取消 HeadersReceived 监听失败:', e); } catch (e) {
} console.log('🛑 取消 HeadersReceived 监听失败:', e);
}
// 然后终止网络请求 // 然后终止网络请求
try { try {
if (requestTask.abort) { if (requestTask.abort) {
requestTask.abort(); requestTask.abort();
} console.log('🛑 已终止网络请求');
} catch (e) { }
console.log('🛑 终止网络请求失败:', e); } catch (e) {
} console.log('🛑 终止网络请求失败:', e);
}
requestTask = null; requestTask = null;
} }
currentMessageId = null; // 递增请求ID，使旧请求的数据无效
console.log('🛑 请求强制终止完成'); requestId++;
console.log('🛑 请求强制终止完成，新请求ID:', requestId);
} }
const agentChatStream = (params, onChunk) => { const agentChatStream = (params, onChunk) => {
const promise = new Promise((resolve, reject) => { const promise = new Promise((resolve, reject) => {
const token = uni.getStorageSync('token'); const token = uni.getStorageSync('token');
let hasError = false; let hasError = false;
isAborted = false; // 重置终止状态 isAborted = false; // 重置终止状态
// 保存当前Promise的reject函数，用于强制终止 // 保存当前Promise的reject函数，用于强制终止
currentPromiseReject = reject; currentPromiseReject = reject;
// 设置当前请求的messageId，用于区分不同请求 // 为当前请求分配ID
const messageId = params.messageId; const currentRequestId = ++requestId;
currentMessageId = messageId; // 记录当前请求的messageId console.log("🚀 发送请求内容: ", params, "请求ID:", currentRequestId)
console.log("🚀 发送请求内容: ", params) // #ifdef MP-WEIXIN
requestTask = uni.request({
url: BASE_URL + API, // 替换为你的接口地址
method: 'POST',
data: params,
enableChunked: true,
header: {
Accept: 'text/event-stream',
'Content-Type': 'application/json',
Authorization: `Bearer ${token}`, // 如需token可加
},
responseType: 'arraybuffer',
success(res) {
if (!isAborted && requestId === currentRequestId) {
console.log("✅ 请求成功，ID:", currentRequestId);
resolve(res.data);
} else {
console.log("❌ 请求已过期或终止，忽略success回调，当前ID:", requestId, "请求ID:", currentRequestId);
}
},
fail(err) {
if (!isAborted && requestId === currentRequestId) {
console.log("❌ 请求失败，ID:", currentRequestId, "错误:", JSON.stringify(err));
reject(err);
} else {
console.log("❌ 请求已过期或终止，忽略fail回调，当前ID:", requestId, "请求ID:", currentRequestId);
}
},
complete(res) {
if (!isAborted && requestId === currentRequestId && res.statusCode !== 200) {
console.log("❌ 请求完成但状态错误，ID:", currentRequestId, "状态:", res.statusCode);
if (onChunk) {
onChunk({ error: true, message: '服务器错误', detail: res });
}
reject(res);
} else if (requestId !== currentRequestId) {
console.log("❌ 请求已过期或终止，忽略complete回调，当前ID:", requestId, "请求ID:", currentRequestId);
}
}
});
// #ifdef MP-WEIXIN requestTask.onHeadersReceived(res => {
requestTask = uni.request({ // 检查请求是否已终止或过期
url: BASE_URL + API, // 替换为你的接口地址 if (isAborted || requestId !== currentRequestId) {
method: 'POST', console.log('🚫 Headers已终止或过期，忽略，当前ID:', requestId, '请求ID:', currentRequestId);
data: params, return;
enableChunked: true, }
header: {
Accept: 'text/event-stream', console.log('📡 onHeadersReceived，ID:', currentRequestId, res);
'Content-Type': 'application/json', const status = res.statusCode || (res.header && res.header.statusCode);
Authorization: `Bearer ${token}`, // 如需token可加 if (status && status !== 200) {
}, hasError = true;
responseType: 'arraybuffer', if (onChunk && !isAborted && requestId === currentRequestId) {
success(res) { onChunk({ error: true, message: `服务器错误(${status})`, detail: res });
if (!isAborted && currentMessageId === messageId) { }
console.log("✅ 请求成功"); if (requestTask && requestTask.abort) {
resolve(res.data); requestTask.abort();
} else if (currentMessageId !== messageId) { }
console.log("❌ 请求已过期（messageId不匹配），忽略success回调"); }
} else { });
console.log("❌ 请求已终止，忽略success回调");
} requestTask.onChunkReceived(res => {
}, // 第一道防线：立即检查请求ID和终止状态
fail(err) { if (isAborted || hasError || requestTask === null || requestId !== currentRequestId) {
if (!isAborted && currentMessageId === messageId) { console.log('🚫 数据块已终止或过期，忽略 - 第一道检查，当前ID:', requestId, '请求ID:', currentRequestId);
console.log("❌ 请求失败，错误:", JSON.stringify(err)); return;
reject(err); }
} else if (currentMessageId !== messageId) {
console.log("❌ 请求已过期（messageId不匹配），忽略fail回调"); console.log("📦 onChunkReceived，ID:", currentRequestId, res)
} else { const base64 = uni.arrayBufferToBase64(res.data);
console.log("❌ 请求已终止，忽略fail回调"); let data = '';
} try {
}, data = decodeURIComponent(escape(weAtob(base64)));
complete(res) { } catch (e) {
if (!isAborted && currentMessageId === messageId && res.statusCode !== 200) { // 某些平台可能不支持 atob，可以直接用 base64
console.log("❌ 请求完成但状态错误，状态:", res.statusCode); data = base64;
if (onChunk) { }
onChunk({ error: true, message: '服务器错误', detail: res });
} // 第二道防线：解析前再次检查
reject(res); if (isAborted || hasError || requestTask === null || requestId !== currentRequestId) {
} else if (currentMessageId !== messageId) { console.log('🚫 数据块已终止或过期，忽略 - 第二道检查，当前ID:', requestId, '请求ID:', currentRequestId);
console.log("❌ 请求已过期（messageId不匹配），忽略complete回调"); return;
} else if (isAborted) { }
console.log("❌ 请求已终止，忽略complete回调");
} const messages = parseSSEChunk(data);
} messages.forEach(msg => {
// 第三道防线：每个消息处理前都检查
if (onChunk && !isAborted && !hasError && requestTask !== null && requestId === currentRequestId) {
console.log(`parseSSEChunk ${currentRequestId}:`, msg)
onChunk(msg);
} else {
console.log('🚫 消息已终止或过期，忽略处理，当前ID:', requestId, '请求ID:', currentRequestId);
}
});
});
// #endif
}); });
requestTask.onHeadersReceived(res => { return promise
// 检查请求是否已终止或过期（messageId不匹配）
if (isAborted || currentMessageId !== messageId) {
if (currentMessageId !== messageId) {
console.log('🚫 Headers已过期（messageId不匹配），忽略');
} else {
console.log('🚫 Headers已终止，忽略');
}
return;
}
console.log('📡 onHeadersReceived，res:', res);
const status = res.statusCode || (res.header && res.header.statusCode);
if (status && status !== 200) {
hasError = true;
if (onChunk && !isAborted && currentMessageId === messageId) {
onChunk({ error: true, message: `服务器错误(${status})`, detail: res });
}
if (requestTask && requestTask.abort) {
requestTask.abort();
}
}
});
requestTask.onChunkReceived(res => {
// 第一道防线：立即检查终止状态和messageId（防止处理过期请求）
if (isAborted || hasError || requestTask === null || currentMessageId !== messageId) {
if (currentMessageId !== messageId) {
console.log('🚫 数据块已过期（messageId不匹配），忽略 - 第一道检查');
} else {
console.log('🚫 数据块已终止或无效，忽略 - 第一道检查');
}
return;
}
const base64 = uni.arrayBufferToBase64(res.data);
let data = '';
try {
data = decodeURIComponent(escape(weAtob(base64)));
} catch (e) {
// 某些平台可能不支持 atob，可以直接用 base64
data = base64;
}
console.log("📦 onChunkReceived，res:", data)
// 第二道防线：解析前再次检查
if (isAborted || hasError || requestTask === null || currentMessageId !== messageId) {
if (currentMessageId !== messageId) {
console.log('🚫 数据块已过期（messageId不匹配），忽略 - 第二道检查');
} else {
console.log('🚫 数据块已终止或无效，忽略 - 第二道检查');
}
return;
}
const messages = parseSSEChunk(data);
messages.forEach(msg => {
// 第三道防线：每个消息处理前都检查（确保只处理当前请求的消息）
if (onChunk && !isAborted && !hasError && requestTask !== null && currentMessageId === messageId) {
onChunk(msg);
} else if (currentMessageId !== messageId) {
console.log('🚫 消息已过期（messageId不匹配），忽略处理');
} else {
console.log('🚫 消息已终止或无效，忽略处理');
}
});
});
// #endif
});
return promise
} }
// window.atob兼容性处理 // window.atob兼容性处理
@ -245,37 +226,23 @@ const weAtob = (string) => {
// 解析SSE分段数据 // 解析SSE分段数据
const parseSSEChunk = (raw) => { function parseSSEChunk(raw) {
const results = []; // 拆分为多段
const lines = raw.split('\n\n');
// 按一个或多个连续换行分段，表示每一个事件块 const results = [];
const chunks = raw.split(/\n\n+/); lines.forEach(line => {
// 只处理包含 data: 的行
for (const chunk of chunks) { const dataMatch = line.match(/data:(\{.*\})/);
const lines = chunk.split(/\r?\n/); if (dataMatch && dataMatch[1]) {
let dataLines = []; try {
const obj = JSON.parse(dataMatch[1]);
for (const line of lines) { results.push(obj);
if (line.startsWith('data:')) { } catch (e) {
// 提取data:后面的内容并去除首尾空格 // 解析失败忽略
dataLines.push(line.slice(5).trim()); }
} }
} });
return results;
if (dataLines.length > 0) {
// 拼接多行数据
const fullData = dataLines.join('\n');
try {
const obj = JSON.parse(fullData);
results.push(obj);
} catch (e) {
console.warn('⚠️ SSE数据解析失败:', e, '原始数据:', fullData);
// 解析失败忽略
}
}
}
return results;
} }
export { agentChatStream, stopAbortTask } export { agentChatStream, stopAbortTask }