Upgrade Actix related Cargo dependencies (#141)

- upgrade 'actix' to 0.10 version
- upgrade 'actix-http' to 2.0 version
- upgrade 'actix-web' to 3.0 version
- upgrade 'actix-web-actors' to 3.0 version
- upgrade 'actix-codec' to 0.3 version
- upgrade 'actix-rt' to 1.1 version
- upgrade 'awc' to 2.0 version

Cargo.toml

@@ -28,10 +28,10 @@ members = [
 lto = "thin"
 
 [dependencies]
-actix = "0.9"
-actix-http = "1.0"
-actix-web = "2.0"
-actix-web-actors = "2.0"
+actix = "0.10"
+actix-http = "2.0"
+actix-web = "3.0"
+actix-web-actors = "3.0"
 async-trait = "0.1"
 bytes = "0.5"
 chrono = "0.4"
@@ -74,10 +74,9 @@ url = "2.1"
     features = ["tokio-comp"]
 
 [dev-dependencies]
-# TODO: Upgrade this dependency when `awc` 2.0 will be released.
-actix-codec = "0.2"
-actix-rt = "1.0"
-awc = "1.0"
+actix-codec = "0.3"
+actix-rt = "1.1"
+awc = "2.0"
 derive_builder = "0.9"
 function_name = "0.2"
 lazy_static = "1.4"

mock/control-api/Cargo.toml

@@ -10,11 +10,10 @@ repository = "https://github.com/instrumentisto/medea"
 publish = false
 
 [dependencies]
-actix = "0.9"
-actix-cors = "0.2"
-actix-rt = "1.0"
-actix-web = "2.0"
-actix-web-actors = "2.0"
+actix = "0.10"
+actix-cors = "0.3"
+actix-web = "3.0"
+actix-web-actors = "3.0"
 clap = "2.33"
 derive_more = "0.99"
 dotenv = "0.15"

mock/control-api/src/main.rs

@@ -19,7 +19,7 @@ use clap::{
 use slog::{o, Drain};
 use slog_scope::GlobalLoggerGuard;
 
-#[actix_rt::main]
+#[actix_web::main]
 async fn main() {
     dotenv::dotenv().ok();
 

src/api/client/server.rs

@@ -2,7 +2,7 @@
 
 use std::io;
 
-use actix::{Actor, Addr, Handler};
+use actix::{Actor, Addr, Handler, ResponseFuture};
 use actix_web::{
     dev::Server as ActixServer,
     middleware,
@@ -26,7 +26,6 @@ use crate::{
     log::prelude::*,
     shutdown::ShutdownGracefully,
     signalling::room_repo::RoomRepository,
-    utils::ResponseAnyFuture,
 };
 
 /// Parameters of new WebSocket connection creation HTTP request.
@@ -49,7 +48,7 @@ async fn ws_index(
     info: Path<RequestParams>,
     state: Data<Context>,
     payload: Payload,
-) -> Result<HttpResponse, actix_web::Error> {
+) -> actix_web::Result<HttpResponse> {
     debug!("Request params: {:?}", info);
     let RequestParams {
         room_id,
@@ -64,7 +63,8 @@ async fn ws_index(
                     member_id: member_id.clone(),
                     credentials,
                 })
-                .await?;
+                .await
+                .map_err(|_| HttpResponse::InternalServerError())?;
             match auth_result {
                 Ok(settings) => ws::start(
                     WsSession::new(
@@ -78,14 +78,14 @@ async fn ws_index(
                     payload,
                 ),
                 Err(AuthorizationError::MemberNotExists) => {
-                    Ok(HttpResponse::NotFound().into())
+                    Err(HttpResponse::NotFound().into())
                 }
                 Err(AuthorizationError::InvalidCredentials) => {
-                    Ok(HttpResponse::Forbidden().into())
+                    Err(HttpResponse::Forbidden().into())
                 }
             }
         }
-        None => Ok(HttpResponse::NotFound().into()),
+        None => Err(HttpResponse::NotFound().into()),
     }
 }
 
@@ -147,15 +147,15 @@ impl Actor for Server {
 }
 
 impl Handler<ShutdownGracefully> for Server {
-    type Result = ResponseAnyFuture<()>;
+    type Result = ResponseFuture<()>;
 
     fn handle(
         &mut self,
         _: ShutdownGracefully,
         _: &mut Self::Context,
     ) -> Self::Result {
         info!("Server received ShutdownGracefully message so shutting down");
-        ResponseAnyFuture(self.0.stop(true).boxed_local())
+        self.0.stop(true).boxed_local()
     }
 }
 

src/api/client/session.rs

@@ -702,12 +702,8 @@ mod test {
             ))
             .await
             .unwrap();
-        // TODO: Change to `FirstText` when upgrade actix-web to v3.
-        //       Fixed in https://github.com/actix/actix-web/pull/1465.
         client
-            .send(Message::Continuation(Item::FirstBinary(
-                command.slice(0..10),
-            )))
+            .send(Message::Continuation(Item::FirstText(command.slice(0..10))))
             .await
             .unwrap();
         client
@@ -716,12 +712,8 @@ mod test {
             )))
             .await
             .unwrap();
-        // TODO: Change to `FirstText` when upgrade actix-web to v3.
-        //       Fixed in https://github.com/actix/actix-web/pull/1465.
         client
-            .send(Message::Continuation(Item::FirstBinary(
-                command.slice(0..10),
-            )))
+            .send(Message::Continuation(Item::FirstText(command.slice(0..10))))
             .await
             .unwrap();
         client

src/api/control/grpc/server.rs

@@ -76,6 +76,7 @@ struct ControlApiService {
     public_url: String,
 }
 
+// TODO: This can be easily refactored to async fn's.
 impl ControlApiService {
     /// Implementation of `Create` method for [`Room`].
     fn create_room(

src/shutdown.rs

@@ -10,8 +10,7 @@ use std::{
 use actix::AsyncContext;
 use actix::{
     prelude::{Actor, Context},
-    Addr, ContextFutureSpawner as _, Handler, Message, Recipient, System,
-    WrapFuture as _,
+    Addr, Handler, Message, Recipient, ResponseFuture, System,
 };
 use derive_more::Display;
 use failure::Fail;
@@ -104,14 +103,18 @@ impl Actor for GracefulShutdown {
 struct OsSignal(i32);
 
 impl Handler<OsSignal> for GracefulShutdown {
-    type Result = ();
+    type Result = ResponseFuture<()>;
 
-    fn handle(&mut self, sig: OsSignal, ctx: &mut Context<Self>) {
+    fn handle(
+        &mut self,
+        sig: OsSignal,
+        _: &mut Context<Self>,
+    ) -> ResponseFuture<()> {
         info!("OS signal '{}' received", sig.0);
 
         match self.state {
             State::ShuttingDown => {
-                return;
+                return future::ready(()).boxed_local();
             }
             State::Listening => {
                 self.state = State::ShuttingDown;
@@ -122,7 +125,7 @@ impl Handler<OsSignal> for GracefulShutdown {
 
         if self.subs.is_empty() {
             System::current().stop();
-            return;
+            return future::ready(()).boxed_local();
         }
 
         let subs = mem::replace(&mut self.subs, BTreeMap::new());
@@ -156,8 +159,7 @@ impl Handler<OsSignal> for GracefulShutdown {
             }
             System::current().stop()
         }
-        .into_actor(self)
-        .wait(ctx);
+        .boxed_local()
     }
 }
 

src/signalling/participants.rs

@@ -37,7 +37,7 @@ use crate::{
             member::MemberError,
             parse_members, Member, MembersLoadError,
         },
-        room::{ActFuture, RoomError},
+        room::RoomError,
         Room,
     },
     AppContext,
@@ -208,14 +208,14 @@ impl ParticipantService {
         ctx: &mut Context<Room>,
         member_id: MemberId,
         conn: Box<dyn RpcConnection>,
-    ) -> ActFuture<Result<Member, ParticipantServiceErr>> {
+    ) -> LocalBoxFuture<'static, Result<Member, ParticipantServiceErr>> {
         let member = match self.get_member_by_id(&member_id) {
             None => {
-                return Box::new(wrap_future(future::err(
+                return Box::pin(future::err(
                     ParticipantServiceErr::ParticipantNotFound(
                         self.get_fid_to_member(member_id),
                     ),
-                )));
+                ));
             }
             Some(member) => member,
         };
@@ -231,14 +231,14 @@ impl ParticipantService {
                 ctx.cancel_future(handler);
             }
             self.insert_connection(member_id, conn);
-            Box::new(wrap_future(
+            Box::pin(
                 connection
                     .close(CloseDescription::new(CloseReason::Reconnected))
                     .map(move |_| Ok(member)),
-            ))
+            )
         } else {
             self.insert_connection(member_id, conn);
-            Box::new(wrap_future(future::ok(member)))
+            Box::pin(future::ok(member))
         }
     }
 

src/signalling/room/dynamic_api.rs

@@ -228,27 +228,28 @@ impl Room {
 
         member.insert_sink(sink);
 
-        let fut = fut::ready(()).map(move |_, this: &mut Self, ctx| {
-            let member_id = member.id();
-            if this.members.member_has_connection(&member_id) {
-                ctx.spawn(this.init_member_connections(&member).map(
-                    move |res, this, ctx| {
-                        if let Err(e) = res {
-                            error!(
-                                "Failed to interconnect Members, because {}. \
-                                 Connection with Member [id = {}, room_id: \
-                                 {}] will be stopped.",
-                                e, member_id, this.id,
-                            );
-                            this.members
-                                .close_member_connection(&member_id, ctx);
-                        }
-                    },
-                ));
-            }
-            Ok(())
-        });
-        Ok(Box::new(fut))
+        Ok(Box::pin(fut::ready(()).map(
+            move |_, this: &mut Self, ctx| {
+                let member_id = member.id();
+                if this.members.member_has_connection(&member_id) {
+                    ctx.spawn(this.init_member_connections(&member).map(
+                        move |res, this, ctx| {
+                            if let Err(e) = res {
+                                error!(
+                                    "Failed to interconnect Members, because \
+                                     {}. Connection with Member [id = {}, \
+                                     room_id: {}] will be stopped.",
+                                    e, member_id, this.id,
+                                );
+                                this.members
+                                    .close_member_connection(&member_id, ctx);
+                            }
+                        },
+                    ));
+                }
+                Ok(())
+            },
+        )))
     }
 
     /// Removes [`Peer`]s and call [`Room::member_peers_removed`] for every
@@ -435,8 +436,8 @@ impl Handler<CreateEndpoint> for Room {
                     msg.endpoint_id.into(),
                     endpoint,
                 ) {
-                    Ok(fut) => Box::new(fut),
-                    Err(e) => Box::new(fut::err(e)),
+                    Ok(fut) => Box::pin(fut),
+                    Err(e) => Box::pin(fut::err(e)),
                 }
             }
             EndpointSpec::WebRtcPublish(endpoint) => {
@@ -445,9 +446,9 @@ impl Handler<CreateEndpoint> for Room {
                     msg.endpoint_id.into(),
                     &endpoint,
                 ) {
-                    Box::new(fut::err(e))
+                    Box::pin(fut::err(e))
                 } else {
-                    Box::new(fut::ok(()))
+                    Box::pin(fut::ok(()))
                 }
             }
         }