diesel/mysql_like/connection/
mod.rs

1mod bind;
2mod raw;
3mod stmt;
4mod url;
5
6use core::num::NonZeroU64;
7
8use self::raw::RawConnection;
9use self::stmt::Statement;
10use self::stmt::iterator::StatementIterator;
11use self::url::ConnectionOptions;
12use crate::RunQueryDsl;
13use crate::connection::instrumentation::{DebugQuery, DynInstrumentation, StrQueryHelper};
14use crate::connection::statement_cache::{MaybeCached, StatementCache};
15use crate::connection::*;
16use crate::expression::QueryMetadata;
17use crate::mysql_like::MysqlLikeBackend;
18use crate::query_builder::bind_collector::RawBytesBindCollector;
19use crate::query_builder::*;
20use crate::result::*;
21
22#[allow(missing_debug_implementations, missing_copy_implementations)]
23/// A connection to a MySQL database. Connection URLs should be in the form
24/// `mysql://[user[:password]@]host/database_name[?unix_socket=socket-path&ssl_mode=SSL_MODE*&ssl_ca=/etc/ssl/certs/ca-certificates.crt&ssl_cert=/etc/ssl/certs/client-cert.crt&ssl_key=/etc/ssl/certs/client-key.crt]`
25///
26///* `host` can be an IP address or a hostname. If it is set to `localhost`, a connection
27///  will be attempted through the socket at `/tmp/mysql.sock`. If you want to connect to
28///  a local server via TCP (e.g. docker containers), use `0.0.0.0` or `127.0.0.1` instead.
29/// * `unix_socket` expects the path to the unix socket
30/// * `ssl_ca` accepts a path to the system's certificate roots
31/// * `ssl_cert` accepts a path to the client's certificate file
32/// * `ssl_key` accepts a path to the client's private key file
33/// * `ssl_mode` expects a value defined for MySQL client command option `--ssl-mode`
34///   See <https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#option_general_ssl-mode>
35///
36/// # Supported loading model implementations
37///
38/// * [`DefaultLoadingMode`](crate::connection::DefaultLoadingMode)
39///
40/// As `MysqlConnection` only supports a single loading mode implementation
41/// it is **not required** to explicitly specify a loading mode
42/// when calling [`RunQueryDsl::load_iter()`] or [`LoadConnection::load`]
43///
44/// ## DefaultLoadingMode
45///
46/// `MysqlConnection` only supports a single loading mode, which loads
47/// values row by row from the result set.
48///
49/// ```rust
50/// # include!("../../doctest_setup.rs");
51/// #
52/// # fn main() {
53/// #     run_test().unwrap();
54/// # }
55/// #
56/// # fn run_test() -> QueryResult<()> {
57/// #     use schema::users;
58/// #     let connection = &mut establish_connection();
59/// use diesel::connection::DefaultLoadingMode;
60/// {
61///     // scope to restrict the lifetime of the iterator
62///     let iter1 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
63///
64///     for r in iter1 {
65///         let (id, name) = r?;
66///         println!("Id: {} Name: {}", id, name);
67///     }
68/// }
69///
70/// // works without specifying the loading mode
71/// let iter2 = users::table.load_iter::<(i32, String), _>(connection)?;
72///
73/// for r in iter2 {
74///     let (id, name) = r?;
75///     println!("Id: {} Name: {}", id, name);
76/// }
77/// #   Ok(())
78/// # }
79/// ```
80///
81/// This mode does **not support** creating
82/// multiple iterators using the same connection.
83///
84/// ```compile_fail
85/// # include!("../../doctest_setup.rs");
86/// #
87/// # fn main() {
88/// #     run_test().unwrap();
89/// # }
90/// #
91/// # fn run_test() -> QueryResult<()> {
92/// #     use schema::users;
93/// #     let connection = &mut establish_connection();
94/// use diesel::connection::DefaultLoadingMode;
95///
96/// let iter1 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
97/// let iter2 = users::table.load_iter::<(i32, String), DefaultLoadingMode>(connection)?;
98///
99/// for r in iter1 {
100///     let (id, name) = r?;
101///     println!("Id: {} Name: {}", id, name);
102/// }
103///
104/// for r in iter2 {
105///     let (id, name) = r?;
106///     println!("Id: {} Name: {}", id, name);
107/// }
108/// #   Ok(())
109/// # }
110/// ```
111pub struct MysqlLikeConnection<DB: MysqlLikeBackend> {
112    raw_connection: RawConnection,
113    transaction_state: AnsiTransactionManager,
114    statement_cache: StatementCache<DB, Statement<DB>>,
115    instrumentation: DynInstrumentation,
116}
117
118// mysql connection can be shared between threads according to libmysqlclients documentation
119#[allow(unsafe_code)]
120unsafe impl<DB: MysqlLikeBackend> Send for MysqlLikeConnection<DB> {}
121
122impl<DB: MysqlLikeBackend> SimpleConnection for MysqlLikeConnection<DB> {
123    fn batch_execute(&mut self, query: &str) -> QueryResult<()> {
124        self.instrumentation
125            .on_connection_event(InstrumentationEvent::StartQuery {
126                query: &StrQueryHelper::new(query),
127            });
128        let r = self
129            .raw_connection
130            .enable_multi_statements(|| self.raw_connection.execute(query));
131        self.instrumentation
132            .on_connection_event(InstrumentationEvent::FinishQuery {
133                query: &StrQueryHelper::new(query),
134                error: r.as_ref().err(),
135            });
136        r
137    }
138}
139
140impl<DB: MysqlLikeBackend> ConnectionSealed for MysqlLikeConnection<DB> {}
141
142impl<DB: MysqlLikeBackend> Connection for MysqlLikeConnection<DB> {
143    type Backend = DB;
144    type TransactionManager = AnsiTransactionManager;
145
146    /// Establishes a new connection to the MySQL database
147    /// `database_url` may be enhanced by GET parameters
148    /// `mysql://[user[:password]@]host[:port]/database_name[?unix_socket=socket-path&ssl_mode=SSL_MODE*&ssl_ca=/etc/ssl/certs/ca-certificates.crt&ssl_cert=/etc/ssl/certs/client-cert.crt&ssl_key=/etc/ssl/certs/client-key.crt&local_infile=true]`
149    ///
150    /// * `host` can be an IP address or a hostname. If it is set to `localhost`, a connection
151    ///   will be attempted through the socket at `/tmp/mysql.sock`. If you want to connect to
152    ///   a local server via TCP (e.g. docker containers), use `0.0.0.0` or `127.0.0.1` instead.
153    /// * `unix_socket` expects the path to the unix socket
154    /// * `ssl_ca` accepts a path to the system's certificate roots
155    /// * `ssl_cert` accepts a path to the client's certificate file
156    /// * `ssl_key` accepts a path to the client's private key file
157    /// * `ssl_mode` expects a value defined for MySQL client command option `--ssl-mode`
158    ///   See <https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#option_general_ssl-mode>
159    /// * `local_infile` expects a boolean to enable or disable LOAD DATA LOCAL
160    fn establish(database_url: &str) -> ConnectionResult<Self> {
161        let mut instrumentation = DynInstrumentation::default_instrumentation();
162        instrumentation.on_connection_event(InstrumentationEvent::StartEstablishConnection {
163            url: database_url,
164        });
165
166        let establish_result = Self::establish_inner(database_url);
167        instrumentation.on_connection_event(InstrumentationEvent::FinishEstablishConnection {
168            url: database_url,
169            error: establish_result.as_ref().err(),
170        });
171        let mut conn = establish_result?;
172        conn.instrumentation = instrumentation;
173        Ok(conn)
174    }
175
176    fn execute_returning_count<T>(&mut self, source: &T) -> QueryResult<usize>
177    where
178        T: QueryFragment<Self::Backend> + QueryId,
179    {
180        #[allow(unsafe_code)] // call to unsafe function
181        update_transaction_manager_status(
182            prepared_query(
183                &source,
184                &mut self.statement_cache,
185                &mut self.raw_connection,
186                &mut *self.instrumentation,
187            )
188            .and_then(|stmt| {
189                // we have not called result yet, so calling `execute` is
190                // fine
191                let stmt_use = unsafe { stmt.execute() }?;
192                stmt_use.affected_rows()
193            }),
194            &mut self.transaction_state,
195            &mut self.instrumentation,
196            &crate::debug_query(source),
197        )
198    }
199
200    fn transaction_state(&mut self) -> &mut AnsiTransactionManager {
201        &mut self.transaction_state
202    }
203
204    fn instrumentation(&mut self) -> &mut dyn Instrumentation {
205        &mut *self.instrumentation
206    }
207
208    fn set_instrumentation(&mut self, instrumentation: impl Instrumentation) {
209        self.instrumentation = instrumentation.into();
210    }
211
212    fn set_prepared_statement_cache_size(&mut self, size: CacheSize) {
213        self.statement_cache.set_cache_size(size);
214    }
215}
216
217#[inline(always)]
218fn update_transaction_manager_status<T>(
219    query_result: QueryResult<T>,
220    transaction_manager: &mut AnsiTransactionManager,
221    instrumentation: &mut DynInstrumentation,
222    query: &dyn DebugQuery,
223) -> QueryResult<T> {
224    fn non_generic_inner(
225        query_result: Result<(), &Error>,
226        transaction_manager: &mut AnsiTransactionManager,
227        instrumentation: &mut DynInstrumentation,
228        query: &dyn DebugQuery,
229    ) {
230        if let Err(Error::DatabaseError(DatabaseErrorKind::SerializationFailure, _)) = query_result
231        {
232            transaction_manager
233                .status
234                .set_requires_rollback_maybe_up_to_top_level(true)
235        }
236        instrumentation.on_connection_event(InstrumentationEvent::FinishQuery {
237            query,
238            error: query_result.err(),
239        });
240    }
241
242    non_generic_inner(
243        query_result.as_ref().map(|_| ()),
244        transaction_manager,
245        instrumentation,
246        query,
247    );
248    query_result
249}
250
251impl<DB: MysqlLikeBackend> LoadConnection<DefaultLoadingMode> for MysqlLikeConnection<DB> {
252    type Cursor<'conn, 'query> = self::stmt::iterator::StatementIterator<'conn, DB>;
253    type Row<'conn, 'query> = self::stmt::iterator::MysqlRow<DB>;
254
255    fn load<'conn, 'query, T>(
256        &'conn mut self,
257        source: T,
258    ) -> QueryResult<Self::Cursor<'conn, 'query>>
259    where
260        T: Query + QueryFragment<Self::Backend> + QueryId + 'query,
261        Self::Backend: QueryMetadata<T::SqlType>,
262    {
263        update_transaction_manager_status(
264            prepared_query(
265                &source,
266                &mut self.statement_cache,
267                &mut self.raw_connection,
268                &mut *self.instrumentation,
269            )
270            .and_then(|stmt| {
271                let mut metadata = Vec::new();
272                DB::row_metadata(&mut (), &mut metadata);
273                StatementIterator::from_stmt(stmt, &metadata)
274            }),
275            &mut self.transaction_state,
276            &mut self.instrumentation,
277            &crate::debug_query(&source),
278        )
279    }
280}
281
282#[cfg(feature = "r2d2")]
283impl<DB: MysqlLikeBackend> crate::r2d2::R2D2Connection for MysqlLikeConnection<DB> {
284    fn ping(&mut self) -> QueryResult<()> {
285        crate::r2d2::CheckConnectionQuery.execute(self).map(|_| ())
286    }
287
288    fn is_broken(&mut self) -> bool {
289        AnsiTransactionManager::is_broken_transaction_manager(self)
290    }
291}
292
293impl<DB: MysqlLikeBackend> MultiConnectionHelper for MysqlLikeConnection<DB> {
294    fn to_any<'a>(
295        lookup: &mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup,
296    ) -> &mut (dyn core::any::Any + 'a) {
297        lookup
298    }
299
300    fn from_any(
301        lookup: &mut dyn core::any::Any,
302    ) -> Option<&mut <Self::Backend as crate::sql_types::TypeMetadata>::MetadataLookup> {
303        lookup.downcast_mut()
304    }
305}
306
307fn prepared_query<'a, DB: MysqlLikeBackend + Default, T: QueryFragment<DB> + QueryId>(
308    source: &'_ T,
309    statement_cache: &'a mut StatementCache<DB, Statement<DB>>,
310    raw_connection: &'a mut RawConnection,
311    instrumentation: &mut dyn Instrumentation,
312) -> QueryResult<MaybeCached<'a, Statement<DB>>> {
313    instrumentation.on_connection_event(InstrumentationEvent::StartQuery {
314        query: &crate::debug_query(source),
315    });
316    let mut stmt = statement_cache.cached_statement(
317        source,
318        &DB::default(),
319        &[],
320        &*raw_connection,
321        RawConnection::prepare,
322        instrumentation,
323    )?;
324
325    let mut bind_collector = RawBytesBindCollector::new();
326    source.collect_binds(&mut bind_collector, &mut (), &DB::default())?;
327    let binds = bind_collector
328        .metadata
329        .into_iter()
330        .zip(bind_collector.binds);
331    stmt.bind(binds)?;
332    Ok(stmt)
333}
334
335impl<DB: MysqlLikeBackend> MysqlLikeConnection<DB> {
336    /// Executes `source` and returns its `mysql_stmt_insert_id`, zero mapped
337    /// to `None`. Public entry point: [`InsertStatement::execute_returning_id`].
338    pub(crate) fn execute_returning_id<T>(&mut self, source: &T) -> QueryResult<Option<NonZeroU64>>
339    where
340        T: QueryFragment<DB> + QueryId,
341    {
342        #[allow(unsafe_code)] // call to unsafe function
343        update_transaction_manager_status(
344            prepared_query(
345                &source,
346                &mut self.statement_cache,
347                &mut self.raw_connection,
348                &mut *self.instrumentation,
349            )
350            .and_then(|stmt| {
351                // SAFETY: `prepared_query` returned this statement freshly
352                // bound, so no result set is pending, which is `execute`'s
353                // requirement.
354                let stmt_use = unsafe { stmt.execute() }?;
355                Ok(NonZeroU64::new(stmt_use.insert_id()))
356            }),
357            &mut self.transaction_state,
358            &mut self.instrumentation,
359            &crate::debug_query(source),
360        )
361    }
362
363    fn set_config_options(&mut self) -> QueryResult<()> {
364        crate::sql_query("SET time_zone = '+00:00';").execute(self)?;
365        crate::sql_query("SET character_set_client = 'utf8mb4'").execute(self)?;
366        crate::sql_query("SET character_set_connection = 'utf8mb4'").execute(self)?;
367        crate::sql_query("SET character_set_results = 'utf8mb4'").execute(self)?;
368        Ok(())
369    }
370
371    fn establish_inner(database_url: &str) -> Result<MysqlLikeConnection<DB>, ConnectionError> {
372        use crate::ConnectionError::CouldntSetupConfiguration;
373
374        let raw_connection = RawConnection::new();
375        let connection_options = ConnectionOptions::parse::<DB>(database_url)?;
376        raw_connection.connect(&connection_options)?;
377        let mut conn = MysqlLikeConnection {
378            raw_connection,
379            transaction_state: AnsiTransactionManager::default(),
380            statement_cache: StatementCache::new(),
381            instrumentation: DynInstrumentation::none(),
382        };
383        conn.set_config_options()
384            .map_err(CouldntSetupConfiguration)?;
385        Ok(conn)
386    }
387}
diesel/mysql_like/connection/mod.rs

diesel/mysql_like/connection/
mod.rs
