Database integration

Diesel

At the moment of 1.0 release Diesel does not support asynchronous operations. But it possible to use actix synchronous actor system as a db interface api. Multiple sync actors could be run in parallel, in this case all of this actors process messages from the same queue (sync actors actually work in mpmc mode).

Let's create simple db api that can insert new user row into sqlite table. We need to define sync actor and connection that this actor will use. Same approach could used for other databases.

use actix::prelude::*;*

struct DbExecutor(SqliteConnection);

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

This is definition of our actor. Now we need to define create user message and response.

struct CreateUser {
    name: String,
}

impl ResponseType for CreateUser {
    type Item = models::User;
    type Error = Error;
}

We can send CreateUser message to DbExecutor actor, and as result we can get User model. Now we need to define actual handler implementation for this message.

impl Handler<CreateUser> for DbExecutor {

    fn handle(&mut self, msg: CreateUser, _: &mut Self::Context) -> Response<Self, CreateUser>
    {
        use self::schema::users::dsl::*;

        // Create insertion model
        let uuid = format!("{}", uuid::Uuid::new_v4());
        let new_user = models::NewUser {
            id: &uuid,
            name: &msg.name,
        };

        // normal diesl operations
        diesel::insert_into(users)
            .values(&new_user)
            .execute(&self.0)
            .expect("Error inserting person");

        let mut items = users
            .filter(id.eq(&uuid))
            .load::<models::User>(&self.0)
            .expect("Error loading person");

        Self::reply(items.pop().unwrap())
    }
}

That is it. Now we can use DbExecutor actor from any http handler or middleware. All we need is to start DbExecutor actors and store address in state where http handler can access it.

/// This is state where we sill store *DbExecutor* address.
struct State {
    db: SyncAddress<DbExecutor>,
}

fn main() {
    let sys = actix::System::new("diesel-example");

    // Start 3 db executors
    let addr = SyncArbiter::start(3, || {
        DbExecutor(SqliteConnection::establish("test.db").unwrap())
    });

    // Start http server
    HttpServer::new(move || {
        Application::with_state(State{db: addr.clone()})
            .resource("/{name}", |r| r.method(Method::GET).a(index))})
        .bind("127.0.0.1:8080").unwrap()
        .start().unwrap();

    println!("Started http server: 127.0.0.1:8080");
    let _ = sys.run();
}

And finally we can use this state in handler function. We get message response asynchronously, so handler needs to return future object, also Route::a() needs to be used for async handler registration.

/// Async handler
fn index(req: HttpRequest<State>) -> Box<Future<Item=HttpResponse, Error=Error>> {
    let name = &req.match_info()["name"];

    Box::new(
        // Send message to `DbExecutor` actor
        req.state().db.call_fut(CreateUser{name: name.to_owned()})
            .and_then(|res| {
                match res {
                    Ok(user) => ok(HTTPOk.build().json(user)),
                    Err(_) => ok(HTTPInternalServerError.response())
                }
            })
            .map_err(|e| error::ErrorInternalServerError(e).into()))
}

Full example is available in examples repository.

More information on sync actors could be found in actix documentation.

3.7 KiB Raw Blame History

Database integration

Diesel

3.7 KiB

Raw Blame History