cdricms/vnidrop
1
1
Fork 0
mirror of https://github.com/sudosylabs/vnidrop.git synced 2026-08-05 02:29:55 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(send): show live progress per receiver

Browse Source 
Attribute provider transfer progress to each remote endpoint so the
receivers panel and catalog can show per-peer send bars for multi-file
collections while a share is active.
This commit is contained in:
Hammed Abass
2026-07-12 18:16:59 +02:00
parent 50d43a3867
commit 07ec727c23
7 changed files with 383 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
crates/vnidrop/src/runtime.rs
View File

@@ -1485,7 +1485,8 @@ impl CoreInner {
message.inner.connection_id,
message.inner.request_id,
message.rx,
);
)
.await;
}
let _ = message.tx.send(Ok(())).await;
}
@@ -1497,7 +1498,8 @@ impl CoreInner {
message.inner.connection_id,
message.inner.request_id,
message.rx,
);
)
.await;
}
}
ProviderMessage::GetManyRequestReceived(message) => {
@@ -1531,7 +1533,8 @@ impl CoreInner {
message.inner.connection_id,
message.inner.request_id,
message.rx,
);
)
.await;
}
let _ = message.tx.send(Ok(())).await;
}
@@ -1545,7 +1548,8 @@ impl CoreInner {
message.inner.connection_id,
message.inner.request_id,
message.rx,
);
)
.await;
}
}
ProviderMessage::ObserveRequestReceived(message) => {
@@ -1618,7 +1622,7 @@ impl CoreInner {
.await
}
fn track_request_updates(
async fn track_request_updates(
self: &Arc<Self>,
transfer_id: u64,
connection_id: u64,
@@ -1628,6 +1632,15 @@ impl CoreInner {
// Request update tasks are tied to individual provider streams. Router
// shutdown closes those streams; only the long-lived provider receiver
// is tracked directly for explicit shutdown.
//
// Attach the remote endpoint id when known so the send UI can attribute
// byte progress to a specific receiver (not just an opaque connection).
let endpoint_id = self
.connection_endpoints
.lock()
.await
.get(&connection_id)
.cloned();
let core = self.clone();
tokio::spawn(async move {
while let Ok(Some(update)) = rx.recv().await {
@@ -1640,6 +1653,7 @@ impl CoreInner {
json!({
"connection_id": connection_id,
"request_id": request_id,
"endpoint_id": endpoint_id,
"hash": started.hash.to_string(),
"size": started.size,
"index": started.index,
@@ -1653,6 +1667,7 @@ impl CoreInner {
json!({
"connection_id": connection_id,
"request_id": request_id,
"endpoint_id": endpoint_id,
"end_offset": progress.end_offset,
}),
),
@@ -1661,14 +1676,22 @@ impl CoreInner {
"send",
"transfer",
"completed",
json!({ "connection_id": connection_id, "request_id": request_id }),
json!({
"connection_id": connection_id,
"request_id": request_id,
"endpoint_id": endpoint_id,
}),
),
RequestUpdate::Aborted(_) => core.emit_transfer(
transfer_id,
"send",
"transfer",
"aborted",
json!({ "connection_id": connection_id, "request_id": request_id }),
json!({
"connection_id": connection_id,
"request_id": request_id,
"endpoint_id": endpoint_id,
}),
),
}
}