کلیه حقوق مادی و معنوی این مقاله متعلق به آموزشگاه تحلیل داده می باشد و هر گونه استفاده غیر قانونی از آن پیگرد قانونی دارد.

شروع کار با Reactive Extension

در این بخش، درباره کتابخانه Rx صحبت خواهیم کرد. Rx یک کتابخانه، برای ترکیب برنامه های غیرهمزمان و مبتنی بر رویداد (event based) با استفاده از مجموعه های قابل مشاهده است. این موضوع زمانی مفید خواهد بود که بخواهید داده ها را از منابع مختلف به صورت غیر همزمان بردارید. در یک چنین سناریویی، باید یکسری glue code بنویسیم که نهایتاً بسیار مستعد خطا خواهد بود. فرض کنید یکی از منابع باعث بروز خطا شود، در این صورت چه اتفاقی می افتد؟

با این روش شما مجبور خواهید بود برای درست کار کردن برنامه، کارهای زیادی انجام دهید. از اینرو، Rx راه حلی برای این مشکل است. به این ترتیب کارها بسیار ساده تر می شوند. Rx کوئری های LINQ را بر روی observable collectionها، می گیرد.

بد نیست قبل از شروع Rx کمی راجع به collectionها صحبت کنیم. IEnumerables یکی از پراستفاده ترین collectionهای Pull Based است و همزمانی ذاتی دارد. در ادامه کد ساده ای در این باره مشاهده می کنید.

interface IEnumerable<out T>     

{     

    IEnumerator GetEnumerator();     

}   

interface IEnumerator<out T>:IDisposable   

{   

    bool moveNext();   

    T currennt { get; }   

    void Reset();   

}   

فرض کنیم به هر دلیلی، منابع داده مدتی از دسترس خارج شوند، در اینصورت چه اتفاقی می افتد؟ باید تا زمانی که پایگاه داده در دسترس قرار بگیرد، منتظر بماند و با پیغامی مانند شکل زیر مواجه خواهیم شد.

اما می توانید رابط مبتنی بر Pull (Pull based interface)  را به رابط مبتنی بر Push (Push based interface) تغییر دهید. در ادامه مثالی از این دست وجود دارد.

//Observables:- Push based    

interface IObservable<out T>   

{   

    IDisposable subscribe(IObserver observer);   

}   

interface IObserver<in T>   

{   

    void onNext(T value);   

    void onError(Exception ex);   

    void onCompleted();   

}   

بسیار کامل، دقیق و پایدار است. با مثال هایی از این قبیل بیشتر آشنا خواهیم شد. در ادامه یک نمونه برنامه در محیط کنسول را می بینید. IObservable و IObserver به صورت پیش فرض در .NET 4.0 و بالاتر قابل دسترس هستند.

با نصب Rx extension  از Nuget ادامه می دهیم.

در صورت موفقیت آمیز بودن نصب، می توانید صحت اسمبلی ها را بررسی کنید.

گرفتن observableها سه مرحله دارد.

بیایید با یک مثال ساده شروع کنیم. همانطور که در تصویر زیر می بینید، به محض شروع ایجاد observable، overloadهای متعددی در اختیار شما قرار می دهد که می توانید از آنها استفاده کنید.

using System;   

using System.Reactive.Linq;   

namespace ReactiveExtensions   

{   

    internal class Program   

    {   

        private static void Main(string[] args)   

        {   

            IObservable<string> obj = Observable.Generate(   

                0, //Sets the initial value like for loop   

                _ => true, //Don't stop till i say so, infinite loop   

                i => i + 1, //Increment the counter by 1 everytime   

                i => new string('#', i), //Append #   

                i => TimeSelector(i)); //delegated this to private method which just calculates time   

            //Subscribe here   

            using (obj.Subscribe(Console.WriteLine))   

            {   

                Console.WriteLine("Press any key to exit!!!");   

                Console.ReadLine();   

            }   

        }   

        //Returns TimeSelector   

        private static TimeSpan TimeSelector(int i)   

        {   

            return TimeSpan.FromSeconds(i);   

        }   

    }   

}  

در قطعه کد فوق، برای هر خط یک کامنت گذاشته شده که توضیح می دهد هر خط چه کاری انجام میدهد. از مزیت های این روش، همزمانی ذاتی آن است که در خروجی زیر آن را مشاهده می کنید:

برنامه من بر روی یک thread اجرا می شود، همزمان در همان صفحه کنسول می توانم تایپ هم کنم. یعنی UI من هنوز بلوکه نشده است. تا به اینجا امیدوارم از این مثال کوچک درباره Observableها خوشتان آمده باشد.

شروع کار با Reactive Extension- بخش دوم

پیرو بخش اول، مبحث را با چند مثال ساده سمت UI ادامه می دهیم. به کد زیر دقت کنید. در این کد یک فرم تحت ویندوز ایجاد کرده ایم و یک textbox به آن اضافه کرده ایم.

using System;   

using System.Reactive.Linq;   

using System.Windows.Forms;   

namespace ReactiveExtensions   

{   

    internal class Program   

    {   

        private static void Main(string[] args)   

        {   

           // ObservableGetValue();   

            var textbox = new TextBox();   

            var form = new Form   

            {   

                Controls = {textbox}   

            };   

          Application.Run(form);   

        }   

        private static void ObservableGetValue()   

        {   

            IObservable<string> obj = Observable.Generate(   

                0, //Sets the initial value like for loop   

                _ => true, //Don't stop till i say so, infinite loop   

                i => i + 1, //Increment the counter by 1 everytime   

                i => new string('#', i), //Append #   

                i => TimeSelector(i)); //delegated this to private method which just calculates time   

            //Subscribe here   

            using (obj.Subscribe(Console.WriteLine))   

            {   

                Console.WriteLine("Press any key to exit!!!");   

                Console.ReadLine();   

            }   

        }   

        //Returns TimeSelector   

        private static TimeSpan TimeSelector(int i)   

        {   

            return TimeSpan.FromSeconds(i);   

        }   

    }   

}  

حالا با اجرای کد فوق یک فرم تحت ویندوز با یک textbox در آن خواهیم داشت.

حالا می خواهیم به رویداد text changed گوش دهیم. باید یک text changed handler به شکل زیر اضافه می کنیم.

بنابراین قطعه کد زیر را ایجاد می کنیم.

using System;   

using System.Reactive.Linq;   

using System.Windows.Forms;   

namespace ReactiveExtensions   

{   

    internal class Program   

    {   

        private static void Main(string[] args)   

        {   

           // ObservableGetValue();   

            var textbox = new TextBox();   

            var form = new Form   

            {   

                Controls = {textbox}   

            };   

            //Trying to fetch the value from textbox   

            textbox.TextChanged += textbox_TextChanged;   

          Application.Run(form);   

        }   

        static void textbox_TextChanged(object sender, EventArgs e)   

        {   

            throw new NotImplementedException();   

        }   

        private static void ObservableGetValue()   

        {   

            IObservable<string> obj = Observable.Generate(   

                0, //Sets the initial value like for loop   

                _ => true, //Don't stop till i say so, infinite loop   

                i => i + 1, //Increment the counter by 1 everytime   

                i => new string('#', i), //Append #   

                i => TimeSelector(i)); //delegated this to private method which just calculates time   

            //Subscribe here   

            using (obj.Subscribe(Console.WriteLine))   

            {   

                Console.WriteLine("Press any key to exit!!!");   

                Console.ReadLine();   

            }   

        }   

        //Returns TimeSelector   

        private static TimeSpan TimeSelector(int i)   

        {   

            return TimeSpan.FromSeconds(i);   

        }   

    }   

}   

همانطور که در تصویر زیر می بینید، استفاده از textchanged ممکن نیست.

به همین دلیل باید از Reactive Extension ها استفاده کنیم و رویداد موردنظر را ثبت کنیم.

private static void Main(string[] args)   

{   

   // ObservableGetValue();   

    var textbox = new TextBox();   

    var form = new Form   

    {   

        Controls = {textbox}   

    };   

    //Using Rx, to get the textchanged    

    var textChanged = Observable.FromEventPattern(textbox, "TextChanged");   

    //then, I will subscribe the same    

  Application.Run(form);   

}  

همانطور که در تصویر زیر می بینید، یک EventArgs در اختیار من قرار میدهد.

حالا برای دریافت changed text، باید قبل از Rx از LINQ استفاده کنم. در ادامه کد کامل شده را مشاهده می کنید. البته برای خوانایی بیشتر و تمیزی، مقداری refactoring انجام داده ام.

using System;     

using System.Reactive.Linq;     

using System.Windows.Forms;     

namespace ReactiveExtensions   

{   

    internal class Program   

    {   

        private static void Main(string[] args)   

        {   

            // ObservableGetValue();   

            GetTextBoxValue();   

        }   

        private static void GetTextBoxValue()   

        {   

            var textbox = new TextBox();   

            var form = new Form   

            {   

                Controls = {textbox}   

            };   

            //Using Rx, to get the textchanged    

            var textChanged = Observable.FromEventPattern(textbox, "TextChanged");   

            //then, I will subscribe the same using LINQ Projection   

            var query = from e in textChanged   

                select ((TextBox) e.Sender).Text;   

            //Hence, if the application quits, i want to unsubscribe   

            using (query.Subscribe(Console.WriteLine))   

            {   

                Application.Run(form);   

            }   

        }   

        private static void ObservableGetValue()   

        {   

            IObservable<string> obj = Observable.Generate(   

                0, //Sets the initial value like for loop   

                _ => true, //Don't stop till i say so, infinite loop   

                i => i + 1, //Increment the counter by 1 everytime   

                i => new string('#', i), //Append #   

                i => TimeSelector(i)); //delegated this to private method which just calculates time   

            //Subscribe here   

            using (obj.Subscribe(Console.WriteLine))   

            {   

                Console.WriteLine("Press any key to exit!!!");   

                Console.ReadLine();   

            }   

        }   

        //Returns TimeSelector   

        private static TimeSpan TimeSelector(int i)   

        {   

            return TimeSpan.FromSeconds(i);   

        }   

    }   

}  

حالا با اعمال این تغییرات، زمانی که برنامه را اجرا کنیم، خروجی زیر را مشاهده خواهیم کرد.