* Copyright (C) 2023 Huawei Device Co., Ltd.
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#![allow(missing_docs)]
use crate::transfer;
use std::fs::metadata;
use std::collections::HashMap;
use std::io;
use std::path::Path;
use std::sync::Arc;
use std::io::{Error, ErrorKind};
#[cfg(feature = "host")]
extern crate ylong_runtime_static as ylong_runtime;
use ylong_runtime::sync::Mutex;
use crate::common::filemanager::FileManager;
use crate::common::hdctransfer::*;
use crate::config::CompressType;
use crate::config::HdcCommand;
use crate::config::MessageLevel;
use crate::config::TaskMessage;
use crate::config::MAX_SIZE_IOBUF;
use crate::serializer::serialize::Serialization;
use super::base::Base;
use super::hdctransfer;
use crate::serializer::native_struct::TransferConfig;
use crate::utils;
#[cfg(not(feature = "host"))]
use crate::utils::hdc_log::*;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct HdcFile {
pub file_cnt: u32,
pub dir_size: u64,
pub file_size: u64,
pub file_begin_time: u64,
pub dir_begin_time: u64,
pub transfer: HdcTransferBase,
}
impl HdcFile {
pub fn new(_session_id: u32, _channel_id: u32) -> Self {
Self {
transfer: HdcTransferBase::new(_session_id, _channel_id),
..Default::default()
}
}
}
type HdcFile_ = Arc<Mutex<HdcFile>>;
type FileTaskMap_ = Arc<Mutex<HashMap<(u32, u32), HdcFile_>>>;
pub struct FileTaskMap {}
impl FileTaskMap {
fn get_instance() -> FileTaskMap_ {
static mut MAP: Option<FileTaskMap_> = None;
unsafe {
MAP.get_or_insert_with(|| Arc::new(Mutex::new(HashMap::new())))
.clone()
}
}
pub async fn put(session_id: u32, channel_id: u32, value: HdcFile) {
let map = Self::get_instance();
let mut map = map.lock().await;
map.insert((session_id, channel_id), Arc::new(Mutex::new(value)));
}
pub async fn exsit(session_id: u32, channel_id: u32) -> bool {
let arc = Self::get_instance();
let map = arc.lock().await;
let task = map.get(&(session_id, channel_id));
task.is_some()
}
pub async fn remove(session_id: u32, channel_id: u32) -> Option<HdcFile_> {
let arc = Self::get_instance();
let mut map = arc.lock().await;
map.remove(&(session_id, channel_id))
}
pub async fn get(session_id: u32, channel_id: u32) -> Option<HdcFile_> {
let arc = Self::get_instance();
let map = arc.lock().await;
let task = map.get(&(session_id, channel_id));
task.cloned()
}
async fn stop_task(session_id: u32) {
let arc = Self::get_instance();
let map = arc.lock().await;
crate::info!("hdcfile stop task, session_id:{}, task_size: {}", session_id, map.len());
for _iter in map.iter() {
if _iter.0 .0 != session_id {
continue;
}
let mut task = _iter.1.lock().await;
task.transfer.stop_run = true;
crate::info!(
"session_id:{}, channel_id:{}, set stop_run as true.",
session_id,
_iter.0 .1
);
}
}
async fn dump_task() -> String {
let arc = Self::get_instance();
let map = arc.lock().await;
let mut result = String::new();
for _iter in map.iter() {
let task = _iter.1.lock().await;
let command = task.transfer.command_str.clone();
let line = format!(
"session_id:{},\tchannel_id:{},\tcommand:{}\n",
_iter.0 .0, _iter.0 .1, command
);
result.push_str(line.as_str());
}
result
}
}
async fn check_local_path(session_id: u32, channel_id: u32) -> bool {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"check_local_path get task is none session_id={session_id},channel_id={channel_id}"
);
return false;
};
let mut file_task = task.lock().await;
let local_path = file_task.transfer.local_path.clone();
let mut file_manager = FileManager::new(local_path);
let (open_result, err_msg) = file_manager.open();
if open_result {
file_task.transfer.transfer_config.file_size = file_manager.file_size();
file_task.transfer.file_size = file_task.transfer.transfer_config.file_size;
file_task.file_size = file_task.transfer.transfer_config.file_size;
file_task.transfer.transfer_config.optional_name = file_task.transfer.local_name.clone();
if transfer::base::CheckCompressVersion::get().await
&& (file_task.transfer.transfer_config.file_size > (MAX_SIZE_IOBUF as u64))
{
file_task.transfer.transfer_config.compress_type = CompressType::Lz4 as u8;
}
file_task.transfer.transfer_config.path = file_task.transfer.remote_path.clone();
let command_str = format!(
"[file send], local_path:{}, optional_name:{}",
file_task.transfer.local_path.clone(),
file_task.transfer.transfer_config.optional_name
);
file_task
.transfer
.command_str
.push_str(command_str.as_str());
return true;
} else {
hdctransfer::echo_client(
session_id,
channel_id,
err_msg.as_str(),
MessageLevel::Fail,
)
.await;
}
false
}
async fn echo_finish(session_id: u32, channel_id: u32, msg: String) {
hdctransfer::echo_client(
session_id,
channel_id,
msg.as_str(),
MessageLevel::Ok,
)
.await;
task_finish(session_id, channel_id).await;
}
pub async fn begin_transfer(session_id: u32, channel_id: u32, command: &String) -> bool {
let (argv, argc) = Base::split_command_to_args(command);
if argc < 2 {
echo_finish(
session_id,
channel_id,
"Transfer path split failed.".to_string(),
)
.await;
return false;
}
match set_master_parameters(session_id, channel_id, command, argc, argv).await {
Ok(_) => (),
Err(e) => {
echo_fail(session_id, channel_id, e, false).await;
return false;
}
}
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"begin_transfer get task is none session_id={session_id},channel_id={channel_id}"
);
return false;
};
let mut task = task.lock().await;
task.transfer.is_master = true;
drop(task);
let ret = check_local_path(session_id, channel_id).await;
if !ret {
do_file_finish(session_id, channel_id, &[1]).await;
return true;
}
put_file_check(session_id, channel_id).await;
true
}
async fn set_master_parameters(
session_id: u32,
channel_id: u32,
_command: &str,
argc: u32,
argv: Vec<String>,
) -> Result<bool, Error> {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"set_master_parameters get task is none session_id={session_id},channel_id={channel_id}"
);
return Err(Error::new(ErrorKind::Other, "Other failed"));
};
let mut task = task.lock().await;
let mut i: usize = 0;
let mut src_argv_index = 0u32;
if task.transfer.server_or_daemon {
src_argv_index += 2;
}
while i < argc as usize {
match &argv[i] as &str {
"-z" => {
task.transfer.transfer_config.compress_type = CompressType::Lz4 as u8;
src_argv_index += 1;
}
"-a" => {
task.transfer.transfer_config.hold_timestamp = true;
src_argv_index += 1;
}
"-sync" => {
task.transfer.transfer_config.update_if_new = true;
src_argv_index += 1;
}
"-m" => {
src_argv_index += 1;
}
"-remote" => {
src_argv_index += 1;
}
"-cwd" => {
src_argv_index += 2;
task.transfer.transfer_config.client_cwd = argv.get(i + 1).unwrap().clone();
}
_ => {}
}
i += 1;
}
if argc == src_argv_index {
crate::error!("set_master_parameters argc = {:#?} return false", argc);
return Err(Error::new(ErrorKind::Other, "There is no local and remote path"));
}
task.transfer.remote_path = argv.last().unwrap().clone();
task.transfer.local_path = argv.get(argv.len() - 2).unwrap().clone();
if task.transfer.server_or_daemon {
if src_argv_index + 1 == argc {
crate::error!("src_argv_index = {:#?} return false", src_argv_index);
return Err(Error::new(ErrorKind::Other, "There is no remote path"));
}
let cwd = task.transfer.transfer_config.client_cwd.clone();
task.transfer.local_path = Base::extract_relative_path(&cwd, &task.transfer.local_path);
} else if src_argv_index + 1 == argc {
task.transfer.remote_path = String::from(".");
task.transfer.local_path = argv.get((argc - 1) as usize).unwrap().clone();
}
task.transfer.local_name = Base::get_file_name(&mut task.transfer.local_path).unwrap();
match metadata(task.transfer.local_path.clone()) {
Ok(metadata) => {
if !metadata.is_dir() {
task.transfer.is_dir = false;
return Ok(true);
}
task.transfer.is_dir = true;
task.transfer.task_queue = get_sub_files_resurively(&task.transfer.local_path.clone());
task.transfer.base_local_path = get_base_path(task.transfer.local_path.clone());
if !task.transfer.task_queue.is_empty() {
task.transfer.local_path = task.transfer.task_queue.pop().unwrap();
task.transfer.local_name =
match Base::get_relative_path(&task.transfer.base_local_path, &task.transfer.local_path) {
Some(relative_path) => relative_path,
None => task.transfer.local_path.clone()
};
} else {
crate::error!("task transfer task_queue is empty");
return Err(Error::new(ErrorKind::Other, "Operation failed, because the source folder is empty."));
}
},
Err(error) => {
let err_msg = format!("Error opening file: {}, path: {}", error, task.transfer.local_path);
crate::error!("{}", err_msg);
return Err(Error::new(ErrorKind::Other, err_msg));
},
}
Ok(true)
}
fn get_base_path(path: String) -> String {
let p = Path::new(path.as_str());
let parent_path = p.parent();
if let Some(pp) = parent_path {
pp.display().to_string()
} else {
path
}
}
async fn put_file_check(session_id: u32, channel_id: u32) {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
return;
};
let task = task.lock().await;
let file_check_message = TaskMessage {
channel_id,
command: HdcCommand::FileCheck,
payload: task.transfer.transfer_config.serialize(),
};
transfer::put(task.transfer.session_id, file_check_message).await;
}
pub async fn check_slaver(session_id: u32, channel_id: u32, _payload: &[u8]) -> Result<bool, Error> {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"check_slaver get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return Err(Error::new(ErrorKind::Other, "Other failed"));
};
let mut task = task.lock().await;
let mut transconfig = TransferConfig {
..Default::default()
};
let _ = transconfig.parse(_payload.to_owned());
task.transfer.file_size = transconfig.file_size;
task.file_size = transconfig.file_size;
task.transfer.local_path = transconfig.path;
task.transfer.is_master = false;
task.transfer.index = 0;
let command_str = format!(
"[file recv],\t local_path:{},\t optional_name:{}\t",
task.transfer.local_path.clone(),
transconfig.optional_name
);
task.transfer.command_str.push_str(command_str.as_str());
let local_path = task.transfer.local_path.clone();
let optional_name = transconfig.optional_name.clone();
task.transfer.transfer_config.compress_type = transconfig.compress_type;
match hdctransfer::check_local_path(&mut task.transfer, &local_path, &optional_name) {
Ok(_) => (),
Err(e) => {
crate::error!("check_local_path return false channel_id={:#?}", channel_id);
return Err(e);
},
}
if task.transfer.transfer_config.update_if_new {
crate::error!("task.transfer.transfer_config.update_if_new is true");
return Err(Error::new(ErrorKind::Other, "Other failed"));
}
if task.dir_begin_time == 0 {
task.dir_begin_time = utils::get_current_time();
}
task.file_begin_time = utils::get_current_time();
Ok(true)
}
pub async fn wake_up_slaver(session_id: u32, channel_id: u32) {
let wake_up_message = TaskMessage {
channel_id,
command: HdcCommand::KernelWakeupSlavetask,
payload: Vec::<u8>::new(),
};
transfer::put(session_id, wake_up_message).await;
}
async fn put_file_begin(session_id: u32, channel_id: u32) {
let file_begin_message = TaskMessage {
channel_id,
command: HdcCommand::FileBegin,
payload: Vec::<u8>::new(),
};
transfer::put(session_id, file_begin_message).await;
}
async fn transfer_next(session_id: u32, channel_id: u32) -> bool {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"transfer_next get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return false;
};
let mut task = task.lock().await;
let Some(local_path) = task.transfer.task_queue.pop() else {
crate::error!(
"transfer_next get local path is none session_id={session_id:?},channel_id={channel_id:?}"
);
return false;
};
task.transfer.local_path = local_path;
task.transfer.local_name =
match Base::get_relative_path(&task.transfer.base_local_path, &task.transfer.local_path) {
Some(relative_path) => relative_path,
None => task.transfer.local_path.clone()
};
drop(task);
check_local_path(session_id, channel_id).await
}
async fn on_all_transfer_finish(session_id: u32, channel_id: u32) {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"on_all_transfer_finish get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return;
};
let task = task.lock().await;
let last_error = task.transfer.last_error;
let size = if task.file_cnt > 1 {
task.dir_size
} else {
task.file_size
};
let time = if task.file_cnt > 1 {
utils::get_current_time() - task.dir_begin_time
} else {
utils::get_current_time() - task.file_begin_time
};
let rate = size as f64 / time as f64;
#[allow(unused_variables)]
let message = if last_error == 0 {
format!(
"FileTransfer finish, Size:{}, File count = {}, time:{}ms rate:{:.2}kB/s",
size, task.file_cnt, time, rate
)
} else {
format!(
"Transfer failed: {}: {}",
task.transfer.local_path,
io::Error::from_raw_os_error(last_error as i32),
)
};
#[cfg(feature = "host")]
{
let level = if last_error == 0 {
transfer::EchoLevel::OK
} else {
transfer::EchoLevel::FAIL
};
let _ =
transfer::send_channel_msg(task.transfer.channel_id, level, message)
.await;
hdctransfer::close_channel(channel_id).await;
return;
}
#[allow(unreachable_code)]
{
let level = if last_error == 0 {
MessageLevel::Ok
} else {
MessageLevel::Fail
};
hdctransfer::echo_client(
task.transfer.session_id,
task.transfer.channel_id,
message.as_str(),
level,
)
.await;
hdctransfer::close_channel(channel_id).await;
}
}
async fn is_task_queue_empty(session_id: u32, channel_id: u32) -> bool {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"do_file_finish get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return false;
};
let task = task.lock().await;
task.transfer.task_queue.is_empty()
}
async fn do_file_finish(session_id: u32, channel_id: u32, _payload: &[u8]) {
if _payload[0] == 1 {
while !is_task_queue_empty(session_id, channel_id).await {
if transfer_next(session_id, channel_id).await {
put_file_check(session_id, channel_id).await;
return;
}
}
if is_task_queue_empty(session_id, channel_id).await {
let _finish_message = TaskMessage {
channel_id,
command: HdcCommand::FileFinish,
payload: [0].to_vec(),
};
transfer::put(session_id, _finish_message).await;
}
} else {
on_all_transfer_finish(session_id, channel_id).await;
task_finish(session_id, channel_id).await;
}
}
async fn put_file_finish(session_id: u32, channel_id: u32) {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"put_file_finish get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return;
};
let mut task = task.lock().await;
let _payload: [u8; 1] = [1];
task.file_cnt += 1;
task.dir_size += task.file_size;
let task_finish_message = TaskMessage {
channel_id,
command: HdcCommand::FileFinish,
payload: _payload.to_vec(),
};
transfer::put(session_id, task_finish_message).await;
}
pub async fn command_dispatch(
session_id: u32,
channel_id: u32,
_command: HdcCommand,
_payload: &[u8],
_payload_size: u16,
) -> bool {
match _command {
HdcCommand::FileInit => {
let s = String::from_utf8(_payload.to_vec());
match s {
Ok(str) => {
wake_up_slaver(session_id, channel_id).await;
begin_transfer(session_id, channel_id, &str).await;
}
Err(e) => {
let err_msg = format!("Transfer failed: arguments is invalid {:?}", e);
crate::error!("HdcCommand::FileInit: {}", err_msg);
echo_finish(session_id, channel_id, err_msg.to_string()).await;
}
}
}
HdcCommand::FileCheck => {
match check_slaver(session_id, channel_id, _payload).await {
Ok(_) => {
put_file_begin(session_id, channel_id).await;
},
Err(e) => {
echo_fail(session_id, channel_id, e, true).await;
}
}
}
HdcCommand::FileBegin => {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"command_dispatch get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return false;
};
let task = task.lock().await;
hdctransfer::transfer_begin(&task.transfer, HdcCommand::FileData).await;
}
HdcCommand::FileData => {
let Some(task) = FileTaskMap::get(session_id, channel_id).await else {
crate::error!(
"command_dispatch get task is none session_id={session_id:?},channel_id={channel_id:?}"
);
return false;
};
let mut task = task.lock().await;
if hdctransfer::transfer_data(&mut task.transfer, _payload) {
drop(task);
put_file_finish(session_id, channel_id).await;
}
}
HdcCommand::FileMode | HdcCommand::DirMode => {
put_file_mode(session_id, channel_id).await;
}
HdcCommand::FileFinish => {
do_file_finish(session_id, channel_id, _payload).await;
}
_ => {
crate::error!("others, command {:?}", _command);
}
}
true
}
async fn put_file_mode(session_id: u32, channel_id: u32) {
let task_message = TaskMessage {
channel_id,
command: HdcCommand::FileMode,
payload: Vec::<u8>::new(),
};
transfer::put(session_id, task_message).await;
}
async fn task_finish(session_id: u32, channel_id: u32) {
hdctransfer::transfer_task_finish(channel_id, session_id).await;
}
pub async fn stop_task(session_id: u32) {
FileTaskMap::stop_task(session_id).await;
}
pub async fn dump_task() -> String {
FileTaskMap::dump_task().await
}
pub async fn echo_fail(session_id: u32, channel_id: u32, error: Error, is_checked: bool) {
let message = match FileTaskMap::get(session_id, channel_id).await {
Some(task) => {
if is_checked {
let task = task.lock().await;
format!("Error opening file: {}, path: {}", error, task.transfer.local_path)
} else {
format!("{}", error)
}
}
None => format!(
"Error opening file: {}, path: {}",
error,
"cannot get file path from FileTaskMap",
)
};
hdctransfer::echo_client(
session_id,
channel_id,
message.as_str(),
MessageLevel::Fail,
)
.await;
task_finish(session_id, channel_id).await;
}
