Is it possible to implement a real Stream?

[nappa85]

To make a non-panicking collect, with the actual code I have to do something like

async fn query<T: FromRow, E: From<FromRowError> + From<Error>>(conn: &mut Conn, query: &str) -> Result<Vec<T>, E> {
  conn.query_iter(query).await?
    .try_collect::<T>().await?
    .into_iter().collect::<Result<Vec<T>, E>>()
}

(this is only for example purposes, could contains errors)

This means allocating the entire resultset even if every single row is an error, or having to implement a wrapper to use next() as a Stream...
Wouldn't be possible, for example, to add an impl of futures_core::Stream that is the same used by tokio-stream and it seems to be a standard de-facto?
If you prefer, under a feature, so it wouldn't be default for everyone.

I can do the PR if you want

[blackbeam]

Hi.

To make a non-panicking collect, with the actual code I have to do something like... This means allocating the entire resultset even if every single row is an error

You meant that you want to stop on the first row that isn't convertible to T? If yes, then the QueryResult::reduce will do this for you in the most effective manner, because mysql protocol requires you to consume the whole QueryResult:

async fn query<T, E>(conn: &mut Conn, query: &str) -> Result<Vec<T>, E>
where
    T: FromRow,
    E: From<FromRowError> + From<Error>,
{
    let mut done = false;
    conn.query_iter(query).await?
        .reduce(Ok(Vec::new()), |mut result, row: Row| {
            match result {
                Ok(mut acc) => {
                    match from_row_opt::<T>(row) {
                        Ok(x) => {
                            acc.push(x);
                            Ok(acc)
                        }
                        Err(e) => Err(e.into())
                    }
                },
                err => err,
            }
        }).await
}

(this is only for example purposes, could contains errors)

Wouldn't be possible, for example, to add an impl of futures_core::Stream that is the same used by tokio-stream and it seems to be a standard de-facto?

This would be great.
You should note, however, that this may be tricky, because you should cover multi-result set case. This means that:

1. QueryResult is the stream of streams
2. 2-nd level streams have Rows of different shapes so you'd better to use the Row type as an Item,
3. It's better to make result set metadata accessible through both 1-st and 2-nd level streams,
4. You have to consume 2-nd level stream before calling next on the 1-st level stream, so (i) 2-nd should borrow 1-st, (ii) 1-st should consume unconsumed 2-nd at call to 1-st's next.

[nappa85]

Ok, this makes everything complicated, I wasn't taking multiple resultsets in consideration...
But, as far as I can see, in the actual architecture it's always next() to return every resultset rows, putting a None at the end of every resultset as a separator.
I can see the stream-of-streams, but it's more like a single stream that changes datatype on the run, because streams aren't concurrent but sequential.
Correct me if I'm wrong, but this kind of behaviour is easily solved by an enum that impl FromRow and uses the data to discriminate between variants, but it's something the user has to write, not the crate itself...

[nappa85]

My proposal for a Stream impl is here, based on a solution by @dodomorandi: https://github.com/nappa85/mysql-stream/blob/master/src/lib.rs
There are both stream and stream_and_drop methods, mimic the style of QueryResult methods.

[blackbeam]

Great!. I'll look into it.

[nappa85]

Hello, any news about the subject?
Also a "your code is bad and you should feel bad" is accepted

[blackbeam]

Oh! No, the code seem OK, I just literally have no spare time at the moment. I also like the approach that there are no "stream of streams".

I'll integrate this into the code base as soon as I can.

.

[blackbeam]

@nappa85, your proposal was really helpful, thanks!

Published as v0.28.1.

[nappa85]

Great to hear that, good work! Il sab 18 set 2021, 14:06 Anatoly I ***@***.***> ha scritto:

…

@nappa85 <https://github.com/nappa85>, your proposal was really helpful, thanks! Published as v0.28.1 <https://github.com/blackbeam/mysql_async/releases/tag/v0.28.1>. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#166 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ABZXINNEUC5R4ESZWWY4HJLUCR6FLANCNFSM5BI4EEBA> . Triage notifications on the go with GitHub Mobile for iOS <https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675> or Android <https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub>.