webclient 0.1.1

webclient
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

use super::WsClient;
use crossbeam::queue::SegQueue;
use dashmap::DashMap;
use futures::future::Future;
use futures::task::Poll;
use std::sync::Arc;
use std::task::Waker;

#[derive(Clone)]
pub struct ClientCommand<T: Send + Sync + serde::ser::Serialize> {
    pub event_id: u64,
    pub queue: Arc<DashMap<u64, Option<T>>>,
    pub wake_mgt: WakerManager,
}

impl<T: Send + Sync + serde::ser::Serialize> ClientCommand<T> {
    pub async fn send_request(
        client: &WsClient,
        in_data: String,
        event_id: u64,
        timeout_seconds: u64,
        queue: &Arc<DashMap<u64, Option<T>>>,
        wake_mgt: &WakerManager,
    ) -> anyhow::Result<T> {
        queue.insert(event_id, None);
        // tracing::info!("send command");
        let send_status = client.send_text(in_data);
        if send_status.is_err() {
            queue.remove(&event_id);
            return Err(anyhow::anyhow!(
                "send socket data to client with error: {:?}",
                send_status.err()
            ));
        }

        let client = ClientCommand {
            event_id: event_id.clone(),
            queue: queue.clone(),
            wake_mgt: wake_mgt.clone(),
        };

        let resp =
            tokio::time::timeout(tokio::time::Duration::from_secs(timeout_seconds), client).await;
        match resp {
            Ok(resp) => return anyhow::Ok(resp),
            Err(_) => {
                queue.remove(&event_id);
                return Err(anyhow::anyhow!("timeout for waiting for client response"));
            }
        }
    }
}

impl<T: Send + Sync + serde::ser::Serialize> Future for ClientCommand<T> {
    type Output = T;

    fn poll(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Self::Output> {
        let Self {
            event_id,
            queue,
            wake_mgt,
        } = &mut *self;

        let value = queue.entry(event_id.clone());
        match value {
            dashmap::mapref::entry::Entry::Occupied(e) => {
                if e.get().is_some() {
                    let out_data = e.remove().unwrap();
                    return Poll::Ready(out_data);
                }
            }
            dashmap::mapref::entry::Entry::Vacant(_) => {}
        }

        wake_mgt.wakers.push(cx.waker().clone());
        return Poll::Pending;
    }
}

#[derive(Default, Clone)]
pub struct WakerManager {
    pub wakers: Arc<SegQueue<Waker>>,
}

impl WakerManager {
    pub fn start(&self, duration_millis: u64) {
        let copied_manager = self.clone();
        tokio::spawn(async move {
            let manager = copied_manager;
            loop {
                // std::thread::sleep(dur);
                tokio::time::sleep(tokio::time::Duration::from_millis(duration_millis)).await;
                let queue_length = manager.wakers.len();
                for _i in 0..queue_length {
                    if let Some(item) = manager.wakers.pop() {
                        item.wake();
                    }
                }
            }
        });
    }
}