let node_subscribe = warp::path!("watch" / "node").and(warp::ws()).map(
    move |ws: warp::ws::Ws| {
        let node_broker_clone = Arc::clone(&node_broker_clone);
        ws.on_upgrade(move |socket| async move {
            node_broker_clone.subscribe("node".to_string(), socket).await;
        })
    },
);

/// Finish the upgrade, passing a function to handle the `WebSocket`.
///
/// The passed function must return a `Future`.
pub fn on_upgrade<F, U>(self, func: F) -> impl Reply
where
    F: FnOnce(WebSocket) -> U + Send + 'static,
    U: Future<Output = ()> + Send + 'static,
{
    WsReply {
        ws: self,
        on_upgrade: func,
    }
}

pub async fn subscribe(&self, topic: Topic, socket: warp::ws::WebSocket) {
        let (ws_sender, mut ws_receiver) = socket.split();
        let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
        let subscriber_id = Uuid::new_v4();
?
        {
            let mut subs = self.subscribers.write().await;
            subs.entry(topic.clone()).or_default().insert(subscriber_id, tx);
        }
?
        let subscribers = Arc::clone(&self.subscribers);
        tokio::task::spawn(async move {
            while let Some(result) = ws_receiver.next().await {
                match result {
                    Ok(message) => {
                        // 處理有效的消息
                        if message.is_text() {
                            println!(
                                "Received message from client: {}",
                                message.to_str().unwrap()
                            );
                        }
                    }
                    Err(e) => {
                        // 處理錯誤
                        eprintln!("WebSocket error: {:?}", e);
                        break;
                    }
                }
            }
            println!("WebSocket connection closed");
            subscribers.write().await.get_mut(&topic).map(|subscribers| subscribers.remove(&subscriber_id));
        });
?
        tokio::task::spawn(async move {
            let mut sender = ws_sender;
?
            while let Some(msg) = rx.recv().await {
                let _ = sender.send(msg).await;
            }
        });
    }

let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
let subscriber_id = Uuid::new_v4();
?
{
    let mut subs = self.subscribers.write().await;
    subs.entry(topic.clone()).or_default().insert(subscriber_id, tx);
}

tokio::task::spawn(async move {
            let mut sender = ws_sender;
?
            while let Some(msg) = rx.recv().await {
                let _ = sender.send(msg).await;
            }
        });

let subscribers = Arc::clone(&self.subscribers);
    tokio::task::spawn(async move {
        while let Some(result) = ws_receiver.next().await {
            match result {
                Ok(message) => {
                    // 處理有效的消息
                    if message.is_text() {
                        println!(
                            "Received message from client: {}",
                            message.to_str().unwrap()
                        );
                    }
                }
                Err(e) => {
                    // 處理錯誤
                    eprintln!("WebSocket error: {:?}", e);
                    break;
                }
            }
        }
        println!("WebSocket connection closed");
        subscribers.write().await.get_mut(&topic).map(|subscribers| subscribers.remove(&subscriber_id));
    });

pub async fn produce_node_event(&self, event: WatchEvent<Node>) {
        self.node_broker.produce("node".to_string(), event).await;
    }
    pub async fn produce_task_event(&self, event: WatchEvent<Task>) {
        self.task_broker.produce("task".to_string(), event).await;
    }
    pub async fn produce_job_event(&self, event: WatchEvent<Job>) {
        self.job_broker.produce("job".to_string(), event).await;
    }

pub async fn produce(&self, topic: Topic, event: WatchEvent<R>) {
        if let Err(e) = self.event_sender.send((topic.clone(), event.clone())) {
            eprintln!("Failed to send event: {}", e);
        }
    }

fn start_event_dispatcher(broker: Arc<Self>, mut event_receiver: UnboundedReceiver<(Topic, WatchEvent<R>)>) {
        tokio::spawn(async move {
            while let Some((topic, event)) = event_receiver.recv().await {
                let event_json = serde_json::to_string(&event).unwrap();
                let subscribers_list;
                {
                    let subscribers = broker.subscribers.read().await;
                    subscribers_list = subscribers.get(&topic).cloned().unwrap_or_default();
                }
                let mut invalid_subscribers = vec![];
                for (id, ws_sender) in subscribers_list {
                    if ws_sender.send(warp::ws::Message::text(event_json.clone())).is_err() {
                        invalid_subscribers.push(id);
                    }
                }
                if !invalid_subscribers.is_empty() {
                    let mut subscribers = broker.subscribers.write().await;
                    if let Some(subscribers) = subscribers.get_mut(&topic) {
                        for id in invalid_subscribers {
                            subscribers.remove(&id);
                        }
                    }
                }
            }
        });
    }

pub async fn list_and_watch<R>(api_client: &ApiClient, resource_name: &str) -> impl Stream<Item = WatchEvent<R>>
where
    R: Resource + Clone + DeserializeOwned + 'static + Send,
{
    // 先通過 HTTP 獲取資源列表
    let initial_resources = get_resource_list::<R>(api_client).await;
    // 解析要連接WebSocket服務器的URL
    let url = Url::parse(&*format!("{}/{}", api_client.watch_url, resource_name)).expect("Invalid URL");
    // 連接到WebSocket服務器
    println!("watch url is {}", url);
    let (ws_stream, _) = connect_async(url).await.expect("Failed to connect");
    println!("Watch client connected");
    let (mut write, read) = ws_stream.split();
    let (tx, rx) = mpsc::unbounded_channel();
    // 先發(fā)送初始資源列表
    match initial_resources {
        Ok(res) => tx.send(WatchEvent::Restarted(res)).unwrap(),
        Err(e) => eprintln!("list resource failed, {}", e),
    };
    // 將 WebSocket 讀流轉換為消息事件流
    tokio::spawn(async move {
        read.for_each(|message| async {
            match message {
                Ok(msg) => {
                    if msg.is_text() {
                        let text = msg.to_text().unwrap();
                        match serde_json::from_str::<WatchEvent<R>>(text) {
                            Ok(event) => {
                                tx.send(event).unwrap();
                            }
                            Err(e) => {
                                eprintln!("Failed to parse message: {:?}", e);
                            }
                        }
                    }
                }
                Err(e) => {
                    eprintln!("Error receiving message: {:?}", e);
                }
            }
        }).await;
    });
    // 保持 WebSocket 連接活躍
    tokio::spawn(async move {
        loop {
            if let Err(e) = write.send(WatchMessage::Text(String::new())).await {
                eprintln!("Error sending ping: {:?}", e);
                break;
            }
            tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
        }
    });
    tokio_stream::wrappers::UnboundedReceiverStream::new(rx)
}