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

use crate::{
    types::{Advice, CometdError, CometdResult, Data, InnerError, Message},
    CometdClient,
};
use serde::de::DeserializeOwned;
use serde_json::json;

impl CometdClient {
    /// Send connect request.
    ///
    /// # Example
    /// ```rust
    /// # use cometd_client::{CometdClientBuilder, types::CometdResult};
    /// # let client = CometdClientBuilder::new(&"http://[::1]:1025/".parse().unwrap()).build().unwrap();
    ///
    /// # async {
    ///     let data = client.connect::<serde_json::Value>().await?;
    /// #   CometdResult::Ok(())
    /// # };
    /// ```
    pub async fn connect<Msg>(&self) -> CometdResult<Vec<Data<Msg>>>
    where
        Msg: DeserializeOwned,
    {
        let client_id = self
            .client_id
            .load_full()
            .ok_or_else(|| CometdError::connect_error(None, InnerError::MissingClientId))?;
        let id = self.next_id();
        let body = json!([{
          "id": id,
          "channel": "/meta/connect",
          "connectionType": "long-polling",
          "clientId": client_id
        }])
        .to_string();

        let request_builder = self.create_request_builder(&self.connect_endpoint);
        let raw_body = self
            .send_request(request_builder, body, |err| {
                CometdError::connect_error(None, err)
            })
            .await?;

        let mut messages = serde_json::from_slice::<Vec<Message>>(raw_body.as_ref())
            .map_err(|err| CometdError::connect_error(None, err))?;

        if let Some(position) = messages
            .iter()
            .position(|message| message.id.as_ref() == Some(&id))
        {
            let Message {
                successful,
                error,
                advice,
                ..
            } = messages.remove(position);

            if successful == Some(false) {
                Err(CometdError::connect_error(
                    Advice::reconnect(&advice),
                    InnerError::WrongResponse(error.unwrap_or_default().into()),
                ))
            } else {
                let data = messages
                    .into_iter()
                    .map(|message| {
                        let Message {
                            channel,
                            data,
                            advice,
                            ..
                        } = message;
                        let message = data
                            .map(serde_json::from_value::<Msg>)
                            .transpose()
                            .map_err(|err| {
                                CometdError::connect_error(Advice::reconnect(&advice), err)
                            })?;

                        Ok::<_, CometdError>(Data { channel, message })
                    })
                    .collect::<CometdResult<Vec<_>>>()?;

                Ok(data)
            }
        } else {
            Err(CometdError::connect_error(
                None,
                InnerError::WrongResponse(
                    "The response corresponding request id cannot be found.".into(),
                ),
            ))
        }
    }
}