PR review

Author: rklaehn

examples/limit.rs

@@ -81,7 +81,7 @@ fn limit_by_node_id(allowed_nodes: HashSet<NodeId>) -> EventSender {
     let mask = EventMask {
         // We want a request for each incoming connection so we can accept
         // or reject them. We don't need any other events.
-        connected: ConnectMode::Request,
+        connected: ConnectMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
@@ -108,7 +108,7 @@ fn limit_by_hash(allowed_hashes: HashSet<Hash>) -> EventSender {
         // We want to get a request for each get request that we can answer
         // with OK or not OK depending on the hash. We do not want detailed
         // events once it has been decided to handle a request.
-        get: RequestMode::Request,
+        get: RequestMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
@@ -136,7 +136,7 @@ fn throttle(delay_ms: u64) -> EventSender {
     let mask = EventMask {
         // We want to get requests for each sent user data blob, so we can add a delay.
         // Other than that, we don't need any events.
-        throttle: ThrottleMode::Throttle,
+        throttle: ThrottleMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);
@@ -190,7 +190,7 @@ fn limit_max_connections(max_connections: usize) -> EventSender {
         // based on the current connection count if we want to accept or reject.
         // We also want detailed logging of events for the get request, so we can
         // detect when the request is finished one way or another.
-        connected: ConnectMode::Request,
+        connected: ConnectMode::Intercept,
         ..EventMask::DEFAULT
     };
     let (tx, mut rx) = EventSender::channel(32, mask);

src/provider/events.rs

@@ -16,6 +16,7 @@ use crate::{
     Hash,
 };
 
+/// Mode for connect events.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[repr(u8)]
 pub enum ConnectMode {
@@ -25,9 +26,10 @@ pub enum ConnectMode {
     /// We get a notification for connect events.
     Notify,
     /// We get a request for connect events and can reject incoming connections.
-    Request,
+    Intercept,
 }
 
+/// Request mode for observe requests.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[repr(u8)]
 pub enum ObserveMode {
@@ -37,9 +39,10 @@ pub enum ObserveMode {
     /// We get a notification for connect events.
     Notify,
     /// We get a request for connect events and can reject incoming connections.
-    Request,
+    Intercept,
 }
 
+/// Request mode for all data related requests.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[repr(u8)]
 pub enum RequestMode {
@@ -49,32 +52,39 @@ pub enum RequestMode {
     /// We get a notification for each request, but no transfer events.
     Notify,
     /// We get a request for each request, and can reject incoming requests, but no transfer events.
-    Request,
+    Intercept,
     /// We get a notification for each request as well as detailed transfer events.
     NotifyLog,
     /// We get a request for each request, and can reject incoming requests.
     /// We also get detailed transfer events.
-    RequestLog,
+    InterceptLog,
     /// This request type is completely disabled. All requests will be rejected.
+    ///
+    /// This means that requests of this kind will always be rejected, whereas
+    /// None means that we don't get any events, but requests will be processed normally.
     Disabled,
 }
 
+/// Throttling mode for requests that support throttling.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 #[repr(u8)]
 pub enum ThrottleMode {
     /// We don't get these kinds of events at all
     #[default]
     None,
     /// We call throttle to give the event handler a way to throttle requests
-    Throttle,
+    Intercept,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 pub enum AbortReason {
+    /// The request was aborted because a limit was exceeded. It is OK to try again later.
     RateLimited,
+    /// The request was aborted because the client does not have permission to perform the operation.
     Permission,
 }
 
+/// Errors that can occur when sending progress updates.
 #[derive(Debug, Snafu)]
 pub enum ProgressError {
     Limit,
@@ -141,6 +151,10 @@ impl From<irpc::channel::SendError> for ProgressError {
 pub type EventResult = Result<(), AbortReason>;
 pub type ClientResult = Result<(), ProgressError>;
 
+/// Event mask to configure which events are sent to the event handler.
+///
+/// This can also be used to completely disable certain request types. E.g.
+/// push requests are disabled by default, as they can write to the local store.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct EventMask {
     /// Connection event mask
@@ -180,12 +194,12 @@ impl EventMask {
     /// need to do it manually. Providing constants that have push enabled would
     /// risk misuse.
     pub const ALL_READONLY: Self = Self {
-        connected: ConnectMode::Request,
-        get: RequestMode::RequestLog,
-        get_many: RequestMode::RequestLog,
+        connected: ConnectMode::Intercept,
+        get: RequestMode::InterceptLog,
+        get_many: RequestMode::InterceptLog,
         push: RequestMode::Disabled,
-        throttle: ThrottleMode::Throttle,
-        observe: ObserveMode::Request,
+        throttle: ThrottleMode::Intercept,
+        observe: ObserveMode::Intercept,
     };
 }
 
@@ -239,7 +253,7 @@ impl RequestTracker {
         throttle: None,
     };
 
-    /// Transfer for index `index` started, size `size`
+    /// Transfer for index `index` started, size `size` in bytes.
     pub async fn transfer_started(&self, index: u64, hash: &Hash, size: u64) -> irpc::Result<()> {
         if let RequestUpdates::Active(tx) = &self.updates {
             tx.send(
@@ -333,7 +347,10 @@ impl EventSender {
             }
             while let Some(msg) = rx.recv().await {
                 match msg {
-                    ProviderMessage::ClientConnected(_) => todo!(),
+                    ProviderMessage::ClientConnected(msg) => {
+                        trace!("{:?}", msg.inner);
+                        msg.tx.send(Ok(())).await.ok();
+                    }
                     ProviderMessage::ClientConnectedNotify(msg) => {
                         trace!("{:?}", msg.inner);
                     }
@@ -395,7 +412,7 @@ impl EventSender {
             match self.mask.connected {
                 ConnectMode::None => {}
                 ConnectMode::Notify => client.notify(Notify(f())).await?,
-                ConnectMode::Request => client.rpc(f()).await??,
+                ConnectMode::Intercept => client.rpc(f()).await??,
             }
         };
         Ok(())
@@ -445,7 +462,7 @@ impl EventSender {
                         client.unwrap().notify_streaming(Notify(msg), 32).await?,
                     )
                 }
-                RequestMode::Request if client.is_some() => {
+                RequestMode::Intercept if client.is_some() => {
                     let msg = RequestReceived {
                         request: f(),
                         connection_id,
@@ -464,7 +481,7 @@ impl EventSender {
                     };
                     RequestUpdates::Active(client.unwrap().notify_streaming(Notify(msg), 32).await?)
                 }
-                RequestMode::RequestLog if client.is_some() => {
+                RequestMode::InterceptLog if client.is_some() => {
                     let msg = RequestReceived {
                         request: f(),
                         connection_id,
@@ -491,7 +508,7 @@ impl EventSender {
     ) -> RequestTracker {
         let throttle = match self.mask.throttle {
             ThrottleMode::None => None,
-            ThrottleMode::Throttle => self
+            ThrottleMode::Intercept => self
                 .inner
                 .clone()
                 .map(|client| (client, connection_id, request_id)),
@@ -515,38 +532,39 @@ pub enum ProviderProto {
     #[rpc(tx = NoSender)]
     ConnectionClosed(ConnectionClosed),
 
-    #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = oneshot::Sender<EventResult>)]
     /// A new get request was received from the provider.
+    #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = oneshot::Sender<EventResult>)]
     GetRequestReceived(RequestReceived<GetRequest>),
 
+    /// A new get request was received from the provider (notify variant).
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = NoSender)]
-    /// A new get request was received from the provider.
     GetRequestReceivedNotify(Notify<RequestReceived<GetRequest>>),
 
-    /// A new get request was received from the provider.
+    /// A new get many request was received from the provider.
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = oneshot::Sender<EventResult>)]
     GetManyRequestReceived(RequestReceived<GetManyRequest>),
 
-    /// A new get request was received from the provider.
+    /// A new get many request was received from the provider (notify variant).
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = NoSender)]
     GetManyRequestReceivedNotify(Notify<RequestReceived<GetManyRequest>>),
 
-    /// A new get request was received from the provider.
+    /// A new push request was received from the provider.
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = oneshot::Sender<EventResult>)]
     PushRequestReceived(RequestReceived<PushRequest>),
 
-    /// A new get request was received from the provider.
+    /// A new push request was received from the provider (notify variant).
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = NoSender)]
     PushRequestReceivedNotify(Notify<RequestReceived<PushRequest>>),
 
-    /// A new get request was received from the provider.
+    /// A new observe request was received from the provider.
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = oneshot::Sender<EventResult>)]
     ObserveRequestReceived(RequestReceived<ObserveRequest>),
 
-    /// A new get request was received from the provider.
+    /// A new observe request was received from the provider (notify variant).
     #[rpc(rx = mpsc::Receiver<RequestUpdate>, tx = NoSender)]
     ObserveRequestReceivedNotify(Notify<RequestReceived<ObserveRequest>>),
 
+    /// Request to throttle sending for a specific data request.
     #[rpc(tx = oneshot::Sender<EventResult>)]
     Throttle(Throttle),
 }