xlink 0.1.0

Unified Multi-Channel Communication SDK
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

#[cfg(test)]
mod tests {
    use std::process::{Command, Stdio};
    use std::sync::Arc;
    use uuid::Uuid;
    use xlink::core::traits::Storage;
    use xlink::core::types::{DeviceId, Message, MessagePayload, MessagePriority};
    use xlink::storage::memory_store::MemoryStorage;

    async fn get_memory_usage() -> Result<u64, Box<dyn std::error::Error + Send + Sync>> {
        let output = Command::new("ps")
            .args(["-o", "rss=", "-p", &std::process::id().to_string()])
            .stdout(Stdio::piped())
            .output()?;

        let memory_kb = String::from_utf8(output.stdout)?.trim().parse::<u64>()?;

        Ok(memory_kb)
    }

    #[tokio::test]
    async fn test_memory_storage_cleanup() {
        let initial_memory = get_memory_usage().await.unwrap();
        println!("Initial memory: {} KB", initial_memory);

        // Create and fill MemoryStorage similar to SDK usage
        {
            let storage = Arc::new(MemoryStorage::new());
            let device_id = DeviceId::new();

            // Fill with data similar to SDK usage
            for i in 0..1000 {
                let message = Message {
                    id: Uuid::new_v4(),
                    sender: device_id,
                    recipient: device_id,
                    group_id: None,
                    payload: MessagePayload::Text(format!("Message {}", i)),
                    priority: MessagePriority::Normal,
                    timestamp: std::time::SystemTime::now()
                        .duration_since(std::time::SystemTime::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs(),
                    require_ack: true,
                };
                storage.save_message(&message).await.unwrap();
                storage.save_pending_message(&message).await.unwrap();
                storage
                    .save_audit_log(format!("Audit log {}", i))
                    .await
                    .unwrap();
            }

            let after_fill = get_memory_usage().await.unwrap();
            println!(
                "After filling storage: {} KB (increase: {} KB)",
                after_fill,
                after_fill - initial_memory
            );

            // Test cleanup_storage method
            let removed = storage.cleanup_storage(1024).await.unwrap();
            println!("Cleanup removed {} bytes", removed);

            let after_cleanup = get_memory_usage().await.unwrap();
            let cleanup_change = after_cleanup.saturating_sub(after_fill);
            println!(
                "After cleanup: {} KB (change: {} KB)",
                after_cleanup, cleanup_change
            );

            // Drop the storage (should trigger Drop implementation)
            drop(storage);
        }

        // Force cleanup
        tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        {
            let _force_gc: Vec<u8> = vec![0; 1024 * 1024];
        }

        let final_memory = get_memory_usage().await.unwrap();
        let net_change = final_memory.saturating_sub(initial_memory);
        println!(
            "Final memory: {} KB (net change: {} KB)",
            final_memory, net_change
        );

        if net_change > 1024 {
            println!("✗ FAIL: MemoryStorage cleanup issue detected");
        } else {
            println!("✓ PASS: MemoryStorage cleaned up properly");
        }
    }
}