fibre 0.5.8

High-performance, safe, memory-efficient sync/async channels built for real-time, low-overhead communication in concurrent Rust applications.
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

use fibre::error::RecvError;
use fibre::spsc;
use std::thread;

mod common_async;

fn main() {
  let capacity = 2;

  println!("--- SPSC: Sync Sender, Sync Receiver ---");
  {
    let (tx, rx) = spsc::bounded_sync::<String>(capacity);

    let sender_handle = thread::spawn(move || {
      for i in 0..5 {
        let msg = format!("SyncSPSC-{}", i);
        println!("[Sync Send Thread] Sending: {}", msg);
        if tx.send(msg).is_err() {
          println!("[Sync Send Thread] Receiver dropped.");
          break;
        }
        thread::yield_now();
      }
      println!("[Sync Send Thread] Done sending.");
    });

    let receiver_handle = thread::spawn(move || {
      for i in 0..5 {
        match rx.recv() {
          Ok(value) => println!(
            "[Sync Recv Thread] Received: {} (expected SyncSPSC-{})",
            value, i
          ),
          Err(RecvError::Disconnected) => {
            println!("[Sync Recv Thread] Sender dropped.");
            break;
          }
        }
      }
      println!("[Sync Recv Thread] Done receiving.");
    });
    sender_handle.join().unwrap();
    receiver_handle.join().unwrap();
  }

  println!("\n--- SPSC: Async Sender, Async Receiver ---");
  common_async::run_async(async {
    let (tx, rx) = spsc::bounded_async::<String>(capacity);

    tokio::spawn(async move {
      // tx is moved here
      for i in 0..5 {
        let msg = format!("AsyncSPSC-{}", i);
        println!("[Async Send Task] Sending: {}", msg);
        if tx.send(msg).await.is_err() {
          println!("[Async Send Task] Receiver dropped.");
          break;
        }
        tokio::task::yield_now().await;
      }
      println!("[Async Send Task] Done sending.");
    });

    for i in 0..5 {
      match rx.recv().await {
        Ok(value) => println!(
          "[Async Recv Task] Received: {} (expected AsyncSPSC-{})",
          value, i
        ),
        Err(RecvError::Disconnected) => {
          println!("[Async Recv Task] Sender dropped.");
          break;
        }
      }
    }
    println!("[Async Recv Task] Done receiving.");
  });

  println!("\n--- SPSC: Sync Sender (Thread) to Async Receiver ---");
  common_async::run_async(async {
    // Start with a sync channel
    let (tx_s, rx_s) = spsc::bounded_sync::<String>(capacity);

    // Convert the receiver to async
    let rx_a = rx_s.to_async();
    // tx_s remains sync and is moved to the thread

    let sender_thread = thread::spawn(move || {
      // tx_s is moved here
      for i in 0..3 {
        let msg = format!("SyncToAsyncSPSC-{}", i);
        println!("[Sync Sender Thread] Sending: {}", msg);
        if tx_s.send(msg).is_err() {
          break;
        }
        thread::yield_now();
      }
      println!("[Sync Sender Thread] Done sending.");
      // tx_s drops here, signaling producer_dropped
    });

    println!("[Async Receiver Task] Waiting for messages...");
    for i in 0..3 {
      match rx_a.recv().await {
        Ok(val) => println!(
          "[Async Receiver Task] Received: {} (expected SyncToAsyncSPSC-{})",
          val, i
        ),
        Err(e) => {
          println!("[Async Receiver Task] Error: {:?}", e);
          break;
        }
      }
    }
    println!("[Async Receiver Task] Done receiving.");
    sender_thread.join().unwrap();
  });

  println!("\n--- SPSC: Async Sender to Sync Receiver ---");
  {
    // New scope for this test
    // Start with an async channel
    let (tx_a, rx_a) = spsc::bounded_async::<String>(capacity);

    // Convert the receiver to sync
    let rx_s = rx_a.to_sync();
    // tx_a remains async and is moved to the tokio task

    // Run the async producer in a separate OS thread with its own runtime
    let sender_os_thread_handle = thread::spawn(move || {
      // tx_a is moved here
      common_async::block_on_tokio_task(async move {
        // tx_a is moved into this async block
        for i in 0..3 {
          let msg = format!("AsyncToSyncSPSC-{}", i);
          println!("[Async Sender Task in OS Thread] Sending: {}", msg);
          if tx_a.send(msg).await.is_err() {
            break;
          }
          tokio::task::yield_now().await;
        }
        println!("[Async Sender Task in OS Thread] Done sending.");
        // tx_a drops here when the async block finishes
      });
    });

    println!("[Sync Receiver Thread] Waiting for messages...");
    for i in 0..3 {
      match rx_s.recv() {
        Ok(val) => println!(
          "[Sync Receiver Thread] Received: {} (expected AsyncToSyncSPSC-{})",
          val, i
        ),
        Err(e) => {
          println!("[Sync Receiver Thread] Error: {:?}", e);
          break;
        }
      }
    }
    println!("[Sync Receiver Thread] Done receiving.");
    sender_os_thread_handle.join().unwrap();
  }
  println!("\nSPSC examples finished.");
}