use super::def::*;
use super::output_request::*;
use super::processor::*;
use crate::base::*;
use crate::requestor::*;
use crate::stack::SharedObjectStackDecID;
use cyfs_base::*;
use http_types::{Method, Request, Response, Url};
use std::sync::Arc;
pub struct NDNRequestorHelper;
impl NDNRequestorHelper {
pub async fn decode_get_data_response(
resp: &mut Response,
) -> BuckyResult<NDNGetDataOutputResponse> {
let data = Box::new(resp.take_body());
let attr: Option<u32> =
RequestorHelper::decode_optional_header(resp, cyfs_base::CYFS_ATTRIBUTES)?;
let attr = attr.map(|v| Attributes::new(v));
let object_id = RequestorHelper::decode_header(resp, cyfs_base::CYFS_OBJECT_ID)?;
let owner_id = RequestorHelper::decode_optional_header(resp, cyfs_base::CYFS_OWNER_ID)?;
let range = RequestorHelper::decode_optional_json_header(resp, cyfs_base::CYFS_DATA_RANGE)?;
let group = RequestorHelper::decode_optional_header_with_utf8_decoding(
resp,
cyfs_base::CYFS_TASK_GROUP,
)?;
let length: u64 =
RequestorHelper::decode_header(resp, http_types::headers::CONTENT_LENGTH)?;
let ret = NDNGetDataOutputResponse {
object_id,
owner_id,
attr,
range,
group,
length,
data,
};
Ok(ret)
}
}
#[derive(Clone)]
pub struct NDNRequestor {
dec_id: Option<SharedObjectStackDecID>,
requestor: HttpRequestorRef,
service_url: Url,
data_requestor: HttpRequestorRef,
data_service_url: Url,
}
impl NDNRequestor {
pub fn new(
dec_id: Option<SharedObjectStackDecID>,
requestor: HttpRequestorRef,
data_requestor: Option<HttpRequestorRef>,
) -> Self {
let url = format!("http://{}/ndn/", requestor.remote_addr());
let url = Url::parse(&url).unwrap();
let data_service_url = match &data_requestor {
Some(requestor) => {
let url = format!("http://{}/ndn/", requestor.remote_addr());
Url::parse(&url).unwrap()
}
None => url.clone(),
};
let data_requestor = data_requestor.unwrap_or(requestor.clone());
Self {
dec_id,
requestor,
service_url: url,
data_requestor,
data_service_url,
}
}
pub fn into_processor(self) -> NDNOutputProcessorRef {
Arc::new(Box::new(self))
}
pub fn clone_processor(&self) -> NDNOutputProcessorRef {
self.clone().into_processor()
}
fn encode_common_headers(
&self,
action: NDNAction,
com_req: &NDNOutputRequestCommon,
http_req: &mut Request,
) {
if let Some(dec_id) = &com_req.dec_id {
http_req.insert_header(cyfs_base::CYFS_DEC_ID, dec_id.to_string());
} else if let Some(dec_id) = &self.dec_id {
if let Some(dec_id) = dec_id.get() {
http_req.insert_header(cyfs_base::CYFS_DEC_ID, dec_id.to_string());
}
}
http_req.insert_header(cyfs_base::CYFS_NDN_ACTION, action.to_string());
http_req.insert_header(cyfs_base::CYFS_API_LEVEL, com_req.level.to_string());
if let Some(target) = &com_req.target {
http_req.insert_header(cyfs_base::CYFS_TARGET, target.to_string());
}
RequestorHelper::encode_opt_header_with_encoding(
http_req,
cyfs_base::CYFS_REQ_PATH,
com_req.req_path.as_deref(),
);
if !com_req.referer_object.is_empty() {
RequestorHelper::insert_headers_with_encoding(
http_req,
cyfs_base::CYFS_REFERER_OBJECT,
&com_req.referer_object,
);
}
http_req.insert_header(cyfs_base::CYFS_FLAGS, com_req.flags.to_string());
}
fn encode_put_data_request(&self, req: &NDNPutDataOutputRequest) -> Request {
let mut http_req = Request::new(Method::Put, self.data_service_url.clone());
self.encode_common_headers(NDNAction::PutData, &req.common, &mut http_req);
http_req.insert_header(cyfs_base::CYFS_OBJECT_ID, req.object_id.to_string());
http_req
}
async fn decode_put_data_response(
&self,
resp: &Response,
) -> BuckyResult<NDNPutDataOutputResponse> {
let result: NDNPutDataResult =
RequestorHelper::decode_header(resp, cyfs_base::CYFS_RESULT)?;
let ret = NDNPutDataOutputResponse { result };
Ok(ret)
}
#[allow(unused_mut)]
pub async fn put_data(
&self,
mut req: NDNPutDataOutputRequest,
) -> BuckyResult<NDNPutDataOutputResponse> {
let mut http_req = self.encode_put_data_request(&req);
#[cfg(debug_assertions)]
{
use async_std::io::ReadExt;
let mut data = Vec::new();
req.data.read_to_end(&mut data).await.map_err(|e| {
let msg = format!("read data failed! chunk={} {}", req.object_id, e);
error!("{}", msg);
BuckyError::new(BuckyErrorCode::IoError, msg)
})?;
if data.len() != req.length as usize {
error!(
"chunk length unmatch: calc={}, expect={}",
data.len(),
req.length,
);
unreachable!();
}
let calc_id = ChunkId::calculate_sync(&data).unwrap();
if calc_id.object_id() != req.object_id {
error!(
"chunk id unmatch: calc_id={}, expect={}",
calc_id, req.object_id,
);
unreachable!();
}
http_req.set_body(data);
}
#[cfg(not(debug_assertions))]
{
let reader = async_std::io::BufReader::new(req.data);
let body = tide::Body::from_reader(reader, Some(req.length as usize));
http_req.set_body(body);
}
let mut resp = self.data_requestor.request(http_req).await?;
if resp.status().is_success() {
info!("put data to ndn service success: {}", req.object_id);
self.decode_put_data_response(&resp).await
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"put data to ndn service error! object={}, {}",
req.object_id, e
);
Err(e)
}
}
fn encode_put_shared_data_request(&self, req: &NDNPutDataOutputRequest) -> Request {
let mut http_req = Request::new(Method::Put, self.service_url.clone());
self.encode_common_headers(NDNAction::PutSharedData, &req.common, &mut http_req);
http_req.insert_header(cyfs_base::CYFS_OBJECT_ID, req.object_id.to_string());
http_req
}
async fn decode_put_shared_data_response(
&self,
resp: &Response,
) -> BuckyResult<NDNPutDataOutputResponse> {
let result: NDNPutDataResult =
RequestorHelper::decode_header(resp, cyfs_base::CYFS_RESULT)?;
let ret = NDNPutDataOutputResponse { result };
Ok(ret)
}
pub async fn put_shared_data(
&self,
req: NDNPutDataOutputRequest,
) -> BuckyResult<NDNPutDataOutputResponse> {
info!("will put_shared_data: {}", req);
let mut http_req = self.encode_put_shared_data_request(&req);
let reader = async_std::io::BufReader::new(req.data);
let body = tide::Body::from_reader(reader, Some(req.length as usize));
http_req.set_body(body);
let mut resp = self.requestor.request(http_req).await?;
if resp.status().is_success() {
info!("put shared data to ndn service success: {}", req.object_id);
self.decode_put_shared_data_response(&resp).await
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"put shared data to ndn service error! object={}, {}",
req.object_id, e
);
Err(e)
}
}
fn encode_get_data_request(&self, action: NDNAction, req: &NDNGetDataOutputRequest) -> Request {
let mut http_req = Request::new(Method::Get, self.data_service_url.clone());
self.encode_common_headers(action, &req.common, &mut http_req);
http_req.insert_header(cyfs_base::CYFS_OBJECT_ID, req.object_id.to_string());
RequestorHelper::encode_opt_header_with_encoding(
&mut http_req,
cyfs_base::CYFS_INNER_PATH,
req.inner_path.as_deref(),
);
RequestorHelper::encode_opt_header_with_encoding(
&mut http_req,
cyfs_base::CYFS_CONTEXT,
req.context.as_deref(),
);
RequestorHelper::encode_opt_header_with_encoding(
&mut http_req,
cyfs_base::CYFS_TASK_GROUP,
req.group.as_deref(),
);
if let Some(ref range) = req.range {
http_req.insert_header("Range", range.encode_string());
}
http_req
}
pub async fn get_data(
&self,
req: NDNGetDataOutputRequest,
) -> BuckyResult<NDNGetDataOutputResponse> {
let http_req = self.encode_get_data_request(NDNAction::GetData, &req);
let mut resp = self.data_requestor.request(http_req).await?;
if resp.status().is_success() {
match NDNRequestorHelper::decode_get_data_response(&mut resp).await {
Ok(resp) => {
info!("get data from ndn service success: {}", resp);
Ok(resp)
}
Err(e) => {
error!("decode get data response error: {}, {}", req.object_id, e);
Err(e)
}
}
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"get data from ndn service error: object={}, {}",
req.object_id, e
);
Err(e)
}
}
async fn decode_get_shared_data_response(
&self,
_req: &NDNGetDataOutputRequest,
resp: &mut Response,
) -> BuckyResult<NDNGetDataOutputResponse> {
let data = Box::new(resp.take_body());
let attr: Option<u32> =
RequestorHelper::decode_optional_header(resp, cyfs_base::CYFS_ATTRIBUTES)?;
let attr = attr.map(|v| Attributes::new(v));
let object_id = RequestorHelper::decode_header(resp, cyfs_base::CYFS_OBJECT_ID)?;
let owner_id = RequestorHelper::decode_optional_header(resp, cyfs_base::CYFS_OWNER_ID)?;
let range = RequestorHelper::decode_optional_json_header(resp, cyfs_base::CYFS_DATA_RANGE)?;
let group = RequestorHelper::decode_optional_header_with_utf8_decoding(
resp,
cyfs_base::CYFS_TASK_GROUP,
)?;
let length: u64 =
RequestorHelper::decode_header(resp, http_types::headers::CONTENT_LENGTH)?;
let ret = NDNGetDataOutputResponse {
object_id,
owner_id,
attr,
range,
group,
length,
data,
};
Ok(ret)
}
pub async fn get_shared_data(
&self,
req: NDNGetDataOutputRequest,
) -> BuckyResult<NDNGetDataOutputResponse> {
let http_req = self.encode_get_data_request(NDNAction::GetSharedData, &req);
let mut resp = self.requestor.request(http_req).await?;
if resp.status().is_success() {
info!("get data from ndn service success: {}", req.object_id);
self.decode_get_shared_data_response(&req, &mut resp).await
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"get data from ndn service error: object={}, {}",
req.object_id, e
);
Err(e)
}
}
fn encode_delete_data_request(&self, req: &NDNDeleteDataOutputRequest) -> Request {
let mut http_req = Request::new(Method::Delete, self.service_url.clone());
self.encode_common_headers(NDNAction::DeleteData, &req.common, &mut http_req);
http_req.insert_header(cyfs_base::CYFS_OBJECT_ID, req.object_id.to_string());
if let Some(inner_path) = &req.inner_path {
http_req.insert_header(cyfs_base::CYFS_INNER_PATH, inner_path);
}
http_req
}
async fn decode_delete_data_response(
&self,
resp: &Response,
) -> BuckyResult<NDNDeleteDataOutputResponse> {
let object_id = RequestorHelper::decode_header(resp, cyfs_base::CYFS_OBJECT_ID)?;
let ret = NDNDeleteDataOutputResponse { object_id };
Ok(ret)
}
pub async fn delete_data(
&self,
req: NDNDeleteDataOutputRequest,
) -> BuckyResult<NDNDeleteDataOutputResponse> {
let http_req = self.encode_delete_data_request(&req);
let mut resp = self.requestor.request(http_req).await?;
if resp.status().is_success() {
info!("delete data from ndn service success: {}", req.object_id);
self.decode_delete_data_response(&resp).await
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"delete data from ndn service error! object={}, {}",
req.object_id, e
);
Err(e)
}
}
fn encode_query_file_request(&self, req: &NDNQueryFileOutputRequest) -> Request {
let mut url = self.service_url.clone();
let (t, v) = req.param.to_key_pair();
url.query_pairs_mut()
.append_pair("type", t)
.append_pair("value", &v);
let mut http_req = Request::new(Method::Get, url);
self.encode_common_headers(NDNAction::QueryFile, &req.common, &mut http_req);
http_req
}
async fn decode_query_file_response(
&self,
resp: &mut Response,
) -> BuckyResult<NDNQueryFileOutputResponse> {
let ret: NDNQueryFileOutputResponse = RequestorHelper::decode_json_body(resp).await?;
Ok(ret)
}
async fn query_file(
&self,
req: NDNQueryFileOutputRequest,
) -> BuckyResult<NDNQueryFileOutputResponse> {
let http_req = self.encode_query_file_request(&req);
let mut resp = self.requestor.request(http_req).await?;
if resp.status().is_success() {
self.decode_query_file_response(&mut resp).await
} else {
let e = RequestorHelper::error_from_resp(&mut resp).await;
error!(
"query file from ndn service error! param={}, {}",
req.param, e
);
Err(e)
}
}
}
#[async_trait::async_trait]
impl NDNOutputProcessor for NDNRequestor {
async fn put_data(
&self,
req: NDNPutDataOutputRequest,
) -> BuckyResult<NDNPutDataOutputResponse> {
self.put_data(req).await
}
async fn get_data(
&self,
req: NDNGetDataOutputRequest,
) -> BuckyResult<NDNGetDataOutputResponse> {
self.get_data(req).await
}
async fn put_shared_data(
&self,
req: NDNPutDataOutputRequest,
) -> BuckyResult<NDNPutDataOutputResponse> {
self.put_shared_data(req).await
}
async fn get_shared_data(
&self,
req: NDNGetDataOutputRequest,
) -> BuckyResult<NDNGetDataOutputResponse> {
self.get_shared_data(req).await
}
async fn delete_data(
&self,
req: NDNDeleteDataOutputRequest,
) -> BuckyResult<NDNDeleteDataOutputResponse> {
self.delete_data(req).await
}
async fn query_file(
&self,
req: NDNQueryFileOutputRequest,
) -> BuckyResult<NDNQueryFileOutputResponse> {
self.query_file(req).await
}
}