C# (.NET) में रिएक्टिव प्रोग्रामिंग: Rx.NET और IObservable

C# (.NET) में रिएक्टिव प्रोग्रामिंग: Rx.NET के साथ पूर्ण गाइड

रिएक्टिव प्रोग्रामिंग एक डेवलपमेंट पैराडाइम है जो एसिंक्रोनस डेटा स्ट्रीम और परिवर्तनों के प्रसार पर केंद्रित है। C# (.NET) इकोसिस्टम में, इस दृष्टिकोण को लागू करने का मुख्य टूल Reactive Extensions (Rx.NET) लाइब्रेरी है। यह आपको समय के साथ बदलने वाले डेटा अनुक्रमों के साथ उसी आसानी से काम करने की अनुमति देता है जैसे आप LINQ के माध्यम से सामान्य कलेक्शन के साथ करते हैं।

इस लेख में, हम रिएक्टिव प्रोग्रामिंग की मुख्य अवधारणाओं को समझेंगे, Rx.NET के प्रमुख घटकों की जांच करेंगे, और PythonLib शैक्षिक साइट के लिए व्यावहारिक कोड उदाहरण प्रस्तुत करेंगे।

रिएक्टिव प्रोग्रामिंग की मुख्य अवधारणाएँ

Rx.NET के केंद्र में दो इंटरफ़ेस हैं:

• IObservable<T> — डेटा स्रोत (observable sequence) जो एलिमेंट, एरर या समाप्ति सिग्नल भेज सकता है।
• IObserver<T> — सब्सक्राइबर (observer) जो स्रोत से नोटिफिकेशन प्राप्त करता है: OnNext, OnError, OnCompleted।

मुख्य विचार: स्रोत से डेटा माँगने (pull-मॉडल) के बजाय, सब्सक्राइबर इवेंट पर प्रतिक्रिया करता है (push-मॉडल)। यह UI-इवेंट, नेटवर्क रिक्वेस्ट, सेंसर डेटा स्ट्रीम या किसी भी अन्य एसिंक्रोनस ऑपरेशन को संभालने के लिए विशेष रूप से उपयोगी है।

Rx.NET के साथ शुरुआत

अपने C# प्रोजेक्ट में Rx.NET का उपयोग करने के लिए NuGet पैकेज इंस्टॉल करें:

dotnet add package System.Reactive

उदाहरण 1: Observable बनाना और सब्सक्राइब करना

आइए एक सरल संख्या अनुक्रम बनाएं जो देरी से मान उत्पन्न करता है:

using System;
using System.Reactive.Linq;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        var observable = Observable
            .Interval(TimeSpan.FromSeconds(1)) // हर सेकंड 0, 1, 2... उत्पन्न करता है
            .Take(5); // केवल पहले 5 नंबर लें

        var subscription = observable.Subscribe(
            onNext: number => Console.WriteLine($"प्राप्त: {number}"),
            onError: error => Console.WriteLine($"त्रुटि: {error.Message}"),
            onCompleted: () => Console.WriteLine("स्ट्रीम समाप्त")
        );

        // स्ट्रीम समाप्त होने की प्रतीक्षा करें
        await Task.Delay(6000);
        subscription.Dispose(); // सब्सक्रिप्शन खत्म करें
    }
}

उदाहरण 2: LINQ के साथ फ़िल्टरिंग और ट्रांसफ़ॉर्मेशन

Rx.NET सभी मानक LINQ ऑपरेटरों को सपोर्ट करता है, जो कोड को संक्षिप्त बनाता है:

var values = Observable.Range(1, 10);

var processed = values
    .Where(x => x % 2 == 0) // केवल सम संख्याएँ
    .Select(x => x * 10)    // 10 से गुणा करें
    .Timestamp();           // टाइमस्टैम्प जोड़ें

processed.Subscribe(x => Console.WriteLine($"[{x.Timestamp}] मान: {x.Value}"));

Rx.NET की उन्नत सुविधाएँ

स्ट्रीम का विलय

आप कई observable अनुक्रमों को जोड़ सकते हैं:

var stream1 = Observable.Interval(TimeSpan.FromSeconds(1)).Select(x => $"A{x}");
var stream2 = Observable.Interval(TimeSpan.FromSeconds(1.5)).Select(x => $"B{x}");

var merged = stream1.Merge(stream2); // एक स्ट्रीम में विलय
merged.Take(6).Subscribe(Console.WriteLine);

मल्टीथ्रेडिंग प्रबंधन

Rx.NET आसानी से निष्पादन संदर्भ बदलने की अनुमति देता है:

var uiScheduler = new SynchronizationContextScheduler(SynchronizationContext.Current);

Observable
    .Interval(TimeSpan.FromSeconds(1))
    .ObserveOn(uiScheduler) // UI थ्रेड में निष्पादन सुनिश्चित करें
    .Subscribe(x => UpdateUI(x)); // इंटरफ़ेस अपडेट करें

त्रुटि प्रबंधन

Rx.NET विफलताओं से उबरने के लिए शक्तिशाली तंत्र प्रदान करता है:

var faultyStream = Observable.Throw<int>(new Exception("त्रुटि!"));

var safeStream = faultyStream
    .Catch((Exception ex) => Observable.Return(-1)) // त्रुटि को -1 से बदलें
    .Subscribe(x => Console.WriteLine($"सुरक्षित मान: {x}"));