[DRAFT] Upgrade to actix-web 4.0

Cargo.toml

@@ -24,14 +24,11 @@ memcached = ["r2d2-memcache", "backoff"]
 
 [dependencies]
 log = "0.4.11"
-actix-web = {version = "3.3.2"}
-actix-http = {version = "2.2.0", features=["actors"]}
-actix = "0.10"
+actix = "0.13.0"
+actix-web = {version = "4.2.1"}
 futures = "0.3.8"
 failure = "0.1.8"
-
 dashmap = {version = "4.0.1", optional = true}
-
 redis_rs = {version = "0.15.1", optional = true, package= "redis"}
 backoff = {version = "0.2.1", optional = true}
 r2d2-memcache = { version = "0.6", optional = true }

src/errors.rs

@@ -1,8 +1,10 @@
 //! Errors that can occur during middleware processing stage
-use actix_web::error::Error as AWError;
-use actix_web::web::HttpResponse;
+use actix_web::body::BoxBody;
+use actix_web::error::ResponseError;
+use actix_web::http::StatusCode;
+use actix_web::HttpResponse;
 use failure::{self, Fail};
-use log::*;
+use std::time::Duration;
 
 /// Custom error type. Useful for logging and debugging different kinds of errors.
 /// This type can be converted to Actix Error, which defaults to
@@ -29,11 +31,33 @@ pub enum ARError {
     /// Identifier error
     #[fail(display = "client identification failed")]
     IdentificationError,
+
+    /// Rate limited error
+    #[fail(display = "rate limit failed")]
+    RateLimitError {
+        max_requests: usize,
+        c: usize,
+        reset: Duration,
+    },
 }
 
-impl From<ARError> for AWError {
-    fn from(err: ARError) -> AWError {
-        error!("{}", &err);
-        HttpResponse::InternalServerError().into()
+impl ResponseError for ARError {
+    fn status_code(&self) -> StatusCode {
+        StatusCode::INTERNAL_SERVER_ERROR
+    }
+
+    fn error_response(&self) -> HttpResponse<BoxBody> {
+        match *self {
+            Self::RateLimitError {
+                max_requests,
+                c,
+                reset,
+            } => HttpResponse::TooManyRequests()
+                .insert_header(("x-ratelimit-limit", max_requests.to_string()))
+                .insert_header(("x-ratelimit-remaining", c.to_string()))
+                .insert_header(("x-ratelimit-reset", reset.as_secs().to_string()))
+                .finish(),
+            _ => HttpResponse::InternalServerError().finish(),
+        }
     }
 }

src/lib.rs

@@ -190,12 +190,12 @@ pub mod stores;
 use errors::ARError;
 pub use middleware::RateLimiter;
 
+#[cfg(feature = "memcached")]
+pub use stores::memcached::{MemcacheStore, MemcacheStoreActor};
 #[cfg(feature = "memory")]
 pub use stores::memory::{MemoryStore, MemoryStoreActor};
 #[cfg(feature = "redis-store")]
 pub use stores::redis::{RedisStore, RedisStoreActor};
-#[cfg(feature = "memcached")]
-pub use stores::memcached::{MemcacheStore, MemcacheStoreActor};
 
 use std::future::Future;
 use std::marker::Send;
@@ -248,9 +248,9 @@ where
     A: Actor,
     M: actix::Message<Result = ActorResponse>,
 {
-    fn handle<R: ResponseChannel<M>>(self, _: &mut A::Context, tx: Option<R>) {
+    fn handle(self, _: &mut A::Context, tx: Option<OneshotSender<M::Result>>) {
         if let Some(tx) = tx {
-            tx.send(self);
+            tx.send(self).ok();
         }
     }
 }

src/middleware.rs

@@ -2,9 +2,8 @@
 use actix::dev::*;
 use actix_web::{
     dev::{Service, ServiceRequest, ServiceResponse, Transform},
-    error::Error as AWError,
-    http::{HeaderName, HeaderValue},
-    HttpResponse,
+    error::{Error as AWError, ErrorInternalServerError},
+    http::header::{HeaderName, HeaderValue},
 };
 use futures::future::{ok, Ready};
 use log::*;
@@ -50,7 +49,7 @@ where
     interval: Duration,
     max_requests: usize,
     store: Addr<T>,
-    identifier: Rc<Box<dyn Fn(&ServiceRequest) -> Result<String, ARError>>>,
+    identifier: Rc<Box<dyn Fn(&ServiceRequest) -> Result<String, AWError>>>,
 }
 
 impl<T> RateLimiter<T>
@@ -63,8 +62,9 @@ where
         let identifier = |req: &ServiceRequest| {
             let connection_info = req.connection_info();
             let ip = connection_info
-                .remote_addr()
-                .ok_or(ARError::IdentificationError)?;
+                .peer_addr()
+                .ok_or(ARError::IdentificationError)
+                .map_err(ErrorInternalServerError)?;
             Ok(String::from(ip))
         };
         RateLimiter {
@@ -88,7 +88,7 @@ where
     }
 
     /// Function to get the identifier for the client request
-    pub fn with_identifier<F: Fn(&ServiceRequest) -> Result<String, ARError> + 'static>(
+    pub fn with_identifier<F: Fn(&ServiceRequest) -> Result<String, AWError> + 'static>(
         mut self,
         identifier: F,
     ) -> Self {
@@ -97,15 +97,14 @@ where
     }
 }
 
-impl<T, S, B> Transform<S> for RateLimiter<T>
+impl<T, S, B> Transform<S, ServiceRequest> for RateLimiter<T>
 where
     T: Handler<ActorMessage> + Send + Sync + 'static,
     T::Context: ToEnvelope<T, ActorMessage>,
-    S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = AWError> + 'static,
+    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = AWError> + 'static,
     S::Future: 'static,
     B: 'static,
 {
-    type Request = ServiceRequest;
     type Response = ServiceResponse<B>;
     type Error = S::Error;
     type InitError = ();
@@ -134,85 +133,91 @@ where
     // Exists here for the sole purpose of knowing the max_requests and interval from RateLimiter
     max_requests: usize,
     interval: u64,
-    identifier: Rc<Box<dyn Fn(&ServiceRequest) -> Result<String, ARError> + 'static>>,
+    identifier: Rc<Box<dyn Fn(&ServiceRequest) -> Result<String, AWError> + 'static>>,
 }
 
-impl<T, S, B> Service for RateLimitMiddleware<S, T>
+impl<T, S, B> Service<ServiceRequest> for RateLimitMiddleware<S, T>
 where
     T: Handler<ActorMessage> + 'static,
-    S: Service<Request = ServiceRequest, Response = ServiceResponse<B>, Error = AWError> + 'static,
+    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = AWError> + 'static,
     S::Future: 'static,
     B: 'static,
     T::Context: ToEnvelope<T, ActorMessage>,
 {
-    type Request = ServiceRequest;
     type Response = ServiceResponse<B>;
     type Error = S::Error;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
 
-    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+    fn poll_ready(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.service.borrow_mut().poll_ready(cx)
     }
 
-    fn call(&mut self, req: ServiceRequest) -> Self::Future {
+    fn call(&self, req: ServiceRequest) -> Self::Future {
         let store = self.store.clone();
-        let mut srv = self.service.clone();
+        let srv = self.service.clone();
         let max_requests = self.max_requests;
         let interval = Duration::from_secs(self.interval);
         let identifier = self.identifier.clone();
         Box::pin(async move {
-            let identifier: String = (identifier)(&req)?;
+            let identifier: String = (identifier)(&req).map_err(ErrorInternalServerError)?;
             let remaining: ActorResponse = store
                 .send(ActorMessage::Get(String::from(&identifier)))
-                .await?;
+                .await
+                .map_err(ErrorInternalServerError)?;
             match remaining {
                 ActorResponse::Get(opt) => {
                     let opt = opt.await?;
                     if let Some(c) = opt {
                         // Existing entry in store
                         let expiry = store
                             .send(ActorMessage::Expire(String::from(&identifier)))
-                            .await?;
+                            .await
+                            .map_err(ErrorInternalServerError)?;
                         let reset: Duration = match expiry {
                             ActorResponse::Expire(dur) => dur.await?,
                             _ => unreachable!(),
                         };
                         if c == 0 {
                             info!("Limit exceeded for client: {}", &identifier);
-                            let mut response = HttpResponse::TooManyRequests();
-                            // let mut response = (error_callback)(&mut response);
-                            response.set_header("x-ratelimit-limit", max_requests.to_string());
-                            response.set_header("x-ratelimit-remaining", c.to_string());
-                            response.set_header("x-ratelimit-reset", reset.as_secs().to_string());
-                            Err(response.into())
+
+                            Err(ARError::RateLimitError {
+                                max_requests,
+                                c,
+                                reset,
+                            }
+                            .into())
                         } else {
                             // Decrement value
                             let res: ActorResponse = store
                                 .send(ActorMessage::Update {
                                     key: identifier,
                                     value: 1,
                                 })
-                                .await?;
+                                .await
+                                .map_err(ErrorInternalServerError)?;
                             let updated_value: usize = match res {
-                                ActorResponse::Update(c) => c.await?,
+                                ActorResponse::Update(c) => {
+                                    c.await.map_err(ErrorInternalServerError)?
+                                }
                                 _ => unreachable!(),
                             };
                             // Execute the request
                             let fut = srv.call(req);
-                            let mut res = fut.await?;
+                            let mut res = fut.await.map_err(ErrorInternalServerError)?;
                             let headers = res.headers_mut();
                             // Safe unwraps, since usize is always convertible to string
                             headers.insert(
                                 HeaderName::from_static("x-ratelimit-limit"),
-                                HeaderValue::from_str(max_requests.to_string().as_str())?,
+                                HeaderValue::from_str(max_requests.to_string().as_str()).unwrap(),
                             );
                             headers.insert(
                                 HeaderName::from_static("x-ratelimit-remaining"),
-                                HeaderValue::from_str(updated_value.to_string().as_str())?,
+                                HeaderValue::from_str(updated_value.to_string().as_str()).unwrap(),
                             );
                             headers.insert(
                                 HeaderName::from_static("x-ratelimit-reset"),
-                                HeaderValue::from_str(reset.as_secs().to_string().as_str())?,
+                                HeaderValue::from_str(reset.as_secs().to_string().as_str())
+                                    .unwrap(),
                             );
                             Ok(res)
                         }
@@ -225,13 +230,16 @@ where
                                 value: current_value,
                                 expiry: interval,
                             })
-                            .await?;
+                            .await
+                            .map_err(ErrorInternalServerError)?;
                         match res {
-                            ActorResponse::Set(c) => c.await?,
+                            ActorResponse::Set(c) => {
+                                c.await.map_err(|err| ErrorInternalServerError(err))?
+                            }
                             _ => unreachable!(),
                         }
                         let fut = srv.call(req);
-                        let mut res = fut.await?;
+                        let mut res = fut.await.map_err(|err| ErrorInternalServerError(err))?;
                         let headers = res.headers_mut();
                         // Safe unwraps, since usize is always convertible to string
                         headers.insert(

src/stores/memcached.rs

@@ -204,7 +204,7 @@ impl Handler<ActorMessage> for MemcacheStoreActor {
                             Ok(c) => match c {
                                 Some(v) => Ok(Some(v as usize)),
                                 None => Ok(None),
-                            }
+                            },
                             Err(e) => Err(ARError::ReadWriteError(format!("{:?}", &e))),
                         }
                     })),
@@ -247,10 +247,6 @@ impl Handler<ActorMessage> for MemcacheStoreActor {
     }
 }
 
-
-
-
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -315,7 +311,7 @@ mod tests {
             _ => panic!("Shouldn't happen!"),
         };
     }
-    
+
     #[actix_rt::test]
     async fn test_expiry() {
         init();
@@ -357,6 +353,5 @@ mod tests {
             },
             _ => panic!("Shouldn't happen!"),
         };
-
     }
 }

tests/version-numbers.rs

@@ -7,4 +7,3 @@ fn test_readme_deps() {
 fn test_html_root_url() {
     version_sync::assert_html_root_url_updated!("src/lib.rs");
 }
-