البرمجة التفاعلية في PHP: مكتبة RxPHP للتدفقات غير المتزامنة

البرمجة التفاعلية في PHP: دليل عملي لـ RxPHP

البرمجة التفاعلية هي نموذج يركز على العمل مع تدفقات البيانات غير المتزامنة. في PHP، وهي لغة متزامنة تقليديًا، يتم تنفيذ هذا المفهوم من خلال مكتبة RxPHP (ReactiveX لـ PHP). تتيح لك معالجة الأحداث، وطلبات HTTP، وتدفقات الملفات، وإجراءات المستخدم بأسلوب تصريحي موحد.

ما هو RxPHP؟

RxPHP هو إصدار من مكتبة ReactiveX الشهيرة. المكونات الرئيسية:

• Observable — مصدر البيانات الذي "يصدر" الأحداث (onNext، onError، onCompleted).
• Observer — المشترك الذي يستجيب للأحداث.
• Schedulers — يديرون سياق التنفيذ (متزامن، غير متزامن، على خيط جديد).
• العوامل (Operators) — دوال لتحويل وتصفية ودمج التدفقات (map، filter، merge، zip).

التثبيت والإعداد

للعمل مع RxPHP، استخدم Composer:

composer require reactivex/rxphp

مثال أساسي: إنشاء Observable والاشتراك:

<?phpuse Rx\\Observable;use Rx\\ObserverInterface;

$observable = Observable::fromArray([1, 2, 3, 4, 5]);

$observer = new class implements ObserverInterface {    public function onNext($value) {        echo "تم الاستلام: $value";    }    public function onError(\\Throwable $error) {        echo "خطأ: " . $error->getMessage() . "";    }    public function onCompleted() {        echo "اكتمل!";    }};

$observable->subscribe($observer);// الإخراج:// تم الاستلام: 1// تم الاستلام: 2// تم الاستلام: 3// تم الاستلام: 4// تم الاستلام: 5// اكتمل!?>

العوامل الأساسية في RxPHP

تسمح العوامل بتحويل البيانات دون تغيير Observable الأصلي:

1. map — تحويل كل عنصر

$observable = Observable::fromArray([1, 2, 3])    ->map(function($value) {        return $value * 2;    });

$observable->subscribe(function($value) {    echo $value . "";});// 2, 4, 6

2. filter — التصفية حسب الشرط

$observable = Observable::fromArray([1, 2, 3, 4, 5])    ->filter(function($value) {        return $value % 2 === 0;    });

// النتيجة: 2, 4

3. merge — دمج التدفقات

$stream1 = Observable::fromArray([1, 2]);$stream2 = Observable::fromArray([3, 4]);

$merged = Observable::merge($stream1, $stream2);// 1, 2, 3, 4 (قد يعتمد الترتيب على Scheduler)

إنشاء Observable مخصص

يمكن إنشاء Observable من أي مصادر: قواعد البيانات، المقابس (sockets)، أو الأحداث المخصصة.

<?phpuse Rx\\Observable;use Rx\\Scheduler\\EventLoopScheduler;

// محاكاة قراءة ملف غير متزامنةfunction readFileAsync(string $filename): Observable {    return Observable::create(function($observer) use ($filename) {        $handle = fopen($filename, 'r');        if (!$handle) {            $observer->onError(new \\RuntimeException("فشل فتح الملف"));            return;        }                while (($line = fgets($handle)) !== false) {            $observer->onNext(trim($line));        }        fclose($handle);        $observer->onCompleted();    });}

// الاستخدام مع EventLoop (يتطلب react/event-loop)$loop = \\React\\EventLoop\\Loop::get();$scheduler = new EventLoopScheduler($loop);

readFileAsync('data.log')    ->subscribeOn($scheduler)    ->filter(function($line) {        return strpos($line, 'ERROR') !== false;    })    ->subscribe(function($line) {        echo "تم العثور على خطأ: $line";    });

$loop->run();?>

معالجة الأخطاء والإكمال

في الأسلوب التفاعلي، تتم معالجة الأخطاء عن طريق