chore: avoid .unwrap() calls

chainwayxyz · Jan 29, 2025 · 63439bb · 63439bb
1 parent d66338f
commit 63439bb
Showing 1 changed file with 9 additions and 21 deletions.
diff --git a/core/src/rpc/aggregator.rs b/core/src/rpc/aggregator.rs
@@ -19,6 +19,7 @@ use bitcoin::{Amount, TapSighash};
 use futures::{future::try_join_all, stream::BoxStream, FutureExt, Stream, StreamExt};
 use secp256k1::musig::{MusigAggNonce, MusigPartialSignature, MusigPubNonce};
 use tokio::sync::mpsc::{channel, Receiver, Sender};
+use tokio_stream::wrappers::ReceiverStream;
 use tonic::{async_trait, Request, Response, Status, Streaming};
 
 struct AggNonceQueueItem {
@@ -308,21 +309,8 @@ async fn pull<T>(mut stream: Streaming<T>) -> Result<Vec<T>, Status> {
     Ok(ret)
 }
 
-/*
-
-vp <- verifiers
-
-op <- operators
-wp <- watchtovers
-
-op -> verifiers
-wp -> verifiers
-
-*/
-
 #[async_trait]
 impl ClementineAggregator for Aggregator {
-    // TODO: HERE #464
     #[tracing::instrument(skip_all, err(level = tracing::Level::ERROR), ret(level = tracing::Level::TRACE))]
     async fn setup(&self, _request: Request<Empty>) -> Result<Response<Empty>, Status> {
         tracing::info!("Collecting verifier details...");
@@ -375,11 +363,11 @@ impl ClementineAggregator for Aggregator {
             let operator_params = operator_params.clone();
             async move {
                 for params in operator_params {
-                    let (tx, rx) = tokio::sync::mpsc::channel(params.len());
-                    let future =
-                        client.set_operator(tokio_stream::wrappers::ReceiverStream::new(rx));
+                    let (tx, rx) = channel(params.len());
+                    let future = client.set_operator(ReceiverStream::new(rx));
                     for param in params {
-                        tx.send(param).await.unwrap();
+                        tx.send(param).await
+                            .map_err(|e| Status::aborted(e.to_string()))?
                     }
                     future.await?;
                 }
@@ -394,11 +382,11 @@ impl ClementineAggregator for Aggregator {
             let watchtower_params = watchtower_params.clone();
             async move {
                 for params in watchtower_params {
-                    let (tx, rx) = tokio::sync::mpsc::channel(params.len());
-                    let future =
-                        client.set_watchtower(tokio_stream::wrappers::ReceiverStream::new(rx));
+                    let (tx, rx) = channel(params.len());
+                    let future = client.set_watchtower(ReceiverStream::new(rx));
                     for param in params {
-                        tx.send(param).await.unwrap();
+                        tx.send(param).await
+                            .map_err(|e| Status::aborted(e.to_string()))?
                     }
                     future.await?;
                 }