fibre 0.4.3

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

// examples/mpsc_examples.rs
use fibre::error::{RecvError, SendError};
use fibre::mpsc;
use std::{
  sync::atomic::{AtomicUsize, Ordering},
  sync::Arc,
  thread,
  time::Duration,
};

mod common_async;

fn main() {
  println!("--- MPSC: Sync Senders, Sync Receiver ---");
  {
    let (tx, mut rx) = mpsc::unbounded::<String>();
    let num_producers = 3;
    let messages_per_producer = 2;
    let total_messages = num_producers * messages_per_producer;
    let received_count = Arc::new(AtomicUsize::new(0));

    for i in 0..num_producers {
      let tx_clone = tx.clone();
      thread::spawn(move || {
        for j in 0..messages_per_producer {
          let msg = format!("SyncMPSC-P{}-M{}", i, j);
          println!("[Sync Sender {}] Sending: {}", i, msg);
          if tx_clone.send(msg).is_err() {
            println!("[Sync Sender {}] Receiver dropped.", i);
            break;
          }
          thread::sleep(Duration::from_millis(10 + i as u64 * 5));
        }
      });
    }
    drop(tx); // Drop original sender

    let recv_count_clone = Arc::clone(&received_count);
    let receiver_handle = thread::spawn(move || {
      for _ in 0..total_messages {
        match rx.recv() {
          Ok(value) => {
            println!("[Sync Receiver] Received: {}", value);
            recv_count_clone.fetch_add(1, Ordering::Relaxed);
          }
          Err(RecvError::Disconnected) => {
            println!("[Sync Receiver] All producers dropped.");
            break;
          }
        }
      }
    });
    receiver_handle.join().unwrap();
    assert_eq!(received_count.load(Ordering::Relaxed), total_messages);
  }

  println!("\n--- MPSC: Async Senders, Async Receiver ---");
  common_async::run_async(async {
    let (tx, mut rx) = mpsc::unbounded_async::<String>();
    let num_producers = 3;
    let messages_per_producer = 2;
    let total_messages = num_producers * messages_per_producer;
    let received_count = Arc::new(AtomicUsize::new(0));

    for i in 0..num_producers {
      let tx_clone = tx.clone();
      tokio::spawn(async move {
        for j in 0..messages_per_producer {
          let msg = format!("AsyncMPSC-P{}-M{}", i, j);
          println!("[Async Sender {}] Sending: {}", i, msg);
          if tx_clone.send(msg).await.is_err() {
            println!("[Async Sender {}] Receiver dropped.", i);
            break;
          }
          tokio::time::sleep(Duration::from_millis(10 + i as u64 * 5)).await;
        }
      });
    }
    drop(tx); // Drop original async sender

    let recv_count_clone = Arc::clone(&received_count);
    for _ in 0..total_messages {
      match rx.recv().await {
        Ok(value) => {
          println!("[Async Receiver] Received: {}", value);
          recv_count_clone.fetch_add(1, Ordering::Relaxed);
        }
        Err(RecvError::Disconnected) => {
          println!("[Async Receiver] All producers dropped.");
          break;
        }
      }
    }
    assert_eq!(received_count.load(Ordering::Relaxed), total_messages);
  });

  println!("\n--- MPSC: Sync Senders (Threads) to Async Receiver ---");
  common_async::run_async(async {
    let (tx_async, mut rx_async) = mpsc::unbounded_async::<String>(); // Start with async channel
    let num_producers = 2;

    for i in 0..num_producers {
      let tx_sync_converted = tx_async.clone().to_sync(); // Convert for each thread
      thread::spawn(move || {
        let msg = format!("SyncToAsyncMPSC-P{}", i);
        println!("[Sync Sender {}] Sending: {}", i, msg);
        if tx_sync_converted.send(msg).is_err() { /* ... */ }
      });
    }
    drop(tx_async); // Drop the original async sender last

    for _ in 0..num_producers {
      if let Ok(val) = rx_async.recv().await {
        println!("[Async Receiver] Received: {}", val);
      } else {
        break;
      }
    }
  });

  println!("\n--- MPSC: Async Senders to Sync Receiver ---");
  {
    let (tx_async, rx_async) = mpsc::unbounded_async::<String>();
    let mut rx_sync = rx_async.to_sync(); // Convert receiver
    let num_producers = 2;

    for i in 0..num_producers {
      let tx_clone = tx_async.clone();
      // Need to run these async senders in a runtime if the main thread is sync
      common_async::block_on_tokio_task(async move {
        let msg = format!("AsyncToSyncMPSC-P{}", i);
        println!("[Async Sender {}] Sending: {}", i, msg);
        if tx_clone.send(msg).await.is_err() { /* ... */ }
      });
    }
    drop(tx_async);

    for _ in 0..num_producers {
      if let Ok(val) = rx_sync.recv() {
        println!("[Sync Receiver] Received: {}", val);
      } else {
        break;
      }
    }
  }
}