2024年07月26日

 
use super::*;
 
use toml;
 
#[derive(Deserialize, Debug, Clone)]
pub struct InfoConfig {
    pub type_code: Option<String>,
    pub type_no: Option<String>,
    pub data_dir: Option<String>,
}
 
#[derive(Deserialize, Debug, Clone)]
pub struct MqttConfig {
    pub id: Option<String>,
    pub host: Option<String>,
    pub port: Option<u16>,
    pub username: Option<String>,
    pub password: Option<String>,
}
 
#[derive(Deserialize, Debug, Clone)]
pub struct WebrtcConfig {
    pub stun_server: Option<String>,
    pub turn_server: Option<String>,
    pub turn_username: Option<String>,
    pub turn_credential: Option<String>,
}
 
#[derive(Deserialize, Debug, Clone)]
pub struct RootConfig {
    pub mqtt: Option<MqttConfig>,
    pub info: Option<InfoConfig>,
    pub webrtc: Option<WebrtcConfig>,
}
 
static ROOT_CONFIG_MAP: Lazy<Arc<RwLock<HashMap<String, RootConfig>>>> =
    Lazy::new(|| Arc::new(RwLock::new(HashMap::new())));
 
impl RootConfig {
    pub async fn from_file(file_path: &str) -> Result<Self, Box<dyn Error>> {
        let mut file_content = String::new();
 
        File::open(file_path)
            .await?
            .read_to_string(&mut file_content)
            .await?;
 
        Ok(toml::from_str(&file_content)?)
    }
 
    pub async fn get(key: &str) -> Result<RootConfig, Box<dyn Error>> {
        let root_config_option = ROOT_CONFIG_MAP.read()?.get(key).cloned();
 
        if let Some(root_config) = root_config_option {
            return Ok(root_config);
        }
 
        let root_config = RootConfig::from_file(key).await?;
 
        ROOT_CONFIG_MAP
            .write()?
            .insert(key.to_string(), root_config.clone());
 
        Ok(root_config)
    }
 
    pub fn insert(key: &str, root_config: RootConfig) -> Result<(), Box<dyn Error>> {
        ROOT_CONFIG_MAP
            .write()?
            .insert(key.to_string(), root_config);
 
        Ok(())
    }
 
    pub async fn get_default() -> Result<RootConfig, Box<dyn Error>> {
        RootConfig::get("config.toml").await
    }
 
    pub fn insert_default(root_config: RootConfig) -> Result<(), Box<dyn Error>> {
        RootConfig::insert("config.toml", root_config)
    }
}
 

 
impl Drop for WebrtcSession {
    fn drop(&mut self) {
        self.operation_objects
            .local_sender
            .try_send(("stop".to_owned(), "".to_owned()))
            .unwrap_or_else(|e| log::error!("{:?}", e));
 
        match WEBRTC_SESSION_MAP.write() {
            Ok(mut v) => {
                v.remove(&self.base.remote_session_id);
            }
            Err(e) => {
                log::error!("{:?}", e);
            }
        }
    }
}
 

 
    async fn file_read_in_chunks(
        &self,
        transaction: String,
        data: String,
    ) -> Result<(), Box<dyn Error>> {
        let root_config = RootConfig::get_default().await?;
 
        let path = format!("{}{}", root_config.info.unwrap().data_dir.unwrap(), data);
 
        let file = File::open(path.clone()).await?;
 
        let file_len = file.metadata().await?.len();
 
        let mut data_map: HashMap<String, Value> = HashMap::<String, Value>::new();
 
        data_map.insert("path".to_owned(), Value::String(data.clone()));
        data_map.insert("file_len".to_owned(), Value::Number(file_len.into()));
 
        let data_channel_result_message_dto: DataChannelResultMessageDto<HashMap<String, Value>> =
            DataChannelResultMessageDto {
                transaction: transaction.clone(),
                signalling_type: Some("result-file-start".to_owned()),
                data: Some(data_map.clone()),
            };
 
        let map_str = serde_json::to_string(&data_channel_result_message_dto)?;
 
        self.operation_objects
            .control_data_channel
            .send_text(map_str)
            .await?;
 
        let mut reader = BufReader::new(file);
 
        let read_size = 65000;
 
        let mut buffer = vec![0u8; read_size];
 
        loop {
            let n = reader.read(&mut buffer).await?;
 
            if n == 0 {
                break;
            }
 
            if n < read_size {
                buffer.truncate(n);
            }
 
            let mut send_data = Vec::<u8>::new();
 
            let file_path = data.as_bytes();
 
            let file_path_len = file_path.len() as u32;
 
            send_data.extend_from_slice(&file_path_len.to_le_bytes());
            send_data.extend_from_slice(&file_path);
            send_data.extend_from_slice(&buffer);
 
            let buf = Bytes::from(send_data);
 
            if let Err(e) = self.operation_objects.other_data_channel.send(&buf).await {
                log::error!("main_data_channel.send: {:?}", e);
 
                return Err(e.into());
            };
 
            loop {
                if matches!(
                    self.operation_objects.peer_connection.connection_state(),
                    RTCPeerConnectionState::Failed
                        | RTCPeerConnectionState::Disconnected
                        | RTCPeerConnectionState::Closed
                ) {
                    return Err(format!(
                        "peer connection  {}",
                        self.operation_objects.peer_connection.connection_state(),
                    )
                    .into());
                }
 
                let buffered_amount = self
                    .operation_objects
                    .other_data_channel
                    .buffered_amount()
                    .await;
 
                if buffered_amount + buf.len() > (1024 * 1024 * 10) {
                    time::sleep(std::time::Duration::from_millis(50)).await;
                } else {
                    break;
                }
            }
        }
 

const [_file_chunks_map, { set: file_chunks_map_set, get: file_chunks_map_get, remove: file_chunks_map_remove }] = useMap<string, Uint8Array[]>([]);
 
const webrtc_data_channel_message_fn = useMemoizedFn((type: String, message: any) => {
	webrtc_data_channel_data_add(`[${dayjs().format("YYYY-MM-DD HH:mm:ss.SSS")}] [${type} receiver] ${message}`);
 
	if (_.isString(message)) {
		let json = JSON.parse(message);
 
		const transaction = transactionMapGet(json?.transaction);
 
		if (transaction) {
			transaction(type, json);
		}
	} else {
		const receivedArrayBuffer = message as ArrayBuffer;
 
		const dataView = new DataView(receivedArrayBuffer);
 
		const headerLength = dataView.getUint32(0, true);
 
		const headerBytes = new Uint8Array(receivedArrayBuffer, 4, headerLength);
 
		const headerString = new TextDecoder().decode(headerBytes);
 
		const remainingBytes = new Uint8Array(receivedArrayBuffer, 4 + headerLength);
 
		const file_chunks = file_chunks_map_get(headerString) ?? [];
 
		file_chunks_map_set(headerString, [...file_chunks, remainingBytes]);
	}
});
 
const webrtc_control_data_channel_message_fn = useMemoizedFn((message: any) => {
	webrtc_data_channel_message_fn("control", message);
});
 
useEffect(() => {
	const type = "webrtc-control-data-channel-message";
 
	globalMitt.on(type, webrtc_control_data_channel_message_fn);
 
	return () => {
		globalMitt.off(type, webrtc_control_data_channel_message_fn);
	};
}, [webrtc_control_data_channel_message_fn]);
 
const webrtc_other_data_channel_message_fn = useMemoizedFn((message: any) => {
	webrtc_data_channel_message_fn("other", message);
});
 
useEffect(() => {
	const type = "webrtc-other-data-channel-message";
 
	globalMitt.on(type, webrtc_other_data_channel_message_fn);
 
	return () => {
		globalMitt.off(type, webrtc_other_data_channel_message_fn);
	};
}, [webrtc_other_data_channel_message_fn]);