diff options
Diffstat (limited to 'src/endpoint.rs')
-rw-r--r-- | src/endpoint.rs | 58 |
1 files changed, 43 insertions, 15 deletions
diff --git a/src/endpoint.rs b/src/endpoint.rs index 8ee64a5..ff626d8 100644 --- a/src/endpoint.rs +++ b/src/endpoint.rs @@ -10,11 +10,13 @@ use crate::netapp::*; use crate::util::*; /// This trait should be implemented by an object of your application -/// that can handle a message of type `M`. +/// that can handle a message of type `M`, if it wishes to handle +/// streams attached to the request and/or to send back streams +/// attached to the response.. /// /// The handler object should be in an Arc, see `Endpoint::set_handler` #[async_trait] -pub trait EndpointHandler<M>: Send + Sync +pub trait StreamingEndpointHandler<M>: Send + Sync where M: Message, { @@ -27,11 +29,34 @@ where /// it will panic if it is ever made to handle request. #[async_trait] impl<M: Message + 'static> EndpointHandler<M> for () { - async fn handle(self: &Arc<()>, _m: Req<M>, _from: NodeID) -> Resp<M> { + async fn handle(self: &Arc<()>, _m: &M, _from: NodeID) -> M::Response { panic!("This endpoint should not have a local handler."); } } +// ---- + +#[async_trait] +pub trait EndpointHandler<M>: Send + Sync +where + M: Message, +{ + async fn handle(self: &Arc<Self>, m: &M, from: NodeID) -> <M as Message>::Response; +} + +#[async_trait] +impl<T, M> StreamingEndpointHandler<M> for T +where + T: EndpointHandler<M>, + M: Message + 'static, +{ + async fn handle(self: &Arc<Self>, m: Req<M>, from: NodeID) -> Resp<M> { + Resp::new(EndpointHandler::handle(self, m.msg(), from).await) + } +} + +// ---- + /// This struct represents an endpoint for message of type `M`. /// /// Creating a new endpoint is done by calling `NetApp::endpoint`. @@ -41,13 +66,13 @@ impl<M: Message + 'static> EndpointHandler<M> for () { /// An `Endpoint` is used both to send requests to remote nodes, /// and to specify the handler for such requests on the local node. /// The type `H` represents the type of the handler object for -/// endpoint messages (see `EndpointHandler`). +/// endpoint messages (see `StreamingEndpointHandler`). pub struct Endpoint<M, H> where M: Message, - H: EndpointHandler<M>, + H: StreamingEndpointHandler<M>, { - phantom: PhantomData<M>, + _phantom: PhantomData<M>, netapp: Arc<NetApp>, path: String, handler: ArcSwapOption<H>, @@ -56,11 +81,11 @@ where impl<M, H> Endpoint<M, H> where M: Message, - H: EndpointHandler<M>, + H: StreamingEndpointHandler<M>, { pub(crate) fn new(netapp: Arc<NetApp>, path: String) -> Self { Self { - phantom: PhantomData::default(), + _phantom: PhantomData::default(), netapp, path, handler: ArcSwapOption::from(None), @@ -79,8 +104,10 @@ where } /// Call this endpoint on a remote node (or on the local node, - /// for that matter) - pub async fn call_full<T>( + /// for that matter). This function invokes the full version that + /// allows to attach a streaming body to the request and to + /// receive such a body attached to the response. + pub async fn call_streaming<T>( &self, target: &NodeID, req: T, @@ -112,15 +139,16 @@ where } } - /// Call this endpoint on a remote node, without the possibility - /// of adding or receiving a body + /// Call this endpoint on a remote node. This function is the simplified + /// version that doesn't allow to have streams attached to the request + /// or the response; see `call_streaming` for the full version. pub async fn call( &self, target: &NodeID, req: M, prio: RequestPriority, ) -> Result<<M as Message>::Response, Error> { - Ok(self.call_full(target, req, prio).await?.into_msg()) + Ok(self.call_streaming(target, req, prio).await?.into_msg()) } } @@ -144,13 +172,13 @@ pub(crate) trait GenericEndpoint { pub(crate) struct EndpointArc<M, H>(pub(crate) Arc<Endpoint<M, H>>) where M: Message, - H: EndpointHandler<M>; + H: StreamingEndpointHandler<M>; #[async_trait] impl<M, H> GenericEndpoint for EndpointArc<M, H> where M: Message + 'static, - H: EndpointHandler<M> + 'static, + H: StreamingEndpointHandler<M> + 'static, { async fn handle( &self, |