websocket/
websocket.rs

1/*
2 * Copyright Stalwart Labs LLC See the COPYING
3 * file at the top-level directory of this distribution.
4 *
5 * Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 * https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 * file at the top-level directory of this distribution.
8 *
9 * Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
10 * https://www.apache.org/licenses/LICENSE-2.0> or the MIT license
11 * <LICENSE-MIT or https://opensource.org/licenses/MIT>, at your
12 * option. This file may not be copied, modified, or distributed
13 * except according to those terms.
14 */
15
16#[cfg(feature = "websockets")]
17use futures_util::StreamExt;
18#[cfg(feature = "websockets")]
19use jmap_client::{client::Client, client_ws::WebSocketMessage, core::set::SetObject};
20#[cfg(feature = "websockets")]
21use tokio::sync::mpsc;
22
23// Make sure the "websockets" feature is enabled!
24#[cfg(feature = "websockets")]
25async fn websocket() {
26    // Connect to the JMAP server using Basic authentication
27
28    use jmap_client::PushObject;
29
30    let client = Client::new()
31        .credentials(("john@example.org", "secret"))
32        .connect("https://jmap.example.org")
33        .await
34        .unwrap();
35
36    // Connect to the WebSocket endpoint
37    let mut ws_stream = client.connect_ws().await.unwrap();
38
39    // Read WS messages on a separate thread
40    let (stream_tx, mut stream_rx) = mpsc::channel::<WebSocketMessage>(100);
41    tokio::spawn(async move {
42        while let Some(change) = ws_stream.next().await {
43            stream_tx.send(change.unwrap()).await.unwrap();
44        }
45    });
46
47    // Create a mailbox over WS
48    let mut request = client.build();
49    let create_id = request
50        .set_mailbox()
51        .create()
52        .name("WebSocket Test")
53        .create_id()
54        .unwrap();
55    let request_id = request.send_ws().await.unwrap();
56
57    // Read response from WS stream
58    let mailbox_id = if let Some(WebSocketMessage::Response(mut response)) = stream_rx.recv().await
59    {
60        assert_eq!(request_id, response.request_id().unwrap());
61        response
62            .pop_method_response()
63            .unwrap()
64            .unwrap_set_mailbox()
65            .unwrap()
66            .created(&create_id)
67            .unwrap()
68            .take_id()
69    } else {
70        unreachable!()
71    };
72
73    // Enable push notifications over WS
74    client
75        .enable_push_ws(None::<Vec<_>>, None::<&str>)
76        .await
77        .unwrap();
78
79    // Make changes over standard HTTP and expect a push notification via WS
80    client
81        .mailbox_update_sort_order(&mailbox_id, 1)
82        .await
83        .unwrap();
84    if let Some(WebSocketMessage::PushNotification(PushObject::StateChange { changed })) =
85        stream_rx.recv().await
86    {
87        println!("Received changes: {:?}", changed);
88    } else {
89        unreachable!()
90    }
91}
92
93fn main() {
94    #[cfg(feature = "websockets")]
95    let _c = websocket();
96}
websocket/websocket.rs

websocket/
websocket.rs
