| 36 | #[allow(dead_code)] |
| 37 | #[instrument(level = "trace", skip(metric_callback))] |
| 38 | pub(crate) fn build( |
| 39 | stmt: Statement, |
| 40 | conn: InnerConnection, |
| 41 | metric_callback: Option<crate::metric::Callback>, |
| 42 | ) -> QueryStream { |
| 43 | QueryStreamBuilder { |
| 44 | stmt, |
| 45 | conn, |
| 46 | metric_callback, |
| 47 | stream_builder: |conn, stmt, _metric_callback| match conn { |
| 48 | #[cfg(feature = "sqlx-mysql")] |
| 49 | InnerConnection::MySql(c) => { |
| 50 | let query = crate::driver::sqlx_mysql::sqlx_query(stmt); |
| 51 | let _start = _metric_callback.is_some().then(std::time::SystemTime::now); |
| 52 | let stream = c |
| 53 | .fetch(query) |
| 54 | .map_ok(Into::into) |
| 55 | .map_err(sqlx_error_to_query_err); |
| 56 | let elapsed = _start.map(|s| s.elapsed().unwrap_or_default()); |
| 57 | MetricStream::new(_metric_callback, stmt, elapsed, stream) |
| 58 | } |
| 59 | #[cfg(feature = "sqlx-postgres")] |
| 60 | InnerConnection::Postgres(c) => { |
| 61 | let query = crate::driver::sqlx_postgres::sqlx_query(stmt); |
| 62 | let _start = _metric_callback.is_some().then(std::time::SystemTime::now); |
| 63 | let stream = c |
| 64 | .fetch(query) |
| 65 | .map_ok(Into::into) |
| 66 | .map_err(sqlx_error_to_query_err); |
| 67 | let elapsed = _start.map(|s| s.elapsed().unwrap_or_default()); |
| 68 | MetricStream::new(_metric_callback, stmt, elapsed, stream) |
| 69 | } |
| 70 | #[cfg(feature = "sqlx-sqlite")] |
| 71 | InnerConnection::Sqlite(c) => { |
| 72 | let query = crate::driver::sqlx_sqlite::sqlx_query(stmt); |
| 73 | let _start = _metric_callback.is_some().then(std::time::SystemTime::now); |
| 74 | let stream = c |
| 75 | .fetch(query) |
| 76 | .map_ok(Into::into) |
| 77 | .map_err(sqlx_error_to_query_err); |
| 78 | let elapsed = _start.map(|s| s.elapsed().unwrap_or_default()); |
| 79 | MetricStream::new(_metric_callback, stmt, elapsed, stream) |
| 80 | } |
| 81 | #[cfg(feature = "mock")] |
| 82 | InnerConnection::Mock(c) => { |
| 83 | let _start = _metric_callback.is_some().then(std::time::SystemTime::now); |
| 84 | let stream = c.fetch(stmt); |
| 85 | let elapsed = _start.map(|s| s.elapsed().unwrap_or_default()); |
| 86 | MetricStream::new(_metric_callback, stmt, elapsed, stream) |
| 87 | } |
| 88 | #[cfg(feature = "proxy")] |
| 89 | InnerConnection::Proxy(c) => { |
| 90 | todo!("Proxy connection is not supported") |
| 91 | } |
| 92 | #[allow(unreachable_patterns)] |
| 93 | _ => unreachable!(), |
| 94 | }, |
| 95 | } |