openigtlink-rust 0.3.2

Rust implementation of the OpenIGTLink protocol for image-guided therapy
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

//! Automatic reconnection demonstration
//!
//! Shows how ReconnectClient automatically handles connection failures
//! and reconnects with exponential backoff.

use openigtlink_rust::{
    io::{builder::ClientBuilder, AsyncIgtlServer, ReconnectConfig},
    protocol::{message::IgtlMessage, types::StatusMessage},
};
use std::sync::Arc;
use tokio::sync::Notify;
use tokio::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Initialize logging
    tracing_subscriber::fmt()
        .with_env_filter("info")
        .with_target(false)
        .init();

    println!("=== Automatic Reconnection Demo ===\n");

    // Create notify for coordination
    let server_ready = Arc::new(Notify::new());
    let server_ready_clone = server_ready.clone();

    let server_shutdown = Arc::new(Notify::new());
    let server_shutdown_clone = server_shutdown.clone();

    // Spawn server task
    let server_handle = tokio::spawn(async move {
        for round in 1..=2 {
            println!("[Server] Round {}: Starting server...", round);

            let server = AsyncIgtlServer::bind("127.0.0.1:18946").await.unwrap();
            println!("[Server] Round {}: Server listening\n", round);

            // Notify client that server is ready
            server_ready_clone.notify_one();

            // Accept one connection and exchange one message
            let mut conn = server.accept().await.unwrap();
            println!("[Server] Round {}: Client connected", round);

            let msg: IgtlMessage<StatusMessage> = conn.receive().await.unwrap();
            println!(
                "[Server] Round {}: Received: {}",
                round, msg.content.status_string
            );

            let response = StatusMessage::ok(&format!("Response from round {}", round));
            let response_msg = IgtlMessage::new(response, "Server").unwrap();
            conn.send(&response_msg).await.unwrap();
            println!("[Server] Round {}: Sent response\n", round);

            // Drop connection to simulate server restart
            drop(conn);
            drop(server);

            if round < 2 {
                println!(
                    "[Server] Round {}: Shutting down (simulating failure)...\n",
                    round
                );
                tokio::time::sleep(Duration::from_millis(500)).await;
            } else {
                println!("[Server] Round {}: Completed\n", round);
                server_shutdown_clone.notify_one();
                break;
            }
        }
    });

    // Give server time to start
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Create reconnecting client
    let config = ReconnectConfig {
        max_attempts: Some(20),
        initial_delay: Duration::from_millis(200),
        max_delay: Duration::from_secs(5),
        backoff_multiplier: 1.5,
        use_jitter: true,
    };

    println!("[Client] Creating reconnecting client...");
    let mut client = ClientBuilder::new()
        .tcp("127.0.0.1:18946")
        .async_mode()
        .with_reconnect(config)
        .build()
        .await?;
    println!("[Client] Initial connection established\n");

    // Send messages through multiple server restarts
    for i in 1..=2 {
        println!("[Client] Sending message {}...", i);

        let status = StatusMessage::ok(&format!("Message {}", i));
        let msg = IgtlMessage::new(status, "ReconnectClient")?;

        // This will automatically reconnect if needed
        client.send(&msg).await?;
        println!("[Client] Message {} sent", i);

        let response: IgtlMessage<StatusMessage> = client.receive().await?;
        println!("[Client] Received: {}", response.content.status_string);
        println!(
            "[Client] Total reconnections so far: {}\n",
            client.reconnect_count()
        );

        tokio::time::sleep(Duration::from_millis(200)).await;
    }

    println!("=== Demo completed successfully ===");
    println!("\nReconnection statistics:");
    println!("  Total messages sent: 2");
    println!("  Total reconnections: {}", client.reconnect_count());
    println!("  Connection success rate: 100%");
    println!("\n✓ Client automatically recovered from server failure!");

    tokio::time::timeout(Duration::from_secs(2), server_shutdown.notified())
        .await
        .ok();
    let _ = tokio::time::timeout(Duration::from_secs(1), server_handle).await;

    Ok(())
}