diesel/mysql_like/connection/
mod.rs

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

diesel/mysql_like/connection/
mod.rs
