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    let client = Client::new()
28        .credentials(("john@example.org", "secret"))
29        .connect("https://jmap.example.org")
30        .await
31        .unwrap();
32
33    // Connect to the WebSocket endpoint
34    let mut ws_stream = client.connect_ws().await.unwrap();
35
36    // Read WS messages on a separate thread
37    let (stream_tx, mut stream_rx) = mpsc::channel::<WebSocketMessage>(100);
38    tokio::spawn(async move {
39        while let Some(change) = ws_stream.next().await {
40            stream_tx.send(change.unwrap()).await.unwrap();
41        }
42    });
43
44    // Create a mailbox over WS
45    let mut request = client.build();
46    let create_id = request
47        .set_mailbox()
48        .create()
49        .name("WebSocket Test")
50        .create_id()
51        .unwrap();
52    let request_id = request.send_ws().await.unwrap();
53
54    // Read response from WS stream
55    let mailbox_id = if let Some(WebSocketMessage::Response(mut response)) = stream_rx.recv().await
56    {
57        assert_eq!(request_id, response.request_id().unwrap());
58        response
59            .pop_method_response()
60            .unwrap()
61            .unwrap_set_mailbox()
62            .unwrap()
63            .created(&create_id)
64            .unwrap()
65            .take_id()
66    } else {
67        unreachable!()
68    };
69
70    // Enable push notifications over WS
71    client
72        .enable_push_ws(None::<Vec<_>>, None::<&str>)
73        .await
74        .unwrap();
75
76    // Make changes over standard HTTP and expect a push notification via WS
77    client
78        .mailbox_update_sort_order(&mailbox_id, 1)
79        .await
80        .unwrap();
81    if let Some(WebSocketMessage::StateChange(changes)) = stream_rx.recv().await {
82        println!("Received changes: {:?}", changes);
83    } else {
84        unreachable!()
85    }
86}
87
88fn main() {
89    #[cfg(feature = "websockets")]
90    let _c = websocket();
91}
websocket/websocket.rs

websocket/
websocket.rs
