Method query_streaming

src/sql-server-util/src/lib.rs:220–259

Executes SQL statements in SQL Server, returning a [`Stream`] of resulting rows. Passthrough method for [`tiberius::Client::query`].

(
        &'c mut self,
        query: Q,
        params: &[&dyn tiberius::ToSql],
    )

    ///
    /// Passthrough method for [`tiberius::Client::query`].
    pub fn query_streaming<'c, 'q, Q>(
        &'c mut self,
        query: Q,
        params: &[&dyn tiberius::ToSql],
    ) -> impl Stream<Item = Result<tiberius::Row, SqlServerError>> + Send + use<'c, Q>
    where
        Q: Into<Cow<'q, str>>,
    {
        let (tx, rx) = tokio::sync::oneshot::channel();
        let params = params
            .iter()
            .map(|p| OwnedColumnData::from(p.to_sql()))
            .collect();
        let kind = RequestKind::QueryStreamed {
            query: query.into().to_string(),
            params,
        };

        // Make our initial request which will return a Stream of Rows.
        let request_future = async move {
            self.tx
                .send(Request { tx, kind })
                .context("sending request")?;

            let response = rx.await.context("channel")??;
            match response {
                Response::RowStream { stream } => {
                    Ok(tokio_stream::wrappers::ReceiverStream::new(stream))
                }
                other @ Response::Execute { .. } | other @ Response::Rows(_) => {
                    Err(SqlServerError::ProgrammingError(format!(
                        "expected Response::Rows, got {other:?}"
                    )))
                }
            }
        };

        // "flatten" our initial request into the returned stream.
        futures::stream::once(request_future).try_flatten()
    }

    /// Executes multiple queries, delimited with `;` and return multiple
    /// result sets; one for each query.