Rust 异步Web 框架 Actix-web（包括 Actor 框架 Actix）

[dependencies]
actix-web = "2.0"

[dependencies]
actix-web = { git = "https://github.com/actix/actix-web" }

git clone https://github.com/actix/examples
cd examples/basics
cargo run

[dependencies]
actix-web = "2.0"
actix-rt = "1.0"

use actix_web::{web, App, HttpResponse, HttpServer, Responder};

// curl http://localhost:8088/
async fn index() -> impl Responder {
    HttpResponse::Ok().body("Hello world!")
}

// curl http://localhost:8088/again
async fn index2() -> impl Responder {
    HttpResponse::Ok().body("Hello world again!")
}

use actix_web::get;

// curl http://localhost:8088/hello
// 使用宏解析
#[get("/hello")]
async fn index3() -> impl Responder {
    HttpResponse::Ok().body("Hey there!")
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .route("/", web::get().to(index))
            .route("/again", web::get().to(index2))
            .service(index3)
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix_web::{web, App, HttpResponse, HttpServer, Responder};

// curl http://localhost:8088/
// curl http://localhost:8088/app/index.html
async fn index() -> impl Responder {
    HttpResponse::Ok().body("Hello world!")
}

use actix_web::get;

// curl http://localhost:8088/hello
// curl http://localhost:8088/app/hello
// 使用宏解析
#[get("/hello")]
async fn index3() -> impl Responder {
    HttpResponse::Ok().body("Hey there!")
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .service(
                web::scope("/app")
                    .route("/index.html", web::get().to(index))
                    .service(index3)
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix_web::{web, App, HttpResponse, HttpServer};

#[actix_rt::main]
async fn main() {
    HttpServer::new(|| {
        App::new().route("/", web::get().to(|| HttpResponse::Ok()))
    })
    .workers(4); // <- Start 4 workers
}

serde = "1.0"
serde_json = "1.0"
futures = "0.3"

use actix_web::{web, App, HttpResponse, HttpServer, Responder, Error, HttpRequest, Either};

// curl http://localhost:8088/responder/str
async fn responder_str() -> &'static str {
    "responder_str"
}

// curl http://localhost:8088/responder/string
async fn responder_string() -> String {
    "responder_string".to_owned()
}

// curl http://localhost:8088/responder/impl_responder
async fn responder_impl_responder() -> impl Responder{
    web::Bytes::from_static(b"responder_string")
}

use serde::Serialize;
use futures::future::{ready, Ready};

// 自定义 Response
#[derive(Serialize)]
struct ResponseWrapper<T> {
    code: i32,
    msg: String,
    data: Option<T>,
}

// Responder
impl <T> Responder for ResponseWrapper<T> where T: Serialize {
    type Error = Error;
    type Future = Ready<Result<HttpResponse, Error>>;

    fn respond_to(self, _req: &HttpRequest) -> Self::Future {
        let body = serde_json::to_string(&self).unwrap();

        // Create response and set content type
        ready(Ok(HttpResponse::Ok()
            .content_type("application/json")
            .body(body)))
    }
}

// curl http://localhost:8088/responder/custom_responder
async fn responder_custom_responder() -> impl Responder {
    ResponseWrapper {
        code: 0,
        msg: "success".to_string(),
        data: Some("custom_responder".to_string()) }
}

use futures::stream::once;
use futures::future::ok;

// curl http://localhost:8088/responder/stream
async fn responder_stream_responder() -> HttpResponse {
    let body = once(ok::<_, Error>(web::Bytes::from_static(b"test")));

    HttpResponse::Ok()
        .content_type("application/json")
        .streaming(body)
}

type RegisterResult = Either<HttpResponse, Result<&'static str, Error>>;

// curl http://localhost:8088/responder/either
async fn responder_either_responder() -> RegisterResult {
    Either::A(HttpResponse::BadRequest().body("Bad data"))
    // Either::B(Ok("Hello!"))
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {

    let c = web::Data::new(AppStateWithCounter {
        counter: Mutex::new(0),
    });
    HttpServer::new(move || {
        App::new()
            .service(
                web::scope("/responder")
                    .route("/str", web::get().to(responder_str))
                    .route("/string", web::get().to(responder_string))
                    .route("/impl_responder", web::get().to(responder_impl_responder))
                    .route("/custom_responder", web::get().to(responder_custom_responder))
                    .route("/stream", web::get().to(responder_stream_responder))
                    .route("/either", web::get().to(responder_either_responder))
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

trait CallFnWithTuple<T, R> {
    fn call_with_tuple(&self, param: T) -> R;
}

impl <Func, A, R> CallFnWithTuple<(A,), R> for Func where Func: Fn(A,) -> R {
    fn call_with_tuple(&self, param: (A,)) -> R {
        (self)(param.0,)
     }

}

impl <Func, A, B, R> CallFnWithTuple<(A, B,), R> for Func where Func: Fn(A, B,) -> R {
    fn call_with_tuple(&self, param: (A, B,)) -> R {
        (self)(param.0, param.1)
     }

}

fn proxy<T, R>(f: impl CallFnWithTuple<T, R>, p: T) -> R {
    f.call_with_tuple(p)
}

fn test_1(a: i32) -> i32 {
    a + 1
}
fn test_2(a: i32, b: i32) -> i32 {
    a + b
}

fn main() {
    println!("{}", proxy(test_1, (1,)));
    println!("{}", proxy(test_2, (1,2)));
    println!("{}", test_2.call_with_tuple((1,2)));
}

/// Request 提取器
///
/// 实现了该特质的类型可以作为 请求处理器 的参数使用
pub trait FromRequest: Sized {
    /// 发送错误时，返回的类型
    type Error: Into<Error>;

    /// 将 Self 转换为一个 Future
    type Future: Future<Output = Result<Self, Self::Error>>;

    /// 该提取器的配置
    type Config: Default + 'static;

    /// 将 request 转换为 Self
    fn from_request(req: &HttpRequest, payload: &mut Payload) -> Self::Future;

    /// 将 request 转换为 Self
    ///
    /// This method uses `Payload::None` as payload stream.
    fn extract(req: &HttpRequest) -> Self::Future {
        Self::from_request(req, &mut Payload::None)
    }

    /// 创建一个Config实例
    fn configure<F>(f: F) -> Self::Config
    where
        F: FnOnce(Self::Config) -> Self::Config,
    {
        f(Self::Config::default())
    }
}

use actix_web::{error, web, App, FromRequest, HttpResponse, HttpServer, Responder, Error, HttpRequest, Either};

use serde::Deserialize;

// 提取器 extractors
#[derive(Deserialize, Debug)]
struct QueryInfo {
    username: String,
}

// curl http://localhost:8088/extractor/multiple/p1/p2?username=xiaoming
async fn extractor_multiple(p: web::Path<(String, String)>, q: web::Query<QueryInfo>) -> String {
    format!("p={:?}, q={:?}", p, q)
}

#[derive(Deserialize, Debug)]
struct PathInfo {
    user_id: u32,
    friend: String,
}

// curl http://localhost:8088/extractor/path/123/friend_name
async fn extractor_path(p: web::Path<PathInfo>) -> String {
    format!("path-param={:?}", p)
}

// curl http://localhost:8088/extractor/manual_path/123/friend_name
async fn extractor_manual_path(req: HttpRequest) -> String {
    let friend: String =
        req.match_info().get("friend").unwrap().parse().unwrap();
    let user_id: i32 = req.match_info().query("user_id").parse().unwrap();

    format!("user_id={}, friend={}", user_id, friend)
}

// curl http://localhost:8088/extractor/query?username=xiaoming
async fn extractor_query(info: web::Query<QueryInfo>) -> String {
    format!("{:?}", info)
}

#[derive(Deserialize, Debug)]
struct JsonInfo {
    username: String,
}

// curl -i -H 'Content-Type: application/json' -d '{"username": "xiaoming"}' -X POST http://localhost:8088/extractor/json
// curl -i -H 'Content-Type: application/json' -d '{"username": 1}' -X POST http://localhost:8088/extractor/json
async fn extractor_json(info: web::Json<JsonInfo>) -> String {
    format!("{:?}", info)
}

#[derive(Deserialize, Debug)]
struct FormData {
    username: String,
}

/// 使用serde提取表单数据
/// 仅当内容类型为*x-www-form-urlencoded*时，才会调用此处理程序
/// 并且请求的内容可以反序列化为FormData结构
// curl -i -H 'Content-Type: application/x-www-form-urlencoded' -d 'username=xiaoming' -X POST http://localhost:8088/extractor/form
async fn extractor_form(form: web::Form<FormData>) -> String {
    format!("{:?}", form)
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(move || {
        App::new()
            // 配置 Json Extractor
            .app_data(web::Json::<JsonInfo>::configure(|cfg| {
                    cfg.limit(4096).error_handler(|err, _req| {
                        error::InternalError::from_response(
                            err,
                            HttpResponse::Conflict().finish(),
                        )
                        .into()
                    })
                }))
            .service(
                web::scope("/extractor")
                    .route("/multiple/{p1}/{p2}", web::get().to(extractor_multiple))
                    .route("/path/{user_id}/{friend}", web::get().to(extractor_path))
                    .route("/manual_path/{user_id}/{friend}", web::get().to(extractor_manual_path))
                    .route("/query", web::get().to(extractor_query))
                    .route("/json", web::post().to(extractor_json))
                    .route("/form", web::post().to(extractor_form))
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

impl<T, E> Responder for Result<T, E>
where
    T: Responder,
    E: Into<Error>,
{

impl<T: ResponseError + 'static> From<T> for Error {
    fn from(err: T) -> Error {
        Error {
            cause: Box::new(err),
        }
    }
}

use actix_web::{error, web, http, App, FromRequest, HttpResponse, HttpServer, Responder, Error, HttpRequest, Either, Result};

use failure::Fail;

#[derive(Fail, Debug)]
#[fail(display = "my error")]
struct MyError {
    name: &'static str,
}

impl error::ResponseError for MyError {}

// curl -i http://localhost:8088/error/custom
async fn error_custom() -> Result<&'static str, MyError> {
    Err(MyError { name: "test" })
}

#[derive(Fail, Debug)]
#[allow(dead_code)]
enum MyErrorEnum {
    #[fail(display = "internal error")]
    InternalError,
    #[fail(display = "bad request")]
    BadClientData,
    #[fail(display = "timeout")]
    Timeout,
}

use actix_http::ResponseBuilder;

impl error::ResponseError for MyErrorEnum {
    fn error_response(&self) -> HttpResponse {
        ResponseBuilder::new(self.status_code())
            .set_header(http::header::CONTENT_TYPE, "text/html; charset=utf-8")
            .body(self.to_string())
    }

    fn status_code(&self) -> http::StatusCode {
        match *self {
            MyErrorEnum::InternalError => http::StatusCode::INTERNAL_SERVER_ERROR,
            MyErrorEnum::BadClientData => http::StatusCode::BAD_REQUEST,
            MyErrorEnum::Timeout => http::StatusCode::GATEWAY_TIMEOUT,
        }
    }
}
// curl -i http://localhost:8088/error/enum
async fn error_enum() -> Result<&'static str, MyErrorEnum> {
    Err(MyErrorEnum::BadClientData)
}

// curl -i http://localhost:8088/error/helper
async fn error_helper() -> Result<&'static str> {
    let result: Result<&'static str, MyError> = Err(MyError { name: "test error" });

    Ok(result.map_err(|e| error::ErrorBadRequest(e.name))?)
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(move || {
        App::new()
            .service(
                web::scope("/error")
                    .route("/custom", web::get().to(error_custom))
                    .route("/enum", web::get().to(error_enum))
                    .route("/helper", web::get().to(error_helper))
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

pub trait Guard {
    fn check(&self, request: &RequestHead) -> bool;
}

App::new()
    .service(web::resource("/").route(web::get().to(index)))
    .default_service(
        web::route()
            .guard(guard::Not(guard::Get()))
            .to(|| HttpResponse::MethodNotAllowed()),
    )

use actix_web::{web, Error, HttpResponse};
use futures::StreamExt;

async fn index(mut body: web::Payload) -> Result<HttpResponse, Error> {
    let mut bytes = web::BytesMut::new();
    while let Some(item) = body.next().await {
        bytes.extend_from_slice(&item?);
    }

    println!("Chunk: {:?}", bytes);
    Ok(HttpResponse::Ok().finish())
}

use actix_web::{error, web, http, App, FromRequest, HttpResponse, HttpServer, Responder, Error, HttpRequest, Either, Result};

use serde::{Serialize, Deserialize};
use futures::future::{ready, Ready};

// 自定义 Response
#[derive(Serialize, Deserialize)]
struct ResponseWrapper<T> {
    code: i32,
    msg: String,
    data: Option<T>,
}

// Responder
impl <T> Responder for ResponseWrapper<T> where T: Serialize {
    type Error = Error;
    type Future = Ready<Result<HttpResponse, Error>>;

    fn respond_to(self, _req: &HttpRequest) -> Self::Future {
        let body = serde_json::to_string(&self).unwrap();

        // Create response and set content type
        ready(Ok(HttpResponse::Ok()
            .content_type("application/json")
            .body(body)))
    }
}

// curl http://localhost:8088/responder/custom_responder
async fn responder_custom_responder() -> impl Responder {
    ResponseWrapper {
        code: 0,
        msg: "success".to_string(),
        data: Some("custom_responder".to_string()) }
}


#[cfg(test)]
mod tests {
    use super::*;
    use actix_web::test;

    // 集成测试
    #[actix_rt::test]
    async fn test_index_get() {
        let mut app = test::init_service(App::new().route("/", web::get().to(index))).await;
        let req = test::TestRequest::with_header("content-type", "text/plain").to_request();
        let resp = test::call_service(&mut app, req).await;
        assert!(resp.status().is_success());
    }

    #[actix_rt::test]
    async fn test_index_post() {
        let mut app = test::init_service(App::new().route("/", web::get().to(index))).await;
        let req = test::TestRequest::post().uri("/").to_request();
        let resp = test::call_service(&mut app, req).await;
        assert!(resp.status().is_client_error());
    }

    #[actix_rt::test]
    async fn test_json_response() {
        let mut app = test::init_service(
            App::new()
                .data(AppState { app_name: "app_name".to_string() })
                .route("/responder/custom_responder", web::get().to(responder_custom_responder)),
        ).await;
        let req = test::TestRequest::get().uri("/responder/custom_responder").to_request();
        let resp: ResponseWrapper<String> = test::read_response_json(&mut app, req).await;

        assert_eq!(resp.code, 0);
    }
}

use std::task::Poll;
use bytes::Bytes;
use futures::stream::poll_fn;

use actix_web::http::{ContentEncoding, StatusCode};
use actix_web::{web, http, App, Error, HttpRequest, HttpResponse};

async fn sse(_req: HttpRequest) -> HttpResponse {
    let mut counter: usize = 5;

    // yields `data: N` where N in [5; 1]
    let server_events = poll_fn(move |_cx| -> Poll<Option<Result<Bytes, Error>>> {
        if counter == 0 {
            return Poll::Ready(None);
        }
        let payload = format!("data: {}\n\n", counter);
        counter -= 1;
        Poll::Ready(Some(Ok(Bytes::from(payload))))
    });

    HttpResponse::build(StatusCode::OK)
        .set_header(http::header::CONTENT_TYPE, "text/event-stream")
        .set_header(
            http::header::CONTENT_ENCODING,
            ContentEncoding::Identity.as_str(),
        )
        .streaming(server_events)
}

pub fn main() {
    App::new().route("/", web::get().to(sse));
}

#[cfg(test)]
mod tests {
    use super::*;
    use actix_rt;

    use futures_util::stream::StreamExt;
    use futures_util::stream::TryStreamExt;

    use actix_web::{test, web, App};

    #[actix_rt::test]
    async fn test_stream() {
        let mut app = test::init_service(App::new().route("/", web::get().to(sse))).await;
        let req = test::TestRequest::get().to_request();

        let mut resp = test::call_service(&mut app, req).await;
        assert!(resp.status().is_success());

        // first chunk
        let (bytes, mut resp) = resp.take_body().into_future().await;
        assert_eq!(bytes.unwrap().unwrap(), Bytes::from_static(b"data: 5\n\n"));

        // second chunk
        let (bytes, mut resp) = resp.take_body().into_future().await;
        assert_eq!(bytes.unwrap().unwrap(), Bytes::from_static(b"data: 4\n\n"));

        // remaining part
        let bytes = test::load_stream(resp.take_body().into_stream()).await;
        assert_eq!(bytes.unwrap(), Bytes::from_static(b"data: 3\n\ndata: 2\n\ndata: 1\n\n"));
    }
}

use actix_service::Service;
use actix_web::{web, App};
use futures::future::FutureExt;

#[actix_rt::main]
async fn main() {
    let app = App::new()
        .wrap_fn(|req, srv| {
            println!("Hi from start. You requested: {}", req.path());
            srv.call(req).map(|res| {
                println!("Hi from response");
                res
            })
        })
        .wrap_fn(|req, srv| {
            println!("Hi from start. You requested: {}", req.path());
            let fut = srv.call(req);
            async {
                let res = fut.await;
                println!("Hi from response");
                res
            }
        })
        .route(
            "/index.html",
            web::get().to(|| async {
                "Hello, middleware!"
            }),
        );
}

env_logger = "0.7"

use actix_web::middleware::Logger;
use env_logger;

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    use actix_web::{App, HttpServer};

    std::env::set_var("RUST_LOG", "actix_web=info");
    env_logger::init();

    HttpServer::new(|| {
        App::new()
            .wrap(Logger::default())
            .wrap(Logger::new("%a %{User-Agent}i"))
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

%a %t "%r" %s %b "%{Referer}i" "%{User-Agent}i" %T

use actix_web::{http, middleware, HttpResponse};

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    use actix_web::{web, App, HttpServer};

    HttpServer::new(|| {
        App::new()
            .wrap(middleware::DefaultHeaders::new().header("X-Version", "0.2"))
            .service(
                web::resource("/test")
                    .route(web::get().to(|| HttpResponse::Ok()))
                    .route(
                        web::method(http::Method::HEAD)
                            .to(|| HttpResponse::MethodNotAllowed()),
                    ),
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix_session::{CookieSession, Session};
use actix_web::{web, App, Error, HttpResponse, HttpServer};

async fn index(session: Session) -> Result<HttpResponse, Error> {
    // access session data
    if let Some(count) = session.get::<i32>("counter")? {
        session.set("counter", count + 1)?;
    } else {
        session.set("counter", 1)?;
    }

    Ok(HttpResponse::Ok().body(format!(
        "Count is {:?}!",
        session.get::<i32>("counter")?.unwrap()
    )))
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .wrap(
                CookieSession::signed(&[0; 32]) // <- create cookie based session middleware
                    .secure(false),
            )
            .service(web::resource("/").to(index))
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix_web::middleware::errhandlers::{ErrorHandlerResponse, ErrorHandlers};
use actix_web::{dev, http, HttpResponse, Result};

fn render_500<B>(mut res: dev::ServiceResponse<B>) -> Result<ErrorHandlerResponse<B>> {
    res.response_mut().headers_mut().insert(
        http::header::CONTENT_TYPE,
        http::HeaderValue::from_static("Error"),
    );
    Ok(ErrorHandlerResponse::Response(res))
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    use actix_web::{web, App, HttpServer};

    HttpServer::new(|| {
        App::new()
            .wrap(
                ErrorHandlers::new()
                    .handler(http::StatusCode::INTERNAL_SERVER_ERROR, render_500),
            )
            .service(
                web::resource("/test")
                    .route(web::get().to(|| HttpResponse::Ok()))
                    .route(web::head().to(|| HttpResponse::MethodNotAllowed())),
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix_files::NamedFile;
use actix_web::{HttpRequest, Result};
use std::path::PathBuf;

async fn index(req: HttpRequest) -> Result<NamedFile> {
    let path: PathBuf = req.match_info().query("filename").parse().unwrap();
    Ok(NamedFile::open(path)?)
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    use actix_web::{web, App, HttpServer};

    HttpServer::new(|| App::new().route("/{filename:.*}", web::get().to(index)))
        .bind("127.0.0.1:8088")?
        .run()
        .await
}

use actix_files as fs;
use actix_web::{App, HttpServer};

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new().service(fs::Files::new("/static", ".").show_files_listing())
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

use actix::{Actor, StreamHandler};
use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer};
use actix_web_actors::ws;

/// 定义Http Actor
struct MyWs;

impl Actor for MyWs {
    type Context = ws::WebsocketContext<Self>;
}

/// 一个 ws::Message 消息的 处理器
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for MyWs {
    fn handle(
        &mut self,
        msg: Result<ws::Message, ws::ProtocolError>,
        ctx: &mut Self::Context,
    ) {
        match msg {
            Ok(ws::Message::Ping(msg)) => ctx.pong(&msg),
            Ok(ws::Message::Text(text)) => ctx.text(text),
            Ok(ws::Message::Binary(bin)) => ctx.binary(bin),
            _ => (),
        }
    }
}

// ws://localhost:8088/ws/echo
/*
curl --include \
     --no-buffer \
     --header "Connection: Upgrade" \
     --header "Upgrade: websocket" \
     --header "Host: echo.websocket.org" \
     --header "Origin: https://echo.websocket.org" \
     --header "Sec-WebSocket-Key: NVwjmQUcWCenfWu98asDmg==" \
     --header "Sec-WebSocket-Version: 13" \
     http://localhost:8088/ws/echo
*/
async fn ws_echo(req: HttpRequest, stream: web::Payload) -> Result<HttpResponse, Error> {
    let resp = ws::start(MyWs {}, &req, stream);
    println!("{:?}", resp);
    resp
}


#[actix_rt::main]
async fn main() -> std::io::Result<()> {

    HttpServer::new(move || {

        App::new()
            .service(
                web::scope("/ws")
                    .route("/echo", web::get().to(ws_echo))
            )
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

cargo install systemfd cargo-watch

[dependencies]
listenfd = "0.3

use actix_web::{web, App, HttpRequest, HttpServer, Responder};
use listenfd::ListenFd;

async fn index(_req: HttpRequest) -> impl Responder {
    "Hello World!"
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    let mut listenfd = ListenFd::from_env();
    let mut server = HttpServer::new(|| App::new().route("/", web::get().to(index)));

    server = if let Some(l) = listenfd.take_tcp_listener(0).unwrap() {
        server.listen(l)?
    } else {
        server.bind("127.0.0.1:3000")?
    };

    server.run().await
}

systemfd --no-pid -s http::8088 -- cargo watch -x run

# 数据库
diesel = { version = "1.4", features = ["mysql", "r2d2", "chrono"] }
dotenv = "0.15.0"
# 时间日期
chrono = "0.4"

DATABASE_URL=mysql://root:123456@localhost/actix_learn

diesel setup
diesel migration generate init

-- Your SQL goes here
CREATE TABLE posts (
  id BIGINT PRIMARY KEY auto_increment comment 'ID',
  user_id BIGINT not null,
  title VARCHAR(256) NOT NULL comment '标题',
  body TEXT NOT NULL comment '内容',
  published BOOLEAN NOT NULL DEFAULT 0 comment '是否发布'
);

CREATE TABLE users (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  name TEXT NOT NULL,
  hair_color TEXT,
  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- This file should undo anything in `up.sql`
DROP TABLE posts
DROP TABLE users;

diesel migration run

#[macro_use]
extern crate diesel;
extern crate dotenv;

use diesel::prelude::*;
use dotenv::dotenv;
use diesel::r2d2;
use std::env;

pub mod schema;
pub mod model;

pub type PoolConnection = r2d2::Pool<r2d2::ConnectionManager<MysqlConnection>>;

pub fn new_connection_pool() ->  PoolConnection {
    dotenv().ok();

    let database_url = env::var("DATABASE_URL")
        .expect("DATABASE_URL must be set");
    let manager = r2d2::ConnectionManager::<MysqlConnection>::new(database_url);
    let pool = r2d2::Pool::builder()
        .max_size(15)
        .build(manager)
        .expect("Failed to create pool.");
    pool
}

use actix_learn::*;
use diesel::prelude::*;

// curl http://localhost:8088/block/user/create
async fn create_user(pool: web::Data<PoolConnection>) -> String {
    let conn = pool.get().expect("couldn't get db connection from pool");

    let r = web::block(move ||  {
        use schema::users;
        let user = model::UserForInsert {
            name: "name".to_string(),
            hair_color: Some("blank".to_string()),
        };
        diesel::insert_into(users::table)
            .values(&user)
            .execute(&conn)
    }).await;
    if let Err(e) = r {
        String::from(format!("{:?}", e))
    } else {
        String::from("create_success")
    }
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .route("/", web::to(|| HttpResponse::Ok()))
    })
    .bind("127.0.0.1:8088")?
    .run()
    .await
}

[dependencies]
actix = "0.9"
futures = "0.3"

extern crate actix;
use actix::prelude::*;

// 测试 Actor 对象
struct MyActor {
    count: usize,
}

// 所有的 Actor 必须实现 Actor 特质
impl Actor for MyActor {
    // 每个 Actor 都有一个执行上下文
    type Context = Context<Self>;
}

// Actor 接收的参数
struct Ping(usize);

// Actor 接收的参数必须实现 Message 对象
impl Message for Ping {
    // 定义 该参数的 返回值类型
    type Result = usize;
}

// Actor 的具体处理函数
impl Handler<Ping> for MyActor {
    // 返回值
    type Result = usize;

    fn handle(&mut self, msg: Ping, _ctx: &mut Context<Self>) -> Self::Result {
        self.count += msg.0;

        self.count
    }
}

#[actix_rt::main]
async fn main() {
    // 启动一个Actor，返回一个 Addr<MyActor>
    let addr = MyActor { count: 10 }.start();

    // 发送消息并获取 Future 结果
    let res = addr.send(Ping(10)).await;

    // handle() returns tokio handle
    // 返回一个结果
    println!("RESULT: {}", res.unwrap() == 20);

    // stop system and exit
    System::current().stop();
}

use std::io;
use actix::prelude::*;

struct Ping;

impl Message for Ping {
    type Result = Result<bool, io::Error>;
}

pub trait MessageResponse<A: Actor, M: Message> {
    fn handle<R: ResponseChannel<M>>(self, ctx: &mut A::Context, tx: Option<R>);
}

extern crate actix;
use actix::prelude::*;
use actix::dev::{MessageResponse, ResponseChannel};

// 测试 Actor 对象
struct MyActor {
    count: usize,
}

// 所有的 Actor 必须实现 Actor 特质
impl Actor for MyActor {
    // 每个 Actor 都有一个执行上下文
    type Context = Context<Self>;
}

// Actor 接收的参数
struct Ping(usize);

// Actor 接收的参数必须实现 Message 对象
impl Message for Ping {
    // 定义 该参数的 返回值类型
    type Result = Pong;
}

// 这是返回值
#[derive(Debug)]
struct Pong(bool);

// 返回值必须实现 MessageResponse 特质
impl<A, M> MessageResponse<A, M> for Pong
where
    A: Actor,
    M: Message<Result = Pong>,
{
    // 将 返回值 发送出去的逻辑
    fn handle<R: ResponseChannel<M>>(self, _: &mut A::Context, tx: Option<R>) {
        if let Some(tx) = tx {
            tx.send(self);
        }
    }
}

// Actor 的具体处理函数
impl Handler<Ping> for MyActor {
    // 返回值
    type Result = Pong;

    fn handle(&mut self, msg: Ping, _ctx: &mut Context<Self>) -> Self::Result {
        self.count += msg.0;

        // self.count
        Pong(true)
    }
}

#[actix_rt::main]
async fn main() {
    // 启动一个Actor，返回一个 Addr<MyActor>
    let addr = MyActor { count: 10 }.start();

    // 发送消息并获取 Future 结果
    let res = addr.send(Ping(10)).await;

    // handle() returns tokio handle
    // 返回一个结果
    println!("RESULT: {:?}", res);

    // stop system and exit
    System::current().stop();
}

struct MyActor;
impl Actor for MyActor {
    type Context = Context<Self>;

    fn started(&mut self, ctx: &mut Self::Context) {
        ctx.set_mailbox_capacity(1);
    }
}

let addr = MyActor.start();

struct MyActor;

struct WhoAmI;

impl Message for WhoAmI {
    type Result = Result<actix::Addr<MyActor>, ()>;
}

impl Actor for MyActor {
    type Context = Context<Self>;
}

impl Handler<WhoAmI> for MyActor {
    type Result = Result<actix::Addr<MyActor>, ()>;

    fn handle(&mut self, msg: WhoAmI, ctx: &mut Context<Self>) -> Self::Result {
        Ok(ctx.address())
    }
}

let who_addr = addr.do_send(WhoAmI {} );

impl Handler<Ping> for MyActor {
    type Result = usize;

    fn handle(&mut self, msg: Ping, ctx: &mut Context<Self>) -> Self::Result {
        self.count += msg.0;

        if self.count > 5 {
            println!("Shutting down ping receiver.");
            ctx.stop()
        }

        self.count
    }
}

extern crate actix;
extern crate futures;
use actix::prelude::*;
use futures::FutureExt;
use futures::TryFutureExt;

struct SumActor {}

impl Actor for SumActor {
    type Context = Context<Self>;
}

struct Value(usize, usize);

impl Message for Value {
    type Result = usize;
}

impl Handler<Value> for SumActor {
    type Result = usize;

    fn handle(&mut self, msg: Value, _ctx: &mut Context<Self>) -> Self::Result {
        msg.0 + msg.1
    }
}

struct DisplayActor {}

impl Actor for DisplayActor {
    type Context = Context<Self>;
}

struct Display(usize);

impl Message for Display {
    type Result = ();
}

impl Handler<Display> for DisplayActor {
    type Result = ();

    fn handle(&mut self, msg: Display, _ctx: &mut Context<Self>) -> Self::Result {
        println!("Got {:?}", msg.0);
    }
}

fn main() {
    let system = System::new("single-arbiter-example");

    // 创建 Addr
    let sum_addr = SumActor {}.start();
    let dis_addr = DisplayActor {}.start();

    // 定义一个执行流的Future
    // 起初发送 `Value(6, 7)` 给 `SumActor`
    // `Addr::send` 返回 `Request` 类型，该类型实现了 `Future`
    // Future::Output = Result<usize, MailboxError>
    let execution = sum_addr
        .send(Value(6, 7))
        // `.map_err` 转换 `Future<usize, MailboxError>` 为 `Future<usize, ()>`
        //   如果有错误将打印错误信息
        // 实现来自于 use futures::TryFutureExt;
        .map_err(|e| {
            eprintln!("Encountered mailbox error: {:?}", e);
        })
        // 假设发送成功，并成功返回，and_then将得到执行，其中参数为上一个Future的Result<T, E> 的 T
        // 实现来自于 use futures::TryFutureExt;
        .and_then(move |res| {
            // `res` 是 `SumActor` 参数为 `Value(6, 7)` 的返回值，类型为 `usize`

            // res 发送给 DisplayActor 展示
            dis_addr.send(Display(res)).map_err(|_| ())
        })
        .map(move |_| {
            // 当 DisplayActor 返回后停止，将关闭所有 Actor
            System::current().stop();
        });

    // 提交 Future 到 Arbiter/event 循环中
    Arbiter::spawn(execution);

    system.run().unwrap();
}

use actix::prelude::*;

struct MySyncActor;

impl Actor for MySyncActor {
    type Context = SyncContext<Self>;
}


fn main() {

    // 同步执行器
    // 创建运行在指定线程中的Actor
    let addr = SyncArbiter::start(2, || MySyncActor);
}