admin/nexus-terminal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Baobhan Sith
2025-05-15 20:17:31 +08:00
parent d59794928a
commit 0ce9f27fc6
3 changed files with 90 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/backend/src/services/sftp.service.ts
+47 -55
View File
@@ -1568,33 +1568,66 @@ export class SftpService {
const chunkBuffer = Buffer.from(dataBase64, 'base64');
const writeSuccess = uploadState.stream.write(chunkBuffer, (err) => {
if (err) {
// This callback handles errors specifically related to *this* write operation.
console.error(`[SFTP Upload ${uploadId}] Error writing chunk ${chunkIndex} to ${uploadState.remotePath}:`, err);
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: `写入块 ${chunkIndex} 失败: ${err.message}` } }));
// Consider cancelling the upload on write error
this.cancelUploadInternal(uploadId, `Write error on chunk ${chunkIndex}`);
} else {
uploadState.bytesWritten += chunkBuffer.length;
if (state.ws && state.ws.readyState === WebSocket.OPEN) {
const progressPercent = Math.round((uploadState.bytesWritten / uploadState.totalSize) * 100);
state.ws.send(JSON.stringify({
type: 'sftp:upload:progress',
uploadId: uploadId,
payload: {
bytesWritten: uploadState.bytesWritten,
totalSize: uploadState.totalSize,
progress: Math.min(100, progressPercent)
}
}));
}
if (uploadState.bytesWritten >= uploadState.totalSize) {
if (!uploadState.stream.writableEnded) {
uploadState.stream.end((endErr: Error & { code?: string } | undefined) => {
const streamStateInEndCallback = uploadState?.stream;
if (endErr) {
if (endErr.code === 'ERR_STREAM_DESTROYED' && uploadState && uploadState.bytesWritten >= uploadState.totalSize) {
console.warn(`[SFTP Upload ${uploadId}] stream.end() CALLBACK reported ERR_STREAM_DESTROYED, but all bytes written. UploadId: ${uploadId}. Error:`, endErr);
console.log(`[SFTP Upload ${uploadId}] Treating ERR_STREAM_DESTROYED as non-fatal for this upload. Expecting 'close' event to finalize success for ${uploadState.remotePath}.`);
} else {
console.error(`[SFTP Upload ${uploadId}] Error from stream.end() CALLBACK for ${uploadState?.remotePath || 'unknown path'}:`, endErr);
if (state && state.ws) {
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: `结束写入流时出错: ${endErr.message}` } }));
}
this.cancelUploadInternal(uploadId, `Stream end error: ${endErr.message}`, endErr);
}
}
});
}
}
}
});
if (!writeSuccess) {
// console.warn(`[SFTP Upload ${uploadId}] Write stream buffer full after chunk ${chunkIndex}. Pausing chunk processing until 'drain'.`);
if (!uploadState.drainPromise) {
// console.log(`[SFTP Upload ${uploadId}] Attaching new drain listener for chunk ${chunkIndex}`);
uploadState.drainPromise = new Promise<void>(resolve => {
uploadState.stream.once('drain', () => {
// console.log(`[SFTP Upload ${uploadId}] Drain event fired, resolving promise for chunk ${chunkIndex}`);
uploadState.drainPromise = null; // Reset for next time
uploadState.drainPromise = null;
resolve();
});
});
} else {
// console.log(`[SFTP Upload ${uploadId}] Waiting for existing drain promise for chunk ${chunkIndex}`);
}
try {
await uploadState.drainPromise;
// console.log(`[SFTP Upload ${uploadId}] Resumed after drain for chunk ${chunkIndex}`);
} catch (drainError) {
console.error(`[SFTP Upload ${uploadId}] Error awaiting drain promise for chunk ${chunkIndex}:`, drainError);
this.cancelUploadInternal(uploadId, 'Error waiting for drain promise');
@@ -1602,51 +1635,10 @@ export class SftpService {
}
}
uploadState.bytesWritten += chunkBuffer.length;
// +++ 发送进度更新 +++
if (state.ws && state.ws.readyState === WebSocket.OPEN) {
const progressPercent = Math.round((uploadState.bytesWritten / uploadState.totalSize) * 100);
// 将 uploadId 放在顶层，payload 中包含具体进度数据
state.ws.send(JSON.stringify({
type: 'sftp:upload:progress',
uploadId: uploadId,
payload: {
bytesWritten: uploadState.bytesWritten,
totalSize: uploadState.totalSize,
progress: Math.min(100, progressPercent) // 确保不超过100
}
}));
}
// --- 结束发送进度更新 ---
// Check if upload is complete
if (uploadState.bytesWritten > uploadState.totalSize) {
console.error(`[SFTP Upload ${uploadId}] Bytes written (${uploadState.bytesWritten}) exceeded total size (${uploadState.totalSize}) for ${uploadState.remotePath}.`);
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: '写入字节数超过文件总大小' } }));
this.cancelUploadInternal(uploadId, 'Bytes written exceeded total size');
} else if (uploadState.bytesWritten >= uploadState.totalSize) {
if (!uploadState.stream.writableEnded) {
uploadState.stream.end((endErr: Error & { code?: string } | undefined) => { // Added code to Error type hint
const streamStateInEndCallback = uploadState?.stream; // Re-fetch in case it changed
if (endErr) {
// Check if it's the specific ERR_STREAM_DESTROYED error and all bytes were written
if (endErr.code === 'ERR_STREAM_DESTROYED' && uploadState && uploadState.bytesWritten >= uploadState.totalSize) {
console.warn(`[SFTP Upload ${uploadId}] stream.end() CALLBACK reported ERR_STREAM_DESTROYED, but all bytes written. UploadId: ${uploadId}. Error:`, endErr);
console.log(`[SFTP Upload ${uploadId}] Treating ERR_STREAM_DESTROYED as non-fatal for this upload. Expecting 'close' event to finalize success for ${uploadState.remotePath}.`);
} else {
console.error(`[SFTP Upload ${uploadId}] Error from stream.end() CALLBACK for ${uploadState?.remotePath || 'unknown path'}:`, endErr);
if (state && state.ws) {
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: `结束写入流时出错: ${endErr.message}` } }));
}
this.cancelUploadInternal(uploadId, `Stream end error: ${endErr.message}`, endErr);
}
}
});
}
}
} catch (error: any) {
console.error(`[SFTP Upload ${uploadId}] Error handling chunk ${chunkIndex} for ${uploadState?.remotePath}:`, error);
state.ws.send(JSON.stringify({ type: 'sftp:upload:error', payload: { uploadId, message: `处理块 ${chunkIndex} 时出错: ${error.message}` } }));
packages/backend/src/websocket/types.ts
+14 -20
View File
@@ -6,7 +6,7 @@ export interface AuthenticatedWebSocket extends WebSocket {
isAlive?: boolean;
userId?: number;
username?: string;
sessionId?: string; // 用于关联 ClientState 的唯一 ID
sessionId?: string;
}
// 中心化的客户端状态接口 (统一版本)
@@ -15,11 +15,11 @@ export interface ClientState { // 导出以便 Service 可以导入
sshClient: Client;
sshShellStream?: ClientChannel;
dbConnectionId: number;
connectionName?: string; // 添加连接名称字段
sftp?: SFTPWrapper; // 添加 sftp 实例 (由 SftpService 管理)
statusIntervalId?: NodeJS.Timeout; // 添加状态轮询 ID (由 StatusMonitorService 管理)
connectionName?: string; // 连接名称字段
sftp?: SFTPWrapper; // sftp 实例 (由 SftpService 管理)
statusIntervalId?: NodeJS.Timeout; // 状态轮询 ID (由 StatusMonitorService 管理)
dockerStatusIntervalId?: NodeJS.Timeout; // Docker 状态轮询 ID
ipAddress?: string; // 添加 IP 地址字段
ipAddress?: string; // IP 地址字段
isShellReady?: boolean; // 标记 Shell 是否已准备好处理输入和调整大小
isSuspendedByService?: boolean; // 标记此会话是否已被 SshSuspendService 接管
isMarkedForSuspend?: boolean; // 标记此会话是否已被用户请求挂起（等待断开连接）
@@ -37,14 +37,14 @@ export interface PortInfo {
// --- Docker Interfaces (Ensure this matches frontend and DockerService) ---
// Stats 接口
export interface DockerStats {
ID: string; // 来自 docker stats
Name: string; // 来自 docker stats
CPUPerc: string; // 来自 docker stats
MemUsage: string; // 来自 docker stats
MemPerc: string; // 来自 docker stats
NetIO: string; // 来自 docker stats
BlockIO: string; // 来自 docker stats
PIDs: string; // 来自 docker stats
ID: string;
Name: string;
CPUPerc: string;
MemUsage: string;
MemPerc: string;
NetIO: string;
BlockIO: string;
PIDs: string;
}
// Container 接口 (包含 stats)
@@ -246,13 +246,7 @@ export type SshSuspendServerToClientMessages =
| SshMarkedForSuspendAck
| SshUnmarkedForSuspendAck;
// It might be useful to have a general type for incoming messages if not already present
// For example, if you have a main message handler:
// export type WebSocketMessage = BaseMessageType | SshSuspendClientToServerMessages | OtherFeatureMessages;
// And for outgoing:
// export type WebSocketResponse = BaseResponseType | SshSuspendServerToClientMessages | OtherFeatureResponses;
// This part depends on the existing structure, so I'm providing the specific types for now.
// --- SFTP Compress/Decompress Message Types ---
// C -> S: Request to compress files/directories
export interface SftpCompressRequestPayload {