البرمجة التفاعلية في JavaScript (Node.js): RxJS & Observables

البرمجة التفاعلية في JavaScript (Node.js) باستخدام مكتبة RxJS

البرمجة التفاعلية هي نموذج يركز على تدفقات البيانات غير المتزامنة ونشر التغييرات. في نظام Node.js البيئي، يتم تنفيذ هذا المفهوم من خلال مكتبة RxJS (الإضافات التفاعلية لـ JavaScript). بدلاً من العمل التقليدي مع دوال الاسترجاع أو الوعود، تقدم RxJS العمل مع Observable - وهي كائنات تمثل تدفق البيانات عبر الزمن.

المفاهيم الأساسية

تعتمد RxJS على ثلاثة كيانات رئيسية:

• Observable - مصدر البيانات الذي يمكنه توليد قيم بمرور الوقت (أحداث الماوس، طلبات HTTP، قراءة الملفات).
• Observer - المستهلك الذي يشترك في Observable ويتفاعل مع ثلاثة أنواع من الأحداث: next (قيمة جديدة)، error (خطأ)، complete (إنهاء التدفق).
• Operators - دوال نقية لتحويل وتصفية ودمج التدفقات (مثل map، filter، mergeMap).

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

للعمل مع RxJS في Node.js، قم بتثبيت الحزمة عبر npm:

npm install rxjs

الاستيراد في الملف:

const { Observable } = require('rxjs');const { map, filter } = require('rxjs/operators');

إنشاء Observable

مثال بسيط - إنشاء تدفق أرقام:

const { Observable } = require('rxjs');

const stream$ = new Observable(subscriber => {    subscriber.next(1);    subscriber.next(2);    subscriber.next(3);    subscriber.complete();});

// الاشتراكstream$.subscribe({    next: value => console.log('تم الاستلام:', value),    error: err => console.error('خطأ:', err),    complete: () => console.log('تم إنهاء التدفق')});

// الإخراج: تم الاستلام: 1, تم الاستلام: 2, تم الاستلام: 3, تم إنهاء التدفق

العوامل: map و filter

تسمح العوامل بمعالجة البيانات في التدفق دون إنشاء آثار جانبية. مثال مع تصفية الأرقام الزوجية وتحويلها:

const { of } = require('rxjs');const { map, filter } = require('rxjs/operators');

const numbers$ = of(1, 2, 3, 4, 5, 6);

numbers$.pipe(    filter(x => x % 2 === 0), // الأرقام الزوجية فقط    map(x => x * 10)          // الضرب في 10).subscribe(result => console.log(result));

// الإخراج: 20, 40, 60

العمل مع العمليات غير المتزامنة

RxJS مناسب تمامًا لمعالجة أحداث الإدخال/الإخراج. على سبيل المثال، قراءة ملف باستخدام from:

const { from } = require('rxjs');const { map } = require('rxjs/operators');const fs = require('fs');const path = require('path');

// قراءة الملف كـ Promiseconst readFilePromise = (filePath) => {    return new Promise((resolve, reject) => {        fs.readFile(filePath, 'utf8', (err, data) => {            if (err) reject(err);            else resolve(data);        });    });};

// إنشاء Observable من Promiseconst fileStream$ = from(readFilePromise('./data.txt'));

fileStream$.pipe(    map(content => content.toUpperCase())).subscribe({    next: data => console.log('محتوى الملف:', data),    error: err => console.error('خطأ في القراءة:', err)});

دمج التدفقات

إحدى نقاط القوة في RxJS هي القدرة على دمج تدفقات متعددة. مثال مع merge (دمج التدفقات):

const { interval, merge } = require('rxjs');const { take, map } = require('rxjs/operators');

// التدفق A: كل ثانية واحدةconst streamA$ = interval(1000).pipe(    take(3),    map(x => `A: ${x}`));

// التدفق B: كل 1.5 ثانيةconst streamB$ = interval(1500).pipe(    take(3),    map(x => `B: ${x}`));

// الدمج