Contractless/src/torrent/torrenting_system/get_nodes.rs

use crate::common::binary_conversions::binary_to_ip;
use crate::records::memory::connections::CONNECTIONS;
use crate::records::memory::enums::ClientType;
use crate::Arc;
use crate::Mutex;
use crate::TcpStream;

pub async fn get_nodes_from_memory() -> Vec<(String, Arc<Mutex<TcpStream>>)> {
    // Snapshot the current connection manager so piece requests can iterate without holding the lock.
    let connection_storage = CONNECTIONS.read().await;
    let mut nodes = Vec::new();
    if let Some(connection) = &*connection_storage {
        for connection_info in connection.connection_map.values() {
            // Only fully initialized miner peers are expected to serve block pieces.
            if ClientType::from_bytes(&connection_info.client_type) != Some(ClientType::Miner) {
                continue;
            }
            if !connection_info.ready {
                continue;
            }
            // Use ip:port as the scheduler key and clone the shared stream handle for requests.
            let ip = binary_to_ip(connection_info.ip.clone());
            let port = connection_info.port;
            let key = format!("{ip}:{port}");
            let stream_arc = Arc::clone(&connection_info.stream);
            nodes.push((key, stream_arc));
        }
    }
    // Release the connection map before returning the cloned stream handles.
    drop(connection_storage);
    nodes
}

pub async fn get_torrent_broadcast_nodes_from_memory() -> Vec<(String, Arc<Mutex<TcpStream>>)> {
    // Torrent announcements are allowed to reach miner peers that are
    // still starting/syncing so they can stage new candidates while
    // catching up. Consensus/piece-selection paths must keep using
    // get_nodes_from_memory(), which requires ready peers.
    let connection_storage = CONNECTIONS.read().await;
    let mut nodes = Vec::new();
    if let Some(connection) = &*connection_storage {
        for connection_info in connection.connection_map.values() {
            if ClientType::from_bytes(&connection_info.client_type) != Some(ClientType::Miner) {
                continue;
            }
            let ip = binary_to_ip(connection_info.ip.clone());
            let port = connection_info.port;
            let key = format!("{ip}:{port}");
            let stream_arc = Arc::clone(&connection_info.stream);
            nodes.push((key, stream_arc));
        }
    }
    nodes
}