数据库

Diesel

目前，Diesel 1.0不支持异步操作，但可以将actix同步actor系统用作数据库接口API。从技术上讲，同步actor是worker风格的actor。 多个同步actors可以并行运行并处理来自同一队列的消息。同步actors以mpsc模式工作。

我们来创建一个简单的数据库api，它可以将一个新的 user row插入到SQLite表中。我们必须定义一个同步actor和该actor将使用的连接。其他数据库可以使用相同的方法。

  1. use actix::prelude::*;
  3. struct DbExecutor(SqliteConnection);
  5. impl Actor for DbExecutor {
  6.     type Context = SyncContext<Self>;
  7. }

这是我们actor的定义。现在，我们必须定义创建用户消息和响应。

  1. struct CreateUser {
  2.     name: String,
  3. }
  5. impl Message for CreateUser {
  6.     type Result = Result<User, Error>;
  7. }

我们可以向演员发送CreateUser消息DbExecutor actor，因此我们将收到一个 User实例。接下来，我们必须为此消息定义处理程序实现。

  1. impl Handler<CreateUser> for DbExecutor {
  2.     type Result = Result<User, Error>;
  4.     fn handle(&mut self, msg: CreateUser, _: &mut Self::Context) -> Self::Result
  5.     {
  6.         use self::schema::users::dsl::*;
  8.         // Create insertion model
  9.         let uuid = format!("{}", uuid::Uuid::new_v4());
  10.         let new_user = models::NewUser {
  11.             id: &uuid,
  12.             name: &msg.name,
  13.         };
  15.         // normal diesel operations
  16.         diesel::insert_into(users)
  17.             .values(&new_user)
  18.             .execute(&self.0)
  19.             .expect("Error inserting person");
  21.         let mut items = users
  22.             .filter(id.eq(&uuid))
  23.             .load::<models::User>(&self.0)
  24.             .expect("Error loading person");
  26.         Ok(items.pop().unwrap())
  27.     }
  28. }

仅此而已！现在，我们可以使用来在于任何http处理程序或中间件的DbExecutor actor。我们需要的只是启动DbExecutor actors并将地址存储在http处理程序可以访问的状态中。

  1. /// This is state where we will store *DbExecutor* address.
  2. struct State {
  3.     db: Addr<Syn, DbExecutor>,
  4. }
  6. fn main() {
  7.     let sys = actix::System::new("diesel-example");
  9.     // Start 3 parallel db executors
  10.     let addr = SyncArbiter::start(3, || {
  11.         DbExecutor(SqliteConnection::establish("test.db").unwrap())
  12.     });
  14.     // Start http server
  15.     HttpServer::new(move || {
  16.         App::with_state(State{db: addr.clone()})
  17.             .resource("/{name}", |r| r.method(Method::GET).a(index))})
  18.         .bind("127.0.0.1:8080").unwrap()
  19.         .start().unwrap();
  21.     println!("Started http server: 127.0.0.1:8080");
  22.     let _ = sys.run();
  23. }

我们将在请求处理程序中使用该地址。处理程序返回future对象; 因此，我们异步接收响应消息。Route::a()必须用于异步处理注册。

  1. /// Async handler
  2. fn index(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
  3.     let name = &req.match_info()["name"];
  5.     // Send message to `DbExecutor` actor
  6.     req.state().db.send(CreateUser{name: name.to_owned()})
  7.         .from_err()
  8.         .and_then(|res| {
  9.             match res {
  10.                 Ok(user) => Ok(HttpResponse::Ok().json(user)),
  11.                 Err(_) => Ok(HttpResponse::InternalServerError().into())
  12.             }
  13.         })
  14.         .responder()
  15. }

diesel directory提供了一个完整的示例。

有关同步actors的更多信息可以在actix documentation文档中找到 。