Skip to content

Commit 815e7f2

Browse files
committed
Simplify test node setup
1 parent 0f10a98 commit 815e7f2

File tree

1 file changed

+19
-51
lines changed

1 file changed

+19
-51
lines changed

src/rpc/client/blobs.rs

Lines changed: 19 additions & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -1005,12 +1005,14 @@ mod tests {
10051005
//! An iroh node that just has the blobs transport
10061006
use std::{path::Path, sync::Arc};
10071007

1008-
use iroh_net::{NodeAddr, NodeId};
1009-
use quic_rpc::transport::{Connector, Listener};
1008+
use iroh_net::{Endpoint, NodeAddr, NodeId};
1009+
use iroh_router::Router;
10101010
use tokio_util::task::AbortOnDropHandle;
10111011

10121012
use super::RpcService;
10131013
use crate::{
1014+
downloader::Downloader,
1015+
net_protocol::Blobs,
10141016
provider::{CustomEventSender, EventSender},
10151017
rpc::client::{blobs, tags},
10161018
util::local_pool::LocalPool,
@@ -1054,40 +1056,20 @@ mod tests {
10541056

10551057
/// Spawns the node
10561058
pub async fn spawn(self) -> anyhow::Result<Node> {
1057-
let (client, router, rpc_task, _local_pool) = self.setup_router().await?;
1058-
Ok(Node {
1059-
router,
1060-
client,
1061-
_rpc_task: AbortOnDropHandle::new(rpc_task),
1062-
_local_pool,
1063-
})
1064-
}
1065-
1066-
async fn setup_router(
1067-
self,
1068-
) -> anyhow::Result<(
1069-
RpcClient,
1070-
iroh_router::Router,
1071-
tokio::task::JoinHandle<()>,
1072-
LocalPool,
1073-
)> {
10741059
let store = self.store;
10751060
let events = self.events;
10761061
let endpoint = self
10771062
.endpoint
1078-
.unwrap_or_else(|| iroh_net::Endpoint::builder().discovery_n0())
1063+
.unwrap_or_else(|| Endpoint::builder().discovery_n0())
10791064
.bind()
10801065
.await?;
10811066
let local_pool = LocalPool::single();
1082-
let mut router = iroh_router::Router::builder(endpoint.clone());
1067+
let mut router = Router::builder(endpoint.clone());
10831068

10841069
// Setup blobs
1085-
let downloader = crate::downloader::Downloader::new(
1086-
store.clone(),
1087-
endpoint.clone(),
1088-
local_pool.handle().clone(),
1089-
);
1090-
let blobs = Arc::new(crate::net_protocol::Blobs::new_with_events(
1070+
let downloader =
1071+
Downloader::new(store.clone(), endpoint.clone(), local_pool.handle().clone());
1072+
let blobs = Arc::new(Blobs::new_with_events(
10911073
store.clone(),
10921074
local_pool.handle().clone(),
10931075
events,
@@ -1101,31 +1083,17 @@ mod tests {
11011083

11021084
// Setup RPC
11031085
let (internal_rpc, controller) = quic_rpc::transport::flume::channel(32);
1104-
let controller = controller.boxed();
1105-
let internal_rpc = internal_rpc.boxed();
1106-
let internal_rpc = quic_rpc::RpcServer::new(internal_rpc);
1107-
1108-
let rpc_server_task: tokio::task::JoinHandle<()> = tokio::task::spawn(async move {
1109-
loop {
1110-
let request = internal_rpc.accept().await;
1111-
match request {
1112-
Ok(accepting) => {
1113-
let blobs = blobs.clone();
1114-
tokio::task::spawn(async move {
1115-
let (msg, chan) = accepting.read_first().await.unwrap();
1116-
blobs.handle_rpc_request(msg, chan).await.unwrap();
1117-
});
1118-
}
1119-
Err(err) => {
1120-
tracing::warn!("rpc error: {:?}", err);
1121-
}
1122-
}
1123-
}
1086+
let internal_rpc = quic_rpc::RpcServer::new(internal_rpc).boxed();
1087+
let _rpc_task = internal_rpc.spawn_accept_loop(move |msg, chan| {
1088+
blobs.clone().handle_rpc_request(msg, chan)
11241089
});
1125-
1126-
let client = quic_rpc::RpcClient::new(controller);
1127-
1128-
Ok((client, router, rpc_server_task, local_pool))
1090+
let client = quic_rpc::RpcClient::new(controller).boxed();
1091+
Ok(Node {
1092+
router,
1093+
client,
1094+
_rpc_task,
1095+
_local_pool: local_pool,
1096+
})
11291097
}
11301098
}
11311099

0 commit comments

Comments
 (0)