Skip to content

Commit 77f862e

Browse files
bordunospbordunosp
committed
test warn
1 parent 783268b commit 77f862e

10 files changed

Lines changed: 95 additions & 102 deletions

File tree

‎clippy.toml‎

Lines changed: 0 additions & 5 deletions
This file was deleted.

‎src/core/event_handlers.rs‎

Lines changed: 3 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -143,9 +143,7 @@ where
143143
}
144144

145145
while let Some(result) = futures.next().await {
146-
if let Err(e) = result {
147-
return Err(e);
148-
}
146+
result?;
149147
}
150148

151149
Ok(())
@@ -172,10 +170,8 @@ where
172170
})
173171
});
174172

175-
handlers
176-
.entry(event_type_name)
177-
.or_insert_with(Vec::new)
178-
.push(erased);
173+
handlers.entry(event_type_name).or_default().push(erased);
174+
179175
Ok(())
180176
}
181177

‎src/core/features/sea_orm/core.rs‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,12 @@
11
use crate::core::error_bus::BusError;
22
use crate::core::features::sea_orm::dto::{DatabaseConnectionDto, DatabaseQueueConfigurationDto};
3-
use sea_orm::{ConnectOptions, DbErr, Iden};
3+
use sea_orm::{ConnectOptions, DbErr};
44
use std::fmt;
55
use std::time::Duration;
66

7+
#[allow(unused_imports)]
8+
use sea_orm::Iden;
9+
710
#[derive(Clone, Debug, PartialEq)]
811
pub enum ArchiveType {
912
None,
@@ -55,6 +58,7 @@ pub struct DatabaseConnection {
5558
}
5659

5760
impl DatabaseConnection {
61+
#[allow(clippy::new_without_default)]
5862
pub fn new(dto: DatabaseConnectionDto) -> Result<Self, BusError> {
5963
if !dto
6064
.table_name

‎src/core/features/sea_orm/dto.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,18 @@ pub(crate) struct BusEvent {
4343

4444
pub(crate) struct BusEventJob {
4545
pub(crate) id: Uuid,
46+
47+
#[cfg_attr(test, allow(dead_code))]
4648
pub(crate) type_name_event: String,
4749
pub(crate) type_name_handler: String,
50+
51+
#[cfg_attr(test, allow(dead_code))]
4852
pub(crate) payload_bin: Vec<u8>,
4953
pub(crate) retries_current: i32,
5054
pub(crate) retries_max: i32,
5155
pub(crate) archive_mode: ArchiveType,
56+
57+
#[cfg_attr(test, allow(dead_code))]
5258
pub(crate) expires_interval: std::time::Duration,
5359
}
5460

‎src/core/features/sea_orm/event_database_pipeline.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ pub(crate) fn register_event_database_pipeline(
2222
.push(Arc::new(factory));
2323
}
2424

25+
#[cfg_attr(test, allow(dead_code))]
2526
pub(crate) struct DatabasePipelineWrapper {
2627
pub pipeline: Arc<dyn IEventDatabasePipeline>,
2728
pub next: Arc<dyn IErasedEventHandlerDatabase>,

‎src/core/features/sea_orm/event_handlers_database.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ use std::pin::Pin;
1919
use std::sync::Arc;
2020
use uuid::Uuid;
2121

22+
#[cfg_attr(test, allow(dead_code))]
2223
pub(crate) struct HandlerRegistration {
2324
pub(crate) factory: Arc<AsyncEventDbFactory>,
2425
pub(crate) settings: EventQueueSettings,

‎src/core/features/sea_orm/sql_mysql.rs‎

Lines changed: 27 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,8 @@ use std::error::Error;
1414
use uuid::Uuid;
1515

1616
pub(crate) async fn get_expires() -> Result<Vec<BusEventJob>, Box<dyn Error + Send + Sync>> {
17+
let table_name = get_queue_config()?.connection().table_name();
18+
1719
let query = Statement::from_sql_and_values(
1820
sea_orm::DatabaseBackend::MySql,
1921
format!(
@@ -26,8 +28,7 @@ pub(crate) async fn get_expires() -> Result<Vec<BusEventJob>, Box<dyn Error + Se
2628
AND expires_at > ?
2729
ORDER BY expires_at ASC
2830
LIMIT 10
29-
"#,
30-
table_name = get_queue_config()?.connection().table_name().to_string()
31+
"#
3132
),
3233
vec![
3334
Value::String(Some(Box::new(EventStatusEnum::Processing.to_string()))),
@@ -70,6 +71,8 @@ pub(crate) async fn get_expires() -> Result<Vec<BusEventJob>, Box<dyn Error + Se
7071
pub(crate) async fn get_id_status(
7172
status: EventStatusEnum,
7273
) -> Result<Vec<IdArchiveJob>, Box<dyn Error + Send + Sync>> {
74+
let table_name = get_queue_config()?.connection().table_name();
75+
7376
get_db_conn()?
7477
.query_all(Statement::from_sql_and_values(
7578
sea_orm::DatabaseBackend::MySql,
@@ -79,8 +82,7 @@ pub(crate) async fn get_id_status(
7982
FROM {table_name}
8083
WHERE status = ?
8184
LIMIT 10
82-
"#,
83-
table_name = get_queue_config()?.connection().table_name().to_string()
85+
"#
8486
),
8587
vec![Value::String(Some(Box::new(status.to_string())))],
8688
))
@@ -104,13 +106,12 @@ pub(crate) async fn get_id_status(
104106
}
105107

106108
pub(crate) async fn delete(id: Uuid) -> Result<(), Box<dyn Error + Send + Sync>> {
109+
let table_name = get_queue_config()?.connection().table_name();
110+
107111
get_db_conn()?
108112
.execute(Statement::from_sql_and_values(
109113
DbBackend::MySql,
110-
format!(
111-
r#"DELETE FROM {table_name} WHERE id = ?;"#,
112-
table_name = get_queue_config()?.connection().table_name().to_string()
113-
),
114+
format!(r#"DELETE FROM {table_name} WHERE id = ?;"#),
114115
vec![Value::Bytes(Some(Box::new(id.as_bytes().to_vec())))],
115116
))
116117
.await?;
@@ -122,6 +123,8 @@ pub(crate) async fn update_scheduled_at(
122123
scheduled_at: chrono::Duration,
123124
err: Box<dyn Error + Send + Sync>,
124125
) -> Result<(), Box<dyn Error + Send + Sync>> {
126+
let table_name = get_queue_config()?.connection().table_name();
127+
125128
get_db_conn()?
126129
.execute(Statement::from_sql_and_values(
127130
DbBackend::MySql,
@@ -135,8 +138,7 @@ pub(crate) async fn update_scheduled_at(
135138
should_start_at = ?,
136139
updated_at = (NOW() AT TIME ZONE 'UTC')
137140
WHERE id = ?
138-
"#,
139-
table_name = get_queue_config()?.connection().table_name().to_string()
141+
"#
140142
),
141143
vec![
142144
Value::String(Some(Box::new(EventStatusEnum::Pending.to_string()))),
@@ -154,11 +156,8 @@ pub(crate) async fn update_scheduled_at(
154156
pub(crate) async fn archive(id: Uuid) -> Result<(), BusError> {
155157
let db = get_db_conn()?;
156158
let txn = db.begin().await?;
157-
let table_name = get_queue_config()?.connection().table_name().to_string();
158-
let table_name_archive = get_queue_config()?
159-
.connection()
160-
.table_name_archive()
161-
.to_string();
159+
let table_name = get_queue_config()?.connection().table_name();
160+
let table_name_archive = get_queue_config()?.connection().table_name_archive();
162161

163162
let select_stmt = Statement::from_sql_and_values(
164163
DbBackend::MySql,
@@ -169,8 +168,7 @@ pub(crate) async fn archive(id: Uuid) -> Result<(), BusError> {
169168
status, payload_json, payload_bin, latest_error, updated_at
170169
FROM {table_name}
171170
WHERE id = ?
172-
"#,
173-
table_name = table_name
171+
"#
174172
),
175173
vec![Value::Bytes(Some(Box::new(id.as_bytes().to_vec())))],
176174
);
@@ -217,12 +215,11 @@ pub(crate) async fn archive(id: Uuid) -> Result<(), BusError> {
217215
DbBackend::MySql,
218216
format!(
219217
r#"
220-
INSERT INTO {table_name} (
218+
INSERT INTO {table_name_archive} (
221219
id, queue_name, type_name_event, type_name_handler,
222220
status, payload_json, payload_bin, latest_error, created_at
223221
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
224-
"#,
225-
table_name = table_name_archive
222+
"#
226223
),
227224
vec![
228225
Value::Bytes(Some(Box::new(event_archive.id.as_bytes().to_vec()))),
@@ -241,10 +238,7 @@ pub(crate) async fn archive(id: Uuid) -> Result<(), BusError> {
241238

242239
let delete_stmt = Statement::from_sql_and_values(
243240
DbBackend::MySql,
244-
format!(
245-
r#"DELETE FROM {table_name} WHERE id = $1"#,
246-
table_name = table_name
247-
),
241+
format!(r#"DELETE FROM {table_name} WHERE id = $1"#),
248242
vec![Value::Bytes(Some(Box::new(id.as_bytes().to_vec())))],
249243
);
250244

@@ -258,13 +252,14 @@ pub(crate) async fn change_status(
258252
status: EventStatusEnum,
259253
err: Option<Box<dyn Error + Send + Sync>>,
260254
) -> Result<(), BusError> {
255+
let table_name = get_queue_config()?.connection().table_name();
256+
261257
let mut sql = format!(
262258
r#"
263259
UPDATE {table_name}
264260
SET status = ?,
265261
updated_at = (NOW() AT TIME ZONE 'UTC')
266-
"#,
267-
table_name = get_queue_config()?.connection().table_name().to_string()
262+
"#
268263
);
269264

270265
let mut values: Vec<Value> = vec![Value::String(Some(Box::new(status.to_string())))];
@@ -290,7 +285,7 @@ pub(crate) async fn get_queues_items(
290285
) -> Result<Vec<BusEventJob>, BusError> {
291286
let now = Utc::now().naive_utc();
292287
let limit = queue_config.batch_size();
293-
let table_name = get_queue_config()?.connection().table_name().to_string();
288+
let table_name = get_queue_config()?.connection().table_name();
294289
let txn = get_db_conn()?.begin().await?;
295290

296291
let select_ids_stmt = Statement::from_sql_and_values(
@@ -305,8 +300,7 @@ pub(crate) async fn get_queues_items(
305300
ORDER BY should_start_at ASC
306301
LIMIT ?
307302
FOR UPDATE SKIP LOCKED
308-
"#,
309-
table_name = table_name
303+
"#
310304
),
311305
vec![
312306
Value::String(Some(Box::new(queue_config.queue_name().to_string()))),
@@ -352,9 +346,7 @@ pub(crate) async fn get_queues_items(
352346
expires_at = DATE_ADD(UTC_TIMESTAMP(), INTERVAL expires_interval SECOND),
353347
updated_at = UTC_TIMESTAMP()
354348
WHERE id IN ({placeholders})
355-
"#,
356-
table_name = table_name,
357-
placeholders = placeholders
349+
"#
358350
),
359351
update_values,
360352
);
@@ -383,9 +375,7 @@ pub(crate) async fn get_queues_items(
383375
expires_interval
384376
FROM {table_name}
385377
WHERE id IN ({placeholders})
386-
"#,
387-
table_name = table_name,
388-
placeholders = placeholders
378+
"#
389379
),
390380
select_values,
391381
);
@@ -425,6 +415,7 @@ pub(crate) async fn insert_to_events(
425415
bus_event: BusEvent,
426416
) -> Result<(), BusError> {
427417
let db_config = get_queue_config()?;
418+
let table_name = db_config.connection().table_name();
428419

429420
let payload_json_value = {
430421
#[cfg(feature = "json-payload")]
@@ -447,8 +438,7 @@ pub(crate) async fn insert_to_events(
447438
archive_mode, should_start_at, expires_at, expires_interval, created_at, updated_at
448439
)
449440
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
450-
"#,
451-
table_name = db_config.connection().table_name().to_string()
441+
"#
452442
),
453443
vec![
454444
Value::Bytes(Some(Box::new(bus_event.id.as_bytes().to_vec()))),

0 commit comments

Comments
 (0)