use futures::stream::{select_all, TryStreamExt};
use reql::{r, Driver, Error};
use serde_json::Value;

#[tokio::test]
async fn changefeeds_should_use_dedicated_connections() {
    tracing_subscriber::fmt::init();

    match changefeeds().await.unwrap_err() {
        Error::Driver(Driver::ConnectionLocked) => {}
        error => panic!("{:?}", error),
    }
}

async fn changefeeds() -> reql::Result<()> {
    let conn = r.connect(()).await?;

    let _ = r
        .table_create("foo")
        .run::<_, Value>(&conn)
        .try_next()
        .await;
    let foo = r.table("foo").changes(()).run::<_, Value>(&conn);

    let _ = r
        .table_create("bar")
        .run::<_, Value>(&conn)
        .try_next()
        .await;
    let bar = r.table("bar").changes(()).run::<_, Value>(&conn);

    let mut list = select_all(vec![foo, bar]);

    while let Some(_) = list.try_next().await? {}

    Ok(())
}