用通道+每对等点任务模式实现可靠的节点通信：告别消息丢失的烦恼

use tokio::sync::mpsc::{self, Sender};
use tokio_tungstenite::tungstenite::Message;
use futures_util::SinkExt;

// 对等点连接结构体，封装消息发送能力
#[derive(Clone)]
pub struct PeerConnection {
    sender: Sender<String>, // 用于发送JSON格式消息的通道发送端
}

impl PeerConnection {
    pub async fn send(&self, msg: String) -> Result<(), SendError<String>> {
        self.sender.send(msg).await.map_err(|e| e.0)
    }
}

pub async fn spawn_peer_task(
    mut ws_sink: tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
) -> PeerConnection {
    let (tx, mut rx) = mpsc::channel::<String>(64);
    // 创建缓冲大小为64的消息通道

    // 为每个对等点连接生成独立处理任务
    tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            // 发送消息到WebSocket连接
            match ws_sink.send(Message::Text(msg)).await {
                Ok(_) => {}
                Err(e) => {
                    eprintln!("发送消息失败: {}", e);
                    break; // 连接异常，终止任务
                }
            }
        }
        // 清理资源，关闭连接
        let _ = ws_sink.close().await;
    });

    PeerConnection { sender: tx }
}

use std::collections::HashMap;
use std::sync::{Arc, RwLock};

// 集群客户端管理所有对等点连接
#[derive(Clone)]
pub struct ClusterClient {
    peers: Arc<RwLock<HashMap<String, PeerConnection>>>,
    self_id: String, // 自己的标识
}

impl ClusterClient {
    pub fn new(self_id: &str) -> Self {
        Self {
            peers: Arc::new(RwLock::new(HashMap::new())),
            self_id: self_id.to_string(),
        }
    }

    // 连接到远程节点并建立通信通道
    pub async fn connect_to(&self, node_id: &str, url: &str) {
        if let Ok((ws_stream, _)) = connect_async(url).await {
            println!("成功连接到 {} @ {}", node_id, url);
            // 创建对等点连接处理任务
            let peer = spawn_peer_task(ws_stream).await;
            // 存储连接信息
            self.peers.write().unwrap().insert(node_id.to_string(), peer);
        } else {
            println!("连接 {} 失败，可能对方不在线", node_id);
        }
    }
}

// 先定义一个网络消息结构体
#[derive(Serialize, Deserialize)]
struct NetworkMessage {
    to: String,      // 接收者
    from: String,    // 发送者
    payload: String, // 消息内容
}

impl ClusterClient {
    pub async fn send_to_remote(&self, address: &str, payload: &str) {
        // 解析地址格式：actor_name@node_id
        if let Some((actor_name, node_id)) = address.split_once('@') {
            // 查找目标节点的联系方式
            if let Some(peer) = self.peers.read().unwrap().get(node_id) {
                let msg = NetworkMessage {
                    to: actor_name.to_string(),
                    from: self.self_id.clone(),
                    payload: payload.to_string(),
                };
                // 序列化成JSON并发送
                let json = serde_json::to_string(&msg).unwrap();
                peer.send(json).await;
            } else {
                println!("🚫 没有到节点 {} 的连接，先connect_to一下吧", node_id);
            }
        } else {
            println!("地址格式不对，应该是：actor@node");
        }
    }
}

cluster.send_to_remote("printer@nodeA", "你好呀 👋").await;

let router = spawn_actor(JobRouter {
    cluster: cluster.clone()
});

router.send(Job::DispatchTo("printer@nodeA".into(), "data".into())).await;