♻️ 文件上传.

orion-ops-ui/src/views/host/terminal/components/transfer/transfer-drawer.vue

@@ -56,6 +56,18 @@
                     {{ item.parentPath }}
                   </span>
                 </a-tooltip>
+                <!-- 错误信息 -->
+                <a-tooltip v-if="item.errorMessage"
+                           position="top"
+                           :mini="true"
+                           :auto-fix-position="false"
+                           content-class="terminal-tooltip-content"
+                           arrow-class="terminal-tooltip-content"
+                           :content="item.errorMessage">
+                  <span class="error-message">
+                    {{ item.errorMessage }}
+                  </span>
+                </a-tooltip>
               </div>
               <!-- 右侧状态/操作-->
               <div class="transfer-item-right">
@@ -147,12 +159,16 @@
         max-width: 100%;
       }
 
-      .transfer-progress, .target-path {
+      .transfer-progress, .target-path, .error-message {
         padding-top: 4px;
         font-size: 13px;
         color: var(--color-neutral-8);
         width: fit-content;
       }
+
+      .error-message {
+        color: rgba(var(--red-6));
+      }
     }
 
     &-right {

orion-ops-ui/src/views/host/terminal/handler/sftp-transfer-manager.ts

@@ -1,16 +1,14 @@
-import type { ISftpTransferManager, SftpTransferItem } from '../types/terminal.type';
+import type { ISftpTransferManager, ISftpTransferUploader, SftpTransferItem } from '../types/terminal.type';
 import { TransferOperatorResponse } from '../types/terminal.type';
 import { TransferOperatorType, TransferStatus, TransferType } from '../types/terminal.const';
-import { sleep } from '@/utils';
 import { Message } from '@arco-design/web-vue';
 import { getTerminalAccessToken } from '@/api/asset/host-terminal';
-import { getPath } from '@/utils/file';
+import SftpTransferUploader from '@/views/host/terminal/handler/sftp-transfer-uploader';
-
-export const BLOCK_SIZE = 1024 * 1024;
 
 export const wsBase = import.meta.env.VITE_WS_BASE_URL;
 
 // todo 考虑一下单文件上传失败 (网络/文件被删除)
+// todo 取消任务
 
 // sftp 传输管理器实现
 export default class SftpTransferManager implements ISftpTransferManager {
@@ -19,9 +17,11 @@ export default class SftpTransferManager implements ISftpTransferManager {
 
   private run: boolean;
 
-  private resp?: TransferOperatorResponse;
+  private currentItem?: SftpTransferItem;
 
-  transferList: Array<SftpTransferItem>;
+  private currentUploader?: ISftpTransferUploader;
+
+  public transferList: Array<SftpTransferItem>;
 
   constructor() {
     this.run = false;
@@ -33,12 +33,13 @@ export default class SftpTransferManager implements ISftpTransferManager {
     this.transferList.push(...items);
     // 开始传输
     if (!this.run) {
-      this.startTransfer();
+      this.openClient();
     }
   }
 
   // 打开会话
   private async openClient() {
+    this.run = true;
     // 获取 access
     const { data: accessToken } = await getTerminalAccessToken();
     // 打开会话
@@ -47,7 +48,11 @@ export default class SftpTransferManager implements ISftpTransferManager {
       // 打开失败将传输列表置为失效
       Message.error('会话打开失败');
       console.error('error', event);
-      this.transferList.forEach(s => {
+      // 将等待中和传输中任务修改为失败状态
+      this.transferList.filter(s => {
+        return s.status === TransferStatus.WAITING
+          || s.status === TransferStatus.TRANSFERRING;
+      }).forEach(s => {
         s.status = TransferStatus.ERROR;
       });
     };
@@ -56,121 +61,83 @@ export default class SftpTransferManager implements ISftpTransferManager {
       this.run = false;
       console.warn('close', event);
     };
+    this.client.onopen = () => {
+      // 打开后自动传输下一个任务
+      this.transferNextItem();
+    };
     this.client.onmessage = this.resolveMessage.bind(this);
-    // 等待会话连接
-    for (let i = 0; i < 100; i++) {
-      await sleep(50);
-      if (this.client.readyState !== WebSocket.CONNECTING) {
-        break;
-      }
-    }
   }
 
-  // 开始传输
+  // 传输下一条任务
-  private async startTransfer() {
+  private transferNextItem() {
-    this.run = true;
+    this.currentUploader = undefined;
-    // 打开会话
+    // 获取任务
-    await this.openClient();
+    this.currentItem = this.transferList.find(s => s.status === TransferStatus.WAITING);
-    if (!this.run) {
+    if (this.currentItem) {
-      return;
-    }
-    // 开始传输
-    while (true) {
-      const item = this.transferList.find(s => s.status === TransferStatus.WAITING);
-      if (!item) {
-        break;
-      }
       // 开始传输
-      try {
+      if (this.currentItem.type === TransferType.UPLOAD) {
-        item.status = TransferStatus.TRANSFERRING;
+        // 上传
-        if (item.type === TransferType.UPLOAD) {
+        this.uploadFile();
-          // 上传
+      } else {
-          await this.uploadFile(item);
+        // 下载
-        } else {
+        this.uploadDownload();
-          // 下载
-          await this.uploadDownload(item);
-        }
-        item.status = TransferStatus.SUCCESS;
-      } catch (e) {
-        item.status = TransferStatus.ERROR;
       }
+    } else {
+      // 无任务关闭会话
+      this.client?.close();
     }
   }
 
   // 接收消息
   private async resolveMessage(message: MessageEvent) {
-    // TODO
+    const data = JSON.parse(message.data) as TransferOperatorResponse;
-    this.resp = JSON.parse(message.data);
+    if (data.type === TransferOperatorType.PROCESSED) {
-    //   // TODO 关闭会话
+      // 接收处理完成
-    //   this.client?.close();
+      this.resolveProcessed(data);
-    // }
+    }
   }
 
   // 上传文件
-  private async uploadFile(item: SftpTransferItem) {
+  private uploadFile() {
-    const file = item.file;
+    // 创建上传器
-    // 发送开始上传信息
+    this.currentUploader = new SftpTransferUploader(this.currentItem as SftpTransferItem, this.client as WebSocket);
-    this.client?.send(JSON.stringify({
+    // 开始上传
-      type: TransferOperatorType.UPLOAD_START,
+    this.currentUploader.startUpload();
-      path: getPath(item.parentPath + '/' + item.name),
-      hostId: item.hostId
-    }));
-    // TODO 等待处理结果 吧错误信息展示出来
-    try {
-      await this.awaitProcessedThrow();
-    } catch (ex: any) {
-      console.log(ex);
-      item.status = TransferStatus.ERROR;
-      item.errorMessage = ex.message;
-      return;
-    }
-    // 计算分片数量
-    const totalBlock = Math.ceil(file.size / BLOCK_SIZE);
-    // 分片上传
-    for (let i = 0; i < totalBlock; i++) {
-
-      // 读取数据
-      const start = i * BLOCK_SIZE;
-      const end = Math.min(file.size, start + BLOCK_SIZE);
-      const chunk = file.slice(start, end);
-      const reader = new FileReader();
-      const arrayBuffer = await new Promise((resolve, reject) => {
-        reader.onload = () => resolve(reader.result);
-        reader.onerror = (error) => reject(error);
-        reader.readAsArrayBuffer(chunk);
-      });
-      this.client?.send(arrayBuffer as ArrayBuffer);
-      // TODO 等待处理结果
-      await this.awaitProcessedThrow();
-    }
-    // TODO 发送 END
   }
 
   // 下载文件
-  private async uploadDownload(item: SftpTransferItem) {
+  private uploadDownload() {
     // TODO
   }
 
-  // 等待处理完成
+  // 接收处理完成回调
-  private async awaitProcessedThrow() {
+  private resolveProcessed(data: TransferOperatorResponse) {
-    for (let i = 0; i < 100; i++) {
+    // 操作回调
-      await sleep(50);
+    if (data.success) {
-      if (this.resp) {
+      // 操作成功
-        break;
+      if (this.currentUploader) {
-      }
+        if (this.currentUploader.hasNextBlock()) {
-    }
+          // 有下一个分片则上传 (上一个分片传输完成)
-    const resp = this.resp;
+          this.currentUploader.uploadNextBlock();
-    // const resp = undefined;
+        } else {
-    this.resp = undefined;
+          // 没有下一个分片则检查是否完成
-    // 抛出异常
+          if (this.currentUploader.finish) {
-    if (resp) {
+            // 已完成 开始下一个传输任务 (发送 finish 后的回调)
-      if (resp.success) {
+            this.transferNextItem();
-        return;
+          } else {
-      } else {
+            // 未完成则发送完成 (最后一个分片传输完成但还未发送 finish 指令)
-        throw new Error(resp.msg || '处理失败');
+            this.currentUploader.uploadFinish();
+          }
+        }
       }
     } else {
-      throw new Error('处理超时');
+      // 操作失败
+      if (this.currentUploader) {
+        // 上传失败
+        this.currentUploader.uploadError(data.msg);
+      }
+      // 开始下一个传输任务
+      this.transferNextItem();
     }
   }
 

orion-ops-ui/src/views/host/terminal/handler/sftp-transfer-uploader.ts

@@ -0,0 +1,78 @@
+import type { ISftpTransferUploader, SftpTransferItem } from '../types/terminal.type';
+import { TransferOperatorType, TransferStatus } from '../types/terminal.const';
+import { getPath } from '@/utils/file';
+
+export const BLOCK_SIZE = 1024 * 1024;
+
+// sftp 上传器实现
+export default class SftpTransferUploader implements ISftpTransferUploader {
+
+  public finish: boolean;
+  private currentBlock: number;
+  private totalBlock: number;
+  private client: WebSocket;
+  private item: SftpTransferItem;
+  private file: File;
+
+  constructor(item: SftpTransferItem, client: WebSocket) {
+    this.finish = false;
+    this.item = item;
+    this.client = client;
+    this.file = item.file;
+    this.currentBlock = 0;
+    this.totalBlock = Math.ceil(item.file.size / BLOCK_SIZE);
+  }
+
+  // 开始上传
+  startUpload() {
+    this.item.status = TransferStatus.TRANSFERRING;
+    // 发送开始上传信息
+    this.client?.send(JSON.stringify({
+      type: TransferOperatorType.UPLOAD_START,
+      path: getPath(this.item.parentPath + '/' + this.item.name),
+      hostId: this.item.hostId
+    }));
+  }
+
+  // 是否有下一个分片
+  hasNextBlock() {
+    return this.currentBlock < this.totalBlock;
+  }
+
+  // 上传下一个分片
+  async uploadNextBlock() {
+    // 读取数据
+    const start = this.currentBlock * BLOCK_SIZE;
+    const end = Math.min(this.file.size, start + BLOCK_SIZE);
+    const chunk = this.file.slice(start, end);
+    const reader = new FileReader();
+    const arrayBuffer = await new Promise((resolve, reject) => {
+      reader.onload = () => resolve(reader.result);
+      reader.onerror = (error) => reject(error);
+      reader.readAsArrayBuffer(chunk);
+    });
+    // 发送数据
+    this.client?.send(arrayBuffer as ArrayBuffer);
+    this.currentBlock++;
+    this.item.currentSize += (end - start);
+  }
+
+  // 上传完成
+  uploadFinish() {
+    this.finish = true;
+    this.item.status = TransferStatus.SUCCESS;
+    // 发送上传完成的信息
+    this.client?.send(JSON.stringify({
+      type: TransferOperatorType.UPLOAD_FINISH,
+      hostId: this.item.hostId
+    }));
+  }
+
+  // 上传失败
+  uploadError(msg: string | undefined) {
+    this.finish = true;
+    this.item.status = TransferStatus.ERROR;
+    this.item.errorMessage = msg || '上传失败';
+  }
+
+}

orion-ops-ui/src/views/host/terminal/types/terminal.type.ts

@@ -374,6 +374,22 @@ export interface ISftpTransferManager {
   addTransfer: (items: Array<SftpTransferItem>) => void;
 }
 
+// sftp 上传器定义
+export interface ISftpTransferUploader {
+  // 是否完成
+  finish: boolean;
+  // 开始上传
+  startUpload: () => void;
+  // 是否有下一个分片
+  hasNextBlock: () => boolean;
+  // 上传下一个分片
+  uploadNextBlock: () => void;
+  // 上传完成
+  uploadFinish: () => void;
+  // 上传失败
+  uploadError: (msg: string | undefined) => void;
+}
+
 // sftp 上传文件项
 export interface SftpTransferItem {
   id: string;