2

I need to collect all the rows of a table using tiberius and output them. My simple code is:

extern crate futures;
extern crate tokio_core;
extern crate tiberius;

use futures::Future;
use tokio_core::reactor::Core;
use tiberius::SqlConnection;
use tiberius::stmt::ResultStreamExt;

fn main() {
    let mut core = Core::new().unwrap();

    let future = SqlConnection::connect(core.handle(), "server=tcp:127.0.0.1,1433;username=SA;password=qweasdZXC123!!;")
        .and_then(|conn| {
            let mut v: Vec<String> = Vec::new();

            conn.simple_query("SELECT id, name FROM test").for_each_row(|row| {

                let id: i32 = row.get(0);
                let name: &str = row.get(1);

                v.push(format!("{} - {}", id, name));

                Ok(())
            });

            println!("{:?}", v);

            Ok(())
        });

    core.run(future).unwrap();
}

This code prints an empty vector, but I need full vector of strings. I have read some articles about futures, but as as newbie to Rust, they look too complicated for me.

2
  • What's not working exactly? Is there an error? Sep 21, 2017 at 13:07
  • It compiles but hangs on request. I think problem in wait() call.
    – Sartor
    Sep 21, 2017 at 13:20

1 Answer 1

1

In your original code, you had

conn.simple_query("SELECT id, name FROM users").for_each_row(|row| {
    // ...
}).wait().unwrap();

And you said "It compiles but hangs on request. I think problem in wait() call.".

If you read the documentation for Future::wait, you'll see this warning, emphasis mine:

Note: This method is not appropriate to call on event loops or similar I/O situations because it will prevent the event loop from making progress (this blocks the thread). This method should only be called when it's guaranteed that the blocking work associated with this future will be completed by another thread.

In your updated code, you have

conn.simple_query("SELECT id, name FROM test").for_each_row(|row| {
    // ...
});

This constructs a future but then drops it immediately, thus nothing ever happens to the outer vector. All of the futures in the futures crate have a warning attached to them for exactly this reason:

warning: unused `futures::FutureResult` which must be used: futures do nothing unless polled

I've filed an issue so that the library adds this.


Here's a completely untested piece of code. I don't have a SQL Server instance around to actually test it, but it does compile and has the right shape.

extern crate futures;
extern crate futures_state_stream;
extern crate tokio_core;
extern crate tiberius;

use futures::{Future, Stream};
use futures_state_stream::StateStream;
use tiberius::SqlConnection;
use tokio_core::reactor::Core;

fn main() {
    let mut core = Core::new().unwrap();

    let connection_string = "server=tcp:127.0.0.1,1433;username=SA;password=qweasdZXC123!!;";

    let future = SqlConnection::connect(core.handle(), connection_string)
        .and_then(|conn| {

            let query = conn.query("SELECT * FROM test WHERE id > @P1", &[&0i32])
                .into_stream()
                .take(1);

            query.flatten()
                .map(|row| {
                    let id: i32 = row.get(0);
                    let name: &str = row.get(1);

                    format!("{} - {}", id, name)
                })
                .collect()
        });

    let all_rows = core.run(future).unwrap();
.
    println!("{:?}", all_rows);
}

Important points:

  1. conn.query can be used with multiple query statements, so it returns a stream of result sets.
  2. conn.query actually implements a StateStream, not a futures::Stream. For the purposes of the example, I convert it back to a futures::Stream with .into_stream(). This is non-ideal because we lose the ability to recover the conn afterwards.
  3. I take only up to the first result set with .take(1).
  4. Since we now have a Stream of Streams, we use Stream::flatten to remove the nesting.
  5. Each row is mapped to a String.
  6. The Stream is collected into a single Future of a Vec<String>
1
  • Thank you very much! This is exactly what I need. Key feature is work with more generic StateStream instead of tiberius::stmt::ResultStreamExt.
    – Sartor
    Sep 21, 2017 at 15:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.