calimero_network/handlers/commands/
query_blob.rs

1use actix::{Context, Handler, Message, ResponseFuture};
2use calimero_network_primitives::messages::QueryBlob;
3use eyre::eyre;
4use libp2p::kad::RecordKey;
5use tokio::sync::oneshot;
6use tracing::debug;
7
8use crate::NetworkManager;
9
10impl Handler<QueryBlob> for NetworkManager {
11    type Result = ResponseFuture<<QueryBlob as Message>::Result>;
12
13    fn handle(&mut self, request: QueryBlob, _ctx: &mut Context<Self>) -> Self::Result {
14        debug!(
15            blob_id = %request.blob_id,
16            context_id = ?request.context_id.as_ref().map(ToString::to_string),
17            "Querying DHT for blob"
18        );
19
20        let (sender, receiver) = oneshot::channel();
21
22        // Create search key based on context
23        let key = if let Some(context_id) = request.context_id {
24            // Search in specific context
25            let key = RecordKey::new(&[context_id.as_slice(), request.blob_id.as_slice()].concat());
26
27            debug!(
28                blob_id = %request.blob_id,
29                context_id = context_id.as_str(),
30                key_len = key.as_ref().len(),
31                "QUERY: searching for blob"
32            );
33
34            key
35        } else {
36            drop(sender.send(Err(eyre!("Global blob queries not yet supported"))));
37            return Box::pin(async { receiver.await.expect("Sender not to be dropped") });
38        };
39
40        // Start the query (get_record returns QueryId directly)
41        let query_id = self.swarm.behaviour_mut().kad.get_record(key);
42
43        // Store the query for completion handling
44        let _previous = self.pending_blob_queries.insert(query_id, sender);
45
46        Box::pin(async { receiver.await.expect("Sender not to be dropped") })
47    }
48}
calimero_network/handlers/commands/query_blob.rs

calimero_network/handlers/commands/
query_blob.rs
