diff --git a/crates/event-notifier/examples/webhook.rs b/crates/event-notifier/examples/webhook.rs index c754f8223..4cdf02c66 100644 --- a/crates/event-notifier/examples/webhook.rs +++ b/crates/event-notifier/examples/webhook.rs @@ -37,7 +37,7 @@ async fn receive_webhook(Json(payload): Json) -> StatusCode { println!("current time:{:04}-{:02}-{:02} {:02}:{:02}:{:02}", year, month, day, hour, minute, second); println!( "received a webhook request time:{} content:\n {}", - seconds.to_string(), + seconds, serde_json::to_string_pretty(&payload).unwrap() ); StatusCode::OK @@ -66,10 +66,10 @@ fn convert_seconds_to_date(seconds: u64) -> (u32, u32, u32, u32, u32, u32) { // calculate month let days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]; - for m in 0..12 { - if total_seconds >= days_in_month[m] * seconds_per_day { + for m in &days_in_month { + if total_seconds >= m * seconds_per_day { month += 1; - total_seconds -= days_in_month[m] * seconds_per_day; + total_seconds -= m * seconds_per_day; } else { break; } diff --git a/crates/event-notifier/src/global.rs b/crates/event-notifier/src/global.rs index 25fadeeb4..19693532d 100644 --- a/crates/event-notifier/src/global.rs +++ b/crates/event-notifier/src/global.rs @@ -189,7 +189,7 @@ mod tests { let config = NotifierConfig::default(); let _ = initialize(config.clone()).await; // first initialization let result = initialize(config).await; // second initialization - assert!(!result.is_ok(), "Initialization should succeed"); + assert!(result.is_err(), "Initialization should succeed"); assert!(result.is_err(), "Re-initialization should fail"); } @@ -211,7 +211,7 @@ mod tests { ..Default::default() }; let result = initialize(config).await; - assert!(!result.is_err(), "Initialization with invalid config should fail"); + assert!(result.is_ok(), "Initialization with invalid config should fail"); assert!(is_initialized(), "System should not be marked as initialized after failure"); assert!(is_ready(), "System should not be marked as ready after failure"); } diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 95bedfc4a..c73d35c0d 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -218,7 +218,7 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { let tracer = tracer_provider.tracer(Cow::Borrowed(service_name).to_string()); // Configure registry to avoid repeated calls to filter methods - let _registry = tracing_subscriber::registry() + tracing_subscriber::registry() .with(filter) .with(ErrorLayer::default()) .with(if config.local_logging_enabled.unwrap_or(false) { diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 56f5959c5..247cadae3 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -53,7 +53,7 @@ impl Erasure { } } - #[tracing::instrument(level = "debug", skip(self, reader, writers))] + #[tracing::instrument(level = "info", skip(self, reader, writers))] pub async fn encode( self: Arc, mut reader: S, @@ -65,23 +65,16 @@ impl Erasure { where S: AsyncRead + Etag + Unpin + Send + 'static, { - // pin_mut!(body); - // let mut reader = tokio_util::io::StreamReader::new( - // body.map(|f| f.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))), - // ); - - let (tx, mut rx) = mpsc::channel(3); - let self_clone = self.clone(); + let (tx, mut rx) = mpsc::channel(5); let task = tokio::spawn(async move { - let mut total: usize = 0; let mut buf = Vec::new(); + let mut total: usize = 0; loop { - let mut blocks = >::new(); if total_size > 0 { let new_len = { let remain = total_size - total; - if remain > self_clone.block_size { - self_clone.block_size + if remain > self.block_size { + self.block_size } else { remain } @@ -104,11 +97,10 @@ impl Erasure { }; total += buf.len(); } - self_clone.clone().encode_data(&buf, &mut blocks)?; + let blocks = Arc::new(Box::pin(self.clone().encode_data(&buf)?)); let _ = tx.send(blocks).await; } - // let etag = reader.etag().await; - let etag = String::new(); + let etag = reader.etag().await; Ok((total, etag)) }); @@ -368,7 +360,7 @@ impl Erasure { } #[tracing::instrument(level = "info", skip_all, fields(data_len=data.len()))] - pub fn encode_data(self: Arc, data: &[u8], shards: &mut SmallVec<[Bytes; 16]>) -> Result<()> { + pub fn encode_data(self: Arc, data: &[u8]) -> Result> { let (shard_size, total_size) = self.need_size(data.len()); // 生成一个新的 所需的所有分片数据长度 @@ -390,14 +382,13 @@ impl Erasure { // 零拷贝分片,所有 shard 引用 data_buffer let mut data_buffer = data_buffer.freeze(); - shards.clear(); - shards.reserve(self.total_shard_count()); + let mut shards = Vec::with_capacity(self.total_shard_count()); for _ in 0..self.total_shard_count() { let shard = data_buffer.split_to(shard_size); shards.push(shard); } - Ok(()) + Ok(shards) } pub fn decode_data(&self, shards: &mut [Option>]) -> Result<()> {