mysql_async 0.20.0

Tokio based asynchronous MySql client library.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

// Copyright (c) 2019 Anatoly Ikorsky
//
// Licensed under the Apache License, Version 2.0
// <LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0> or the MIT
// license <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
// option. All files in the project carrying such notice may not be copied,
// modified, or distributed except according to those terms.

use ::futures::Future;
use mysql_async::error::Error;
use mysql_async::prelude::*;
use mysql_async::{MyFuture, Opts, Pool, QueryResult};

use std::env;
use std::io;

/// Same as `tokio::run`, but will panic if future panics and will return the result
/// of future execution.
fn run<F, T, U>(future: F) -> Result<T, U>
where
    F: Future<Item = T, Error = U> + Send + 'static,
    T: Send + 'static,
    U: Send + 'static,
{
    let mut runtime = tokio::runtime::Runtime::new().unwrap();
    let result = runtime.block_on(future);
    runtime.shutdown_on_idle().wait().unwrap();
    result
}

fn get_url() -> String {
    if let Ok(url) = env::var("DATABASE_URL") {
        let opts = Opts::from_url(&url).expect("DATABASE_URL invalid");
        if opts
            .get_db_name()
            .expect("a database name is required")
            .is_empty()
        {
            panic!("database name is empty");
        }
        url
    } else {
        "mysql://root:password@127.0.0.1:3307/mysql".into()
    }
}

pub fn get_all_results<TupleType, T, P>(result: QueryResult<T, P>) -> impl MyFuture<Vec<TupleType>>
where
    TupleType: FromRow + Send + 'static,
    P: Protocol + Send + 'static,
    T: ConnectionLike + Sized + Send + 'static,
{
    result.collect().map(|(_, data)| data)
}

pub fn get_single_result<TupleType, T, P>(result: QueryResult<T, P>) -> impl MyFuture<TupleType>
where
    TupleType: FromRow + Send + 'static,
    P: Protocol + Send + 'static,
    T: ConnectionLike + Sized + Send + 'static,
{
    get_all_results(result).and_then(|mut data| {
        if data.len() != 1 {
            Err(Error::from(io::Error::from(io::ErrorKind::InvalidData)))
        } else {
            Ok(data.remove(0))
        }
    })
}

#[test]
fn use_generic_code() {
    let pool = Pool::new(Opts::from_url(&*get_url()).unwrap());
    let fut = pool
        .get_conn()
        .and_then(move |conn| conn.query("SELECT 1, 2, 3"))
        .and_then(get_single_result::<(u8, u8, u8), _, _>)
        .and_then(|out| pool.disconnect().map(move |_| out));

    let result = run(fut).unwrap();
    assert_eq!(result, (1, 2, 3));
}