6
6
//!
7
7
//! * using a sync Connection implementation in async context
8
8
//! * using the same code base for async crates needing multiple backends
9
- use std:: error:: Error ;
10
9
use futures_util:: future:: BoxFuture ;
10
+ use std:: error:: Error ;
11
11
12
12
#[ cfg( feature = "sqlite" ) ]
13
13
mod sqlite;
@@ -71,7 +71,8 @@ pub trait SpawnBlocking {
71
71
/// # }
72
72
/// ```
73
73
#[ cfg( feature = "tokio" ) ]
74
- pub type SyncConnectionWrapper < C , B = self :: implementation:: Tokio > = self :: implementation:: SyncConnectionWrapper < C , B > ;
74
+ pub type SyncConnectionWrapper < C , B = self :: implementation:: Tokio > =
75
+ self :: implementation:: SyncConnectionWrapper < C , B > ;
75
76
76
77
/// A wrapper of a [`diesel::connection::Connection`] usable in async context.
77
78
///
@@ -106,7 +107,9 @@ mod implementation {
106
107
107
108
use super :: * ;
108
109
109
- fn from_spawn_blocking_error ( error : Box < dyn Error + Send + Sync + ' static > ) -> diesel:: result:: Error {
110
+ fn from_spawn_blocking_error (
111
+ error : Box < dyn Error + Send + Sync + ' static > ,
112
+ ) -> diesel:: result:: Error {
110
113
diesel:: result:: Error :: DatabaseError (
111
114
diesel:: result:: DatabaseErrorKind :: UnableToSendCommand ,
112
115
Box :: new ( error. to_string ( ) ) ,
@@ -149,18 +152,21 @@ mod implementation {
149
152
// SpawnBlocking bounds
150
153
S : SpawnBlocking + Send ,
151
154
{
152
- type LoadFuture < ' conn , ' query > = BoxFuture < ' query , QueryResult < Self :: Stream < ' conn , ' query > > > ;
155
+ type LoadFuture < ' conn , ' query > =
156
+ BoxFuture < ' query , QueryResult < Self :: Stream < ' conn , ' query > > > ;
153
157
type ExecuteFuture < ' conn , ' query > = BoxFuture < ' query , QueryResult < usize > > ;
154
158
type Stream < ' conn , ' query > = BoxStream < ' static , QueryResult < Self :: Row < ' conn , ' query > > > ;
155
159
type Row < ' conn , ' query > = O ;
156
160
type Backend = <C as Connection >:: Backend ;
157
- type TransactionManager = SyncTransactionManagerWrapper < <C as Connection >:: TransactionManager > ;
161
+ type TransactionManager =
162
+ SyncTransactionManagerWrapper < <C as Connection >:: TransactionManager > ;
158
163
159
164
async fn establish ( database_url : & str ) -> ConnectionResult < Self > {
160
165
let database_url = database_url. to_string ( ) ;
161
166
let mut runtime = S :: get_runtime ( ) ;
162
167
163
- runtime. spawn_blocking ( move || C :: establish ( & database_url) )
168
+ runtime
169
+ . spawn_blocking ( move || C :: establish ( & database_url) )
164
170
. await
165
171
. unwrap_or_else ( |e| Err ( diesel:: ConnectionError :: BadConnection ( e. to_string ( ) ) ) )
166
172
. map ( move |c| SyncConnectionWrapper :: with_runtime ( c, runtime) )
@@ -192,16 +198,22 @@ mod implementation {
192
198
. boxed ( )
193
199
}
194
200
195
- fn execute_returning_count < ' query , T > ( & mut self , source : T ) -> Self :: ExecuteFuture < ' _ , ' query >
201
+ fn execute_returning_count < ' query , T > (
202
+ & mut self ,
203
+ source : T ,
204
+ ) -> Self :: ExecuteFuture < ' _ , ' query >
196
205
where
197
206
T : QueryFragment < Self :: Backend > + QueryId ,
198
207
{
199
- self . execute_with_prepared_query ( source, |conn, query| conn. execute_returning_count ( & query) )
208
+ self . execute_with_prepared_query ( source, |conn, query| {
209
+ conn. execute_returning_count ( & query)
210
+ } )
200
211
}
201
212
202
213
fn transaction_state (
203
214
& mut self ,
204
- ) -> & mut <Self :: TransactionManager as TransactionManager < Self > >:: TransactionStateData {
215
+ ) -> & mut <Self :: TransactionManager as TransactionManager < Self > >:: TransactionStateData
216
+ {
205
217
self . exclusive_connection ( ) . transaction_state ( )
206
218
}
207
219
@@ -344,16 +356,17 @@ mod implementation {
344
356
S : SpawnBlocking ,
345
357
{
346
358
let inner = self . inner . clone ( ) ;
347
- self . runtime . spawn_blocking ( move || {
348
- let mut inner = inner. lock ( ) . unwrap_or_else ( |poison| {
349
- // try to be resilient by providing the guard
350
- inner. clear_poison ( ) ;
351
- poison. into_inner ( )
352
- } ) ;
353
- task ( & mut inner)
354
- } )
355
- . unwrap_or_else ( |err| QueryResult :: Err ( from_spawn_blocking_error ( err) ) )
356
- . boxed ( )
359
+ self . runtime
360
+ . spawn_blocking ( move || {
361
+ let mut inner = inner. lock ( ) . unwrap_or_else ( |poison| {
362
+ // try to be resilient by providing the guard
363
+ inner. clear_poison ( ) ;
364
+ poison. into_inner ( )
365
+ } ) ;
366
+ task ( & mut inner)
367
+ } )
368
+ . unwrap_or_else ( |err| QueryResult :: Err ( from_spawn_blocking_error ( err) ) )
369
+ . boxed ( )
357
370
}
358
371
359
372
fn execute_with_prepared_query < ' a , MD , Q , R > (
@@ -448,7 +461,7 @@ mod implementation {
448
461
#[ cfg( feature = "tokio" ) ]
449
462
pub enum Tokio {
450
463
Handle ( tokio:: runtime:: Handle ) ,
451
- Runtime ( tokio:: runtime:: Runtime )
464
+ Runtime ( tokio:: runtime:: Runtime ) ,
452
465
}
453
466
454
467
#[ cfg( feature = "tokio" ) ]
@@ -462,12 +475,10 @@ mod implementation {
462
475
{
463
476
let fut = match self {
464
477
Tokio :: Handle ( handle) => handle. spawn_blocking ( task) ,
465
- Tokio :: Runtime ( runtime) => runtime. spawn_blocking ( task)
478
+ Tokio :: Runtime ( runtime) => runtime. spawn_blocking ( task) ,
466
479
} ;
467
480
468
- fut
469
- . map_err ( |err| Box :: from ( err) )
470
- . boxed ( )
481
+ fut. map_err ( |err| Box :: from ( err) ) . boxed ( )
471
482
}
472
483
473
484
fn get_runtime ( ) -> Self {
0 commit comments