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

\n\n

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

\n\n

ما هو RxPHP؟

RxPHP هو إصدار PHP من إطار العمل الشهير ReactiveX (Rx). يوفر أدوات لتكوين البرامج اللاتزامنية والموجهة بالأحداث باستخدام تسلسلات قابلة للملاحظة وعوامل استعلام شبيهة بـ LINQ.

\n\n

المفاهيم الأساسية لـ RxPHP:

Observable — مصدر بيانات يمكنه إصدار قيم أو أخطاء أو إشارة اكتمال.
Observer — مشترك يستجيب للأحداث من Observable.
Subscription — العلاقة بين Observable و Observer، مما يسمح بإلغاء الاشتراك.
Operators — دوال لتحويل وتصفية ودمج التدفقات (map, filter, merge, zip وغيرها).
Scheduler — إدارة التنفيذ (متزامن، غير متزامن، مع تأخير).

\n\n

تثبيت RxPHP

يتم التثبيت عبر Composer:

composer require reactivex/rxphp

\n\n

مثال 1: إنشاء Observable بسيط

لننشئ تدفقًا يصدر ثلاثة أرقام بالتسلسل ثم يكتمل:

<?php\nrequire 'vendor/autoload.php';\n\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([10, 20, 30]);\n\n$disposable = $observable->subscribe(\n    function ($value) {\n        echo "تم الاستلام: $value\\n";\n    },\n    function (\\Throwable $error) {\n        echo "خطأ: " . $error->getMessage() . "\\n";\n    },\n    function () {\n        echo "اكتمل!\\n";\n    }\n);\n

نتيجة التنفيذ:

تم الاستلام: 10\nتم الاستلام: 20\nتم الاستلام: 30\nاكتمل!

\n\n

مثال 2: استخدام عاملي map و filter

لنحول تدفق الأرقام: نفلتر الأرقام الزوجية ونضربها في 10:

<?php\nuse Rx\\Observable;\n\n$observable = Observable::fromArray([1, 2, 3, 4, 5])\n    ->filter(function ($value) {\n        return $value % 2 === 0;\n    })\n    ->map(function ($value) {\n        return $value * 10;\n    });\n\n$observable->subscribe(function ($value) {\n    echo "النتيجة: $value\\n";\n});\n

الإخراج:

النتيجة: 20\nالنتيجة: 40

\n\n

مثال 3: العمل اللاتزامني مع HttpClient

RxPHP مناسب تمامًا للعمل مع طلبات HTTP اللاتزامنية. نستخدم Observable::fromPromise مع مكتبة GuzzleHttp:

<?php\nuse Rx\\Observable;\nuse GuzzleHttp\\Client;\nuse GuzzleHttp\\Promise;\n\n$client = new Client();\n$promise1 = $client->getAsync('https://api.example.com/data1');\n$promise2 = $client->getAsync('https://api.example.com/data2');\n\nObservable::fromArray([$promise1, $promise2])\n    ->flatMap(function ($promise) {\n        return Observable::fromPromise($promise);\n    })\n    ->map(function ($response) {\n        return json_decode($response->getBody(), true);\n    })\n    ->subscribe(function ($data) {\n        print_r($data);\n    });\n

\n\n

مثال 4: دمج التدفقات (merge و zip)

غالبًا ما نحتاج لدمج عدة تدفقات في تدفق واحد. لننظر إلى عامليين:

\n\n

merge — يدمج التدفقات، مع تبادل عناصرها عند وصولها:

<?php\nuse Rx\\Observable;\n\n$stream1 = Observable::fromArray(['A1', 'A2', 'A3']);\n$stream2 = Observable::fromArray(['B1', 'B2']);\n\nObservable::merge([$stream1, $stream2])\n    ->subscribe(function ($value) {\n        echo "$value ";\n    });\n// الإخراج: A1 B1 A2 B2 A3\n

\n\n