1. ما هي Flume ولماذا تحتاجها؟

مكتبة Flume هي تطبيق سريع وآمن وسهل الاستخدام لقنوات تعدد المهام (channels) في لغة Rust. توفر قوائم انتظار غير متزامنة ومتزامنة لنقل الرسائل بين الخيوط، مستوحاة من القنوات الكلاسيكية في Go ومكتبة crossbeam. تتميز Flume بأقل عدد من عمليات تخصيص الذاكرة، وعدم الاعتماد على المكتبة القياسية (توافق no_std)، وأداء ممتاز غالبًا ما يتفوق على قنوات mpsc القياسية من std::sync.

لماذا تحتاج Flume؟ في تطبيقات Rust متعددة الخيوط، غالبًا ما تحتاج إلى تنظيم تبادل البيانات بين الخيوط: خيط ينتج البيانات (منتج) وآخر يستهلكها (مستهلك). القنوات القياسية من std::sync::mpsc لها قيود: فهي تدعم مستهلكًا واحدًا فقط (single-consumer). بينما توفر Flume قنوات بمستهلك واحد (SPSC) وبعدة مستهلكين (MPSC)، كما تدعم البرمجة غير المتزامنة عبر async/await. تُستخدم المكتبة بنشاط في الأنظمة عالية التحميل، وخوادم الويب، ومحركات الألعاب بفضل خفتها وسرعتها.

الميزات الرئيسية لـ Flume: تجريدات بتكلفة صفرية (zero-cost abstractions)، دعم الضغط العكسي (backpressure) للحد من حجم المخزن المؤقت، إمكانية استنساخ المرسلين والمستقبلين، والتوافق الكامل مع tokio و async-std. إذا كنت بحاجة إلى قنوات موثوقة وسريعة لـ Rust، فإن Flume ستكون اختيارًا ممتازًا.

2. التثبيت

لتثبيت Flume، أضف التبعية إلى ملف Cargo.toml الخاص بمشروعك:

[dependencies]
flume = "0.11"

ثم نفّذ الأمر cargo build لتحميل المكتبة وترجمتها. إذا كنت تستخدم الإمكانيات غير المتزامنة، فلا حاجة لتبعيات إضافية - Flume تدعم async مباشرة.

3. البداية السريعة - مثال عملي بسيط

لننشئ قناة بسيطة، نرسل رسالة واحدة ونستقبلها:

use flume;

fn main() {
    // إنشاء قناة غير محدودة
    let (tx, rx) = flume::unbounded();

    // إرسال رسالة
    tx.send("مرحباً من Flume!").unwrap();

    // استقبال الرسالة
    let msg = rx.recv().unwrap();
    println!("تم الاستلام: {}", msg);
}

يوضح هذا المثال العمل الأساسي: unbounded() تنشئ قناة بدون حد لعدد الرسائل، send ترسل البيانات، و recv تحظر الخيط حتى استلام الرسالة.

4. أهم الدوال والطرق

4.1. flume::unbounded()

التوقيع: pub fn unbounded<T>() -> (Sender<T>, Receiver<T>)

تنشئ قناة غير محدودة. يمكن استنساخ المرسل والمستقبل. تُخزّن الرسائل في مخزن مؤقت ينمو ديناميكيًا.

let (tx, rx) = flume::unbounded::<String>();
tx.send("مرحباً".to_string()).unwrap();
let msg = rx.recv().unwrap();
println!("{}", msg);

4.2. flume::bounded(cap: usize)

التوقيع: pub fn bounded<T>(cap: usize) -> (Sender<T>, Receiver<T>)

تنشئ قناة بمخزن مؤقت محدود السعة. عند امتلاء المخزن، سيتم حظر send حتى يتوفر مكان.

let (tx, rx) = flume::bounded(3);
for i in 0..3 {
    tx.send(i).unwrap();
}
// الإرسال الرابع سيتم حظره حتى يتوفر مكان
// rx.recv(); // أزل التعليق لتحرير مكان
println!("تم إرسال 3 رسائل");

4.3. tx.send(val: T)

التوقيع: pub fn send(&self, val: T) -> Result<(), SendError<T>>

ترسل قيمة عبر القناة. في القنوات المحدودة، قد يتم حظرها إذا كان المخزن ممتلئًا. تُرجع خطأ إذا كان المستقبل قد أُغلِق.

let (tx, rx) = flume::unbounded();
tx.send(42).unwrap();
drop(rx); // إغلاق المستقبل
// tx.send(100).unwrap_err(); // سيفشل لأن المستقبل مغلق

4.4. rx.recv()

التوقيع: pub fn recv(&self) -> Result<T, RecvError>

تستقبل قيمة من القناة. تحظر الخيط الحالي حتى تتوفر رسالة أو يُغلَق القناة.

let (tx, rx) = flume::bounded(1);
tx.send("بيانات").unwrap();
let data = rx.recv().unwrap();
println!("استلمت: {}", data);

4.5. rx.try_recv()

التوقيع: pub fn try_recv(&self) -> Result<T, TryRecvError>

تحاول استقبال رسالة دون حظر. تُرجع خطأ Empty إذا لم تكن هناك رسائل، أو Disconnected إذا كانت القناة مغلقة.

let (tx, rx) = flume::unbounded();
match rx.try_recv() {
    Ok(msg) => println!("استلمت: {}", msg),
    Err(flume::TryRecvError::Empty) => println!("لا توجد رسائل"),
    Err(flume::TryRecvError::Disconnected) => println!("القناة مغلقة"),
}

4.6. rx.recv_async() / tx.send_async()

التوقيع: pub async fn recv_async(&self) -> Result<T, RecvError>

إصدارات غير متزامنة من recv و send للاستخدام مع async/await. مثالية للاستخدام مع tokio أو async-std.

async fn example() {
    let (tx, rx) = flume::unbounded();
    tx.send_async(99).await.unwrap();
    let val = rx.recv_async().await.unwrap();
    println!("قيمة غير متزامنة: {}", val);
}

4.7. tx.clone() / rx.clone()

يمكن استنساخ كل من المرسل والمستقبل. هذا يسمح بوجود عدة منتجين أو عدة مستهلكين.

let (tx, rx) = flume::unbounded();
let tx2 = tx.clone();
tx.send(1).unwrap();
tx2.send(2).unwrap();
println!("{}", rx.recv().unwrap()); // 1
println!("{}", rx.recv().unwrap()); // 2

4.8. rx.drain()

التوقيع: pub fn drain(&self) -> Drain<'_, T>

تُفرغ جميع الرسائل المعلقة من القناة وتُعيد مكررًا (iterator) عليها. مفيدة لتنظيف القناة.

let (tx, rx) = flume::unbounded();
tx.send(10).unwrap();
tx.send(20).unwrap();
for msg in rx.drain() {
    println!("تم التخلص من: {}", msg);
}

5. مثال متكامل: منتج ومستهلك

هذا مثال يوضح نمط المنتج-المستهلك باستخدام Flume مع خيوط متعددة:

use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = flume::bounded(5);

    // خيط المنتج
    let producer = thread::spawn(move || {
        for i in 1..=10 {
            tx.send(i).unwrap();
            println!("أرسل: {}", i);
            thread::sleep(Duration::from_millis(100));
        }
    });

    // خيط المستهلك
    let consumer = thread::spawn(move || {
        for _ in 1..=10 {
            let val = rx.recv().unwrap();
            println!("استلم: {}", val);
        }
    });

    producer.join().unwrap();
    consumer.join().unwrap();
}

يُظهر هذا المثال كيفية استخدام قناة محدودة لتنسيق العمل بين منتج ومستهلك في خيطين منفصلين.

6. نصائح وأفضل الممارسات

• اختيار نوع القناة: استخدم bounded عندما تريد التحكم في حجم المخزن وتطبيق الضغط العكسي. استخدم unbounded للبساطة عندما لا يكون حجم الرسائل مشكلة.
• إغلاق القنوات: عند إسقاط (drop) جميع المرسلين، يتم إغلاق القناة تلقائيًا. يمكنك استخدام tx.downgrade() لتحويل مرسل إلى مرسل ضعيف (weak sender).
• الأداء: Flume أسرع من std::sync::mpsc في معظم السيناريوهات. اختبر مع حالتك الخاصة.
• التوافق مع async: استخدم recv_async و send_async مع tokio أو async-std للحصول على أداء غير متزامن.