manager/orion-visor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ 优化上传逻辑.

Browse Source
This commit is contained in:
lijiahang
2024-02-22 16:31:45 +08:00
parent 5bd49d97f7
commit 1711981d80
11 changed files with 277 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
orion-ops-module-asset/orion-ops-module-asset-service/src/main/java/com/orion/ops/module/asset/handler/host/transfer/enums/TransferOperatorType.java
View File

@@ -4,7 +4,7 @@ import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 消息操作类型
* 传输操作类型
*
* @author Jiahang Li
* @version 1.0.0
@@ -14,20 +14,20 @@ import lombok.Getter;
@AllArgsConstructor
public enum TransferOperatorType {
/**
* 处理完成
*/
PROCESSED("processed"),
/**
* 开始上传
*/
UPLOAD_START("upload_start"),
UPLOAD_START("uploadStart"),
/**
* 上传完成
*/
UPLOAD_FINISH("upload_finish"),
UPLOAD_FINISH("uploadFinish"),
/**
* 上传失败
*/
UPLOAD_ERROR("uploadError"),
;

43
orion-ops-module-asset/orion-ops-module-asset-service/src/main/java/com/orion/ops/module/asset/handler/host/transfer/enums/TransferReceiverType.java Normal file
View File

@@ -0,0 +1,43 @@
package com.orion.ops.module.asset.handler.host.transfer.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 传输响应类型
*
* @author Jiahang Li
* @version 1.0.0
* @since 2024/2/21 22:03
*/
@Getter
@AllArgsConstructor
public enum TransferReceiverType {
/**
* 请求下一块上传数据
*/
NEXT_BLOCK("nextBlock"),
/**
* 请求下一个传输任务
*/
NEXT_TRANSFER("nextTransfer"),
;
private final String type;
public static TransferReceiverType of(String type) {
if (type == null) {
return null;
}
for (TransferReceiverType value : values()) {
if (value.type.equals(type)) {
return value;
}
}
return null;
}
}

72
orion-ops-module-asset/orion-ops-module-asset-service/src/main/java/com/orion/ops/module/asset/handler/host/transfer/handler/TransferHandler.java
View File

@@ -4,12 +4,14 @@ import com.alibaba.fastjson.JSON;
import com.orion.lang.exception.argument.InvalidArgumentException;
import com.orion.lang.utils.io.Streams;
import com.orion.net.host.SessionStore;
import com.orion.ops.framework.common.constant.Const;
import com.orion.ops.framework.common.constant.ErrorMessage;
import com.orion.ops.framework.common.constant.ExtraFieldConst;
import com.orion.ops.framework.websocket.core.utils.WebSockets;
import com.orion.ops.module.asset.entity.dto.HostTerminalConnectDTO;
import com.orion.ops.module.asset.enums.HostConnectTypeEnum;
import com.orion.ops.module.asset.handler.host.transfer.enums.TransferOperatorType;
import com.orion.ops.module.asset.handler.host.transfer.enums.TransferReceiverType;
import com.orion.ops.module.asset.handler.host.transfer.model.TransferOperatorRequest;
import com.orion.ops.module.asset.handler.host.transfer.model.TransferOperatorResponse;
import com.orion.ops.module.asset.handler.host.transfer.session.ITransferHostSession;
@@ -72,6 +74,10 @@ public class TransferHandler implements ITransferHandler {
// 上传完成
this.uploadFinish();
break;
case UPLOAD_ERROR:
// 上传失败
this.uploadError();
break;
default:
break;
}
@@ -79,23 +85,18 @@ public class TransferHandler implements ITransferHandler {
@Override
public void putContent(byte[] content) {
Exception ex = null;
try {
// 写入内容
currentSession.putContent(content);
// 响应结果
this.sendMessage(TransferReceiverType.NEXT_BLOCK, null);
} catch (IOException e) {
ex = e;
log.error("TransferHandler.putContent error", e);
// 写入完成
currentSession.putFinish();
// 响应结果
this.sendMessage(TransferReceiverType.NEXT_TRANSFER, e);
}
// 响应结果
TransferOperatorResponse resp = TransferOperatorResponse.builder()
.type(TransferOperatorType.PROCESSED.getType())
.success(ex == null)
.msg(this.getErrorMessage(ex))
.build();
WebSockets.sendText(this.channel, JSON.toJSONString(resp));
}
/**
@@ -104,20 +105,18 @@ public class TransferHandler implements ITransferHandler {
* @param payload payload
*/
private void uploadStart(TransferOperatorRequest payload) {
Exception ex = null;
try {
// 开始上传
currentSession.startUpload(payload.getPath());
// 响应结果
this.sendMessage(TransferReceiverType.NEXT_BLOCK, null);
} catch (Exception e) {
ex = e;
log.error("TransferHandler.uploadStart error", e);
// 传输完成
currentSession.putFinish();
// 响应结果
this.sendMessage(TransferReceiverType.NEXT_TRANSFER, e);
}
// 响应结果
TransferOperatorResponse resp = TransferOperatorResponse.builder()
.type(TransferOperatorType.PROCESSED.getType())
.success(ex == null)
.msg(this.getErrorMessage(ex))
.build();
WebSockets.sendText(this.channel, JSON.toJSONString(resp));
}
/**
@@ -126,11 +125,16 @@ public class TransferHandler implements ITransferHandler {
private void uploadFinish() {
currentSession.putFinish();
// 响应结果
TransferOperatorResponse resp = TransferOperatorResponse.builder()
.type(TransferOperatorType.PROCESSED.getType())
.success(true)
.build();
WebSockets.sendText(this.channel, JSON.toJSONString(resp));
this.sendMessage(TransferReceiverType.NEXT_TRANSFER, null);
}
/**
* 上传失败
*/
private void uploadError() {
currentSession.putFinish();
// 响应结果
this.sendMessage(TransferReceiverType.NEXT_TRANSFER, new InvalidArgumentException(Const.EMPTY));
}
/**
@@ -158,16 +162,26 @@ public class TransferHandler implements ITransferHandler {
} catch (Exception e) {
log.error("TransferHandler.getAndInitSession error", e);
// 响应结果
TransferOperatorResponse resp = TransferOperatorResponse.builder()
.type(TransferOperatorType.PROCESSED.getType())
.success(false)
.msg(this.getErrorMessage(e))
.build();
WebSockets.sendText(this.channel, JSON.toJSONString(resp));
this.sendMessage(TransferReceiverType.NEXT_TRANSFER, e);
return false;
}
}
/**
* 发送消息
*
* @param type type
* @param ex ex
*/
private void sendMessage(TransferReceiverType type, Exception ex) {
TransferOperatorResponse resp = TransferOperatorResponse.builder()
.type(type.getType())
.success(ex == null)
.msg(this.getErrorMessage(ex))
.build();
WebSockets.sendText(this.channel, JSON.toJSONString(resp));
}
/**
* 获取错误信息
*