dce_router/
protocol.rs

1use std::any::Any;
2use std::fmt::Debug;
3use bytes::{Bytes, BytesMut};
4use dce_util::result::{DceError, DceResult};
5use log::{error, warn};
6use std::collections::HashMap;
7use std::ops::{Deref, DerefMut};
8use crate::api::{Api};
9
10#[cfg(feature = "async")]
11use async_trait::async_trait;
12
13
14pub const CTX_KEY_SESSION: &str = "$#session#";
15pub const CTX_KEY_RESP_SID: &str = "$#response-sid#";
16
17#[derive(Debug)]
18pub struct RpMeta<Req> {
19    req: Req,
20    resp_buffer: BytesMut,
21    ctx_data: HashMap<String, Box<dyn Any + Send>>,
22}
23
24impl<Req> RpMeta<Req> {
25    pub fn new(req: Req, ctx_data: Option<HashMap<String, Box<dyn Any + Send>>>) -> Self {
26        Self { req, resp_buffer: Default::default(), ctx_data: ctx_data.unwrap_or_default() }
27    }
28
29    pub fn raw(&self) -> &Req {
30        &self.req
31    }
32
33    pub fn raw_mut(&mut self) -> &mut Req {
34        &mut self.req
35    }
36
37    pub fn clear_buffer(&mut self) -> Vec<u8> {
38        self.resp_buffer.split().to_vec()
39    }
40
41    pub fn is_resp_empty(&self) -> bool {
42        self.resp_buffer.is_empty()
43    }
44
45    pub fn write(&mut self, bytes: Bytes) {
46        self.resp_buffer.extend(bytes);
47    }
48
49    pub fn write_string(&mut self, str: String) {
50        self.resp_buffer.extend(str.as_bytes());
51    }
52
53    pub fn set_ctx_data(&mut self, key: String, value: Box<dyn Any + Send>) {
54        self.ctx_data.insert(key, value);
55    }
56    
57    pub fn ctx_data(&self, key: &str) -> Option<&Box<dyn Any + Send>> {
58        return self.ctx_data.get(key)
59    }
60    
61    pub fn ctx_data_remove(&mut self, key: &str) -> Option<Box<dyn Any + Send>> {
62        return self.ctx_data.remove(key)
63    }
64
65    pub fn sid(&self) -> Option<&str> {
66        None
67    }
68
69    pub fn set_session(&mut self) {
70        // todo
71    }
72    
73    pub fn session(&self) {
74        // todo
75    }
76    
77    pub fn set_resp_sid(&mut self, sid: String) {
78        self.ctx_data.insert(CTX_KEY_RESP_SID.to_string(), Box::new(sid));
79    }
80    
81    pub fn resp_sid(&self) -> Option<&str> {
82        self.ctx_data.get(CTX_KEY_RESP_SID).map(
83            | boxed | boxed.downcast_ref::<String>().map(
84                | resp_sid | resp_sid.as_str())
85        ).flatten()
86    }
87}
88
89
90#[cfg_attr(feature = "async", async_trait)]
91pub trait RoutableProtocol: Sized + Deref<Target = RpMeta<Self::Req>> + DerefMut + Send {
92    type Req;
93
94    fn id(&self) -> isize { 0 }
95
96    fn path(&self) -> &str;
97
98    fn sync_body(&mut self) -> DceResult<Bytes> {
99        unreachable!("Not implemented yet")
100    }
101
102    fn sync_body_string(&mut self) -> DceResult<String> {
103        self.sync_body().map(|bts| String::from_utf8_lossy(&bts).into_owned())
104    }
105
106    #[cfg(feature = "async")]
107    async fn body(&mut self) -> DceResult<Bytes> {
108        unreachable!("Not implemented yet")
109    }
110
111    #[cfg(feature = "async")]
112    async fn body_string(&mut self) -> DceResult<String> {
113        self.body().await.map(|bts| String::from_utf8_lossy(&bts).into_owned())
114    }
115
116    fn match_api(&self, apis: &Vec<&'static Api<Self>>) -> Option<&'static Api<Self>> {
117        apis.last().map(|a| *a)
118    }
119
120    fn try_print_err(&self, err: &Option<DceError>) {
121        if let Some(err) = err {
122            if err.public {
123                warn!("{}", err);
124            } else {
125                error!("{}", err);
126            }
127        }        
128    }
129}
dce_router/protocol.rs

dce_router/
protocol.rs
