Rx - Executing anonymous code asynchronously

In the previous post we discuss about how we can execute a method asynchronously using Reactive Extensions. In this post we will discuss how we can execute anonymous code asynchronously using Reactive Extensions. Rx provides this support using Observable.Start.

Invoking Actions (No data returned):
Let us start with discussing how we can execute code asynchronously with no data to return.


The event handler of Click event of the button is as follows:

private void button2_Click(object sender, RoutedEventArgs e)
{
    Observable.Start(() =>
        {
            Thread.Sleep(2000);

            return;
        }
    )
    .ObserveOnDispatcher()
    .Subscribe(
                (result) => { ;},
                (ex) => this.Background = Brushes.Red,
                () => this.Background = Brushes.Green);
}

This code just executes simple code asynchronously in a ThreadPool thread. It generates OnNext, OnCompleted and OnError messages on the Dispatcher thread as we are observing on Dispatcher. After atleast two seconds the background of window should turn green.

Invoking anonymous code (data returned from lambda) asynchronously:
We can also execute anonymous asynchronous code using Observable.Start which actually return a value. This code is in the form of lambda expressions or lambda statements which explicitly return data through return statement. Let us add a button on the same window. When user clicks the button it should update the operand boxes with any random number.

private void btnFillRandomOperands_Click(object sender, RoutedEventArgs e)
{
    Observable.Start<int>(
        () =>
        {
            Thread.Sleep(2000);
            return new Random().Next(1, 10);
        }).ObserveOnDispatcher<int>()
        .Subscribe(
                       (result) =>
                       {
                           this.textBoxOperand1.Text = result.ToString();
                           this.textBoxOperand2.Text = result.ToString();
                       },
                        (ex) => this.Background = Brushes.Red,
                        () => this.Background = Brushes.Green
                    );

}

The above code uses Random class to generate a random number between 1 and 10. The returned number is provided to OnNext handler by Rx runtime. This handler updates both operand text boxes with this number. After this, the runtime places OnCompleted handler which turns the background of the window to green.

When we run this and click this button, it results in the following window after atleast 2 seconds. Definitely, it might be a different random number generated and filled in the operands text boxes.



Passing Arguments to Code executed by Observable.Start:
It doesn’t seem that we can have any parameters to Observable.Start directly. But we can use our old technique of using Captured Variables. As we know that lambdas can capture variables from the scope they are generated from. The variables used by lambda like this are called Captured variables. Like a regular lambda statement executed asynchronously, these variables also have Closure issues. I have discussed here how to avoid closure issues by assigning the value of captured variables to local variables inside lambdas.

http://shujaatsiddiqi.blogspot.com/2010/12/wpf-dispatcherbegininvoke-and-closures.html

private void button1_Click(object sender, RoutedEventArgs e)
{
    decimal operand1 = Decimal.Parse(this.textBoxOperand1.Text);
    decimal operand2 = Decimal.Parse(this.textBoxOperand2.Text);

    Observable.Start(() =>
                         {
                             Thread.Sleep(2000);

                             return operand1 + operand2;
                         })
                         .ObserveOnDispatcher()
                         .Subscribe(
                                (result) => { this.textBoxResult.Text = result.ToString(); });

    operand1 = 3;
    operand2 = 4;
}

In the above case what do you guess would be shown in the result textbox if we enter 2.1 and 3.5 in the two operand text boxes. If you guess 5.6 then you would be surprised to see this result.


This is the result of modified captured variables as the modified value of those captured variables are used which are 3 and 4. The result of addition of 3 and 4 is 7, which is displayed in the Result text box. In order to fix this, we can define two new local variables localOperand1 and localOperand2 and initialize them with the values of captured variables.

private void button1_Click(object sender, RoutedEventArgs e)
{
    decimal operand1 = Decimal.Parse(this.textBoxOperand1.Text);
    decimal operand2 = Decimal.Parse(this.textBoxOperand2.Text);

    Observable.Start(() =>
                         {
                             decimal localOperand1 = operand1;
                             decimal localOperand2 = operand2;

                             Thread.Sleep(2000);
 
                             return localOperand1 + localOperand2;
                         })
                         .ObserveOnDispatcher()
                         .Subscribe(
                                (result) => { this.textBoxResult.Text = result.ToString(); });

    Thread.Sleep(2);

    operand1 = 3;
    operand2 = 4;
}

When we run the application and enter data 2.1 and 3.5 then the correct result is displayed in the Result text box.


Although Resharper still shows warning but it is just fine as we have seen the correct result being calculated.


Observable.Start and Exceptions in the code executed:
As a result of an exception Observable.Start behaves same way as Observable.ToAsync. We can represent the behavior in marble diagram as follows:


Let us update the code executed asynchronously using Observable.Start so that if the sum of number entered is either zero or lesser, it should throw an Exception.

[DebuggerStepThrough]
private void button1_Click(object sender, RoutedEventArgs e)
{
    decimal operand1 = Decimal.Parse(this.textBoxOperand1.Text);
    decimal operand2 = Decimal.Parse(this.textBoxOperand2.Text);

    Observable.Start(() =>
    {
        decimal localOperand1 = operand1;
        decimal localOperand2 = operand2;
        decimal sum = localOperand1 + localOperand2;

        Thread.Sleep(2000);

        if (sum <= 0)
        {
            throw new System.Exception("Error in computation");
        }

        return sum;
    })
    .ObserveOnDispatcher()
    .Subscribe(
        (result) => { this.textBoxResult.Text = result.ToString(); },
        (ex) => this.Background = Brushes.Red,
        () => this.Background = Brushes.Green);

}

You can notice that we are still subscribing on Dispatcher. In addition to OnNext, we have provided the code for OnCompleted and OrError. The window should turn green when the operation completed successfully resulting in OnCompleted. In case of exception, OnError is executed.



Using IScheduler to run Asynchronous code:
Like Observable.ToAsync, Observable.Start also provides overloads to run the asynchronous code using different IScheduler. We can run code which returns or does not return any data in the form of lambda and we can run it using the IScheduler of our choice. You can see that just by providing Scheduler.Dispatcher info to Start we can turn this method to run on UI thread:

Observable.Start(() =>
    {
        Thread.Sleep(2000);

        return;
    }, Scheduler.Dispatcher
)

Note:
There is one more overload of Observable.Start which returns ListObservable<TSource>. Since it is not about executing anonymous asynchronous code, we are discussing it here.

Download Code:

IObservable.Throttle causes OnNext on ThreadPool thread

This is a continuation of our following post:

http://shujaatsiddiqi.blogspot.com/2011/02/mvvm-observable-inotifypropertychangedp.html

In the above post we discussed about using PropertyChanged event of INotifyPropertyChanged by observing EventStream support of Reactive Extensions Rx. In this post we change the view model a bit. We remove the implementation of INotifyPropertyChanged. We rather define it as a DependencyObject to support a different change notification mechanism from view model. This is to demonstrate the following:

IObservable.Throttle causes OnNext to be generated on a ThreadPool thread instead.

We will see what problems might be caused by this and how we can work around this.

public class StudentViewModel : DependencyObject
{
    #region Model
    private Student _model;
    #endregion Model

    #region Constructor

    public StudentViewModel(Student student)
    {
        _model = student;

        Observable.FromEvent<PropertyChangedEventArgs>(_model, "PropertyChanged")
            .Where(et => et.EventArgs.PropertyName == "StudentName")
            .Subscribe((arg) =>
                           {
                               this.StudentId = _model.StudentId;
                               this.StudentName = _model.StudentName;
                           });
    }

    #endregion Constructor

    #region Dependency Properties

    public static DependencyProperty StudentIdProperty =
        DependencyProperty.Register("StudentId", typeof(int), typeof(StudentViewModel));
    public int StudentId
    {
        get { return (int)GetValue(StudentIdProperty); }
        set { SetValue(StudentIdProperty, value); }
    }

    public static DependencyProperty StudentNameProperty =
        DependencyProperty.Register("StudentName", typeof(string), typeof(StudentViewModel));
    public string StudentName
    {
        get { return (string)GetValue(StudentNameProperty); }
        set { SetValue(StudentNameProperty, value); }
    }

    #endregion Dependency Properties

    #region Overriden methods of DependencyObject
    protected override void OnPropertyChanged(DependencyPropertyChangedEventArgs e)
    {
        base.OnPropertyChanged(e);
        switch (e.Property.Name)
        {
            case "StudentId":
                _model.StudentId = (int)e.NewValue;
                break;
            case "StudentName":
                _model.StudentName = (string)e.NewValue;
                break;
        }
    }
    #endregion Overriden methods of DependencyObject
}

We have update the properties StudentId and StudentName to be Dependency properties. To synchronize the model's corresponding properties we have overridden OnPropertyChanged method of DependencyObject. The argument DependencyPropertyChangedEventArgs , not only, provides us the details about the property being updated but it also provides the old and new values of this property. As you can see we are just using the new values of these properties to update the corresponding model's properties.

We have kept the handling of model's PropertyChanged event observation using Reactive extension's Observable event stream support available in CoreEx assembly. Let's run the application now and open two child windows.


When we enter StudentName in one child window, the same changes are reflected in the other child window.


Now we add throttling support in order to avoid frequent updates to the other child windows due to updates in model's properties. Let's update the constructor of StudentViewModel as follows:

public StudentViewModel(Student student)
{
    _model = student;

    Observable.FromEvent<PropertyChangedEventArgs>(_model, "PropertyChanged")
        .Where(et => et.EventArgs.PropertyName == "StudentName")
        .Throttle(TimeSpan.FromMilliseconds(500))
        .Subscribe((arg) =>
                       {
                           this.StudentId = _model.StudentId;
                           this.StudentName = _model.StudentName;
                       });
}

This definition is same as previous. We have just used Throttling for PropertyChanged event for 500 milliseconds. Now we run the application. Open two child windows and enter data in StudentName property in one of the child window. As soon as we settle, the application shuts down. Let's see what is going on by updating the code further as follows:

public StudentViewModel(Student student)
{
    _model = student;

    Observable.FromEvent<PropertyChangedEventArgs>(_model, "PropertyChanged")
        .Where(et => et.EventArgs.PropertyName == "StudentName")
        .Throttle(TimeSpan.FromMilliseconds(500))
        .Subscribe((arg) =>
                       {
                           try
                           {
                               this.StudentId = _model.StudentId;
                               this.StudentName = _model.StudentName;
                           }
                           catch (Exception ex)
                           {
                               MessageBox.Show(ex.Message);
                           }
                       });
}

Now we run the application and enter data in StudentName text box as previous. The exception handling logic causes the following MessageBox to be displayed.


This is because of the reason that IObservable.Throttle causes OnNext to be placed in a ThreadPool thread.


Using Invoke / BeginInvoke on Dispatcher:
The first solution could be to directly using Dispatcher.Invoke / Dispatcher.BeginInvoke. We can specify the code that we want to execute in the Action delegate. The delegate's code is executed here on UI thread. We update the constructor as follows:

public StudentViewModel(Student student)
{
    _model = student;

    Observable.FromEvent<PropertyChangedEventArgs>(_model, "PropertyChanged")
        .Where(et => et.EventArgs.PropertyName == "StudentName")
        .Throttle(TimeSpan.FromMilliseconds(500))
        .Subscribe((arg) =>
                       {
                           Dispatcher.BeginInvoke(new Action(() =>
                                  {
                                      this.StudentId = _model.StudentId;
                                      this.StudentName = _model.StudentName;
                                  }), null);                                   
                           
                       });
}

Now we run the application. Again we open two windows and enter data in StudentName text box. We see that the data is successfully updated in the other window now. Let's insert the break point again in OnNext code.


Using other overload of Throttle:
We can use other overload of IObservable.Throttle method. This requires an additional overload which has an additional parameter of type IScheduler. IScheduler is a Rx interface available in CoreEx assembly and System.Concurrency namespace.

public StudentViewModel(Student student)
{
    _model = student;

    Observable.FromEvent<PropertyChangedEventArgs>(_model, "PropertyChanged")
        .Where(et => et.EventArgs.PropertyName == "StudentName")
        .Throttle(TimeSpan.FromMilliseconds(500), Scheduler.Dispatcher)
        .Subscribe((arg) =>
                       {
                           this.StudentId = _model.StudentId;
                           this.StudentName = _model.StudentName;
                       });
}

We have specified Scheduler.Dispatcher for other parameter. This is also available in System.Concurrency namespace. It is available in CoreEx assembly. This is saving us from dispatching in our code but Throttle's generated OnNext messages will be generated in the UI thread. Let's run the application and open two child windows like before.


Download:

MVVM - Joining Rx's IObservables using LINQ to IObservable in a WPF Application

In this post, we will see how we can combine two IObservable(s) using LINQ. The project is developed on top of the code as developed in this post:

http://shujaatsiddiqi.blogspot.com/2011/02/mvvm-view-model-iobserver-observing.html

We will just be discussing the changes required for this example. For the discussion of complete code please refer to the link provided above.

As in the previous post, the view model will be observing the data streams from to IObservable. It needs to combine them both using LINQ (as specified above). It would be interesting as we would see how data is combined when data is pushed by the IObservable. As soon as the data is available, it would automatically combine it using different operators provided by Reactive Extensions (Rx). Since we don't need to display data separately in the view, we update the view just to have one ListBox. The ListBox should show Id and Name of each Student. It should also show the name of the course the student is enrolled in, which could be obtained from observing CourseModel.


After completion of data stream, the use should be notified by changing the Fill color of ellipse as green. The expected final display is shown as follows:



To update the view as described above, we need to update the design of the view as follows:

<Window x:Class="WpfApp_MVVM_ReactiveModel.MainWindow"
        xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
        xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
        xmlns:local="clr-namespace:WpfApp_MVVM_ReactiveModel"
        Title="MainWindow" Height="350" Width="525" >    
    <Window.DataContext>
        <local:MainWindowViewModel />
    </Window.DataContext>   
    <Grid>       
        <ListBox Height="250" HorizontalAlignment="Left" Margin="-2,0,0,0" 
                 Name="listBox1" VerticalAlignment="Top" Width="493" 
                 ItemsSource="{Binding StudentList}">
            <ListBox.ItemTemplate>
                <DataTemplate >
                    <StackPanel Orientation="Horizontal">
                        <TextBlock Text="{Binding StudentId, StringFormat=Id:{0}}" />
                        <TextBlock Text="       " />
                        <TextBlock Text="{Binding StudentName, StringFormat=Name:{0}}" />
                        <TextBlock Text="       " />
                        <TextBlock Text="{Binding CourseName, StringFormat=Name:{0}}" />
                    </StackPanel>
                </DataTemplate>
            </ListBox.ItemTemplate>
        </ListBox>
        <Ellipse Height="39" HorizontalAlignment="Left" Margin="370,257,0,0" 
                 Name="ellipse1" Stroke="Black" VerticalAlignment="Top" Width="116" >
            <Ellipse.Style>                
                <Style TargetType="{x:Type Ellipse}">
                    <Style.Setters>
                        <Setter Property="Fill" Value ="Blue" />
                    </Style.Setters>
                    <Style.Triggers>
                        <DataTrigger Binding="{Binding IsLoaded}" Value="true" >
                            <Setter Property="Fill" Value="Green" />
                        </DataTrigger>
                    </Style.Triggers>
                </Style>
            </Ellipse.Style>
        </Ellipse>
    </Grid>
</Window>

The view is still using MainWindowViewModel as DataContext. You can see that an additional TextBlock is added to the template of items of ListBox. It is bound to CourseName from each item of the collection bound to the list box. The DataContext is expected to have a collection StudentList with items having properties StudentId, StudentName and CourseName. This is implemented as an ObservableCollection so that as the items are entered in the collection, the view could update itself using the notification mechanism built-in for this type of collection. It should also have a boolean property, IsDataLoaded. It should turn true when data is completely loaded. This is used in DataTrigger in the view to turn the notification ellipse to be filled as Green. We need to implement INotifyPropertyChanged just because of this property so that view can be notified for this change and the data triggered could trigger the fill color of the ellipse.

class MainWindowViewModel : INotifyPropertyChanged
{
    private readonly ObservableCollection<StudentViewModel> _studentList = 
        new ObservableCollection<StudentViewModel>();

    public ObservableCollection<StudentViewModel> StudentList
    {
        get { return _studentList; }
    }

    private bool _isLoaded = false;
    public bool IsLoaded
    {
        get { return _isLoaded; }
        set 
        {
            _isLoaded = value;
            onPropertyChanged("IsLoaded");
        }
    }

    public MainWindowViewModel()
    {
        var courseModel2 = new CourseModel();
        var studentModel2 = new StudentsModel();

        (from c in courseModel2
         from s in studentModel2
         where s.CourseId == c.CourseId
         select new { s.StudentName, s.StudentId, c.CourseName })
            .Subscribe(
                (on) => _studentList.Add(new StudentViewModel(on)),
                (ex) => Console.WriteLine(@"On error"),
                () => IsLoaded = true);
     
    }

    #region INotifyPropertyChanged implementation
    public event PropertyChangedEventHandler PropertyChanged;
    private void onPropertyChanged(string propertyname)
    {
        if (PropertyChanged != null)
        {
            PropertyChanged(this, new PropertyChangedEventArgs(propertyname));
        }
    }
    #endregion INotifyPropertyChanged implementation
}

The focal point of this post is the constructor of MainWindowViewModel. Here we are combining to IObservable based collections using LINQ. This is a special new flavor of LINQ released with Rx. This is called LINQ to IObservable. This is available in System.Reactive assembly. Let's have a close look at the constructor.

>
public MainWindowViewModel()
{
    var courseModel2 = new CourseModel();
    var studentModel2 = new StudentsModel();

    (from c in courseModel2
     from s in studentModel2
     where s.CourseId == c.CourseId
     select new { s.StudentName, s.StudentId, c.CourseName })
        .Subscribe(
            (on) => _studentList.Add(new StudentViewModel(on)),
            (ex) => Console.WriteLine(@"On error"),
            () => IsLoaded = true);
 
}

So we are cross-joining instances of CourseModel and StudentsModel. This would be creating a Cartesian Product. We are filtering it for only those which have matching CourseId(s).

As specified in the definition of the view, we are interested in StudentId and StudentName from StudentsModel and CourseName from CourseModel.

The cross product of Cartesian Product of two or more IObservable(s) is also an IObservable. This enables us to subscribe to it by providing the OnNext, OnCompleted and OnError of Observer. It is to remember that MainWindowViewModel is not the observable for the new IObservable created by combining CourseModel and StudentsModel. Rx run-time would take care of how the provided lambda expressions for OnNext, OnCompleted and OnError are to be executed.

The overload of Subscribe method that we have used takes three arguments OnNext, OnError and OnCompleted respectively. We are not doing anything useful in the lambda expression provided for OnError except just writing on the exception message on the console. In the OnCompleted expressoin we are setting IsLoaded to true. This would trigger the DataTrigger in the view and change the fill color of Notification ellipse to Green. The type of parameter of OnNext is the anonymous type created for Select. This would have same elements with similar names. As we have data available in OnNext, we are creating a StudentViewModel object and add it to the StudentsList ObservableCollection . This is bound to the ListBox in the view which would show this new element. Though we can not use foreach for IObservable as it is a feature of just IEnumerable, we can understand it by this nested foreach loop. Don't even try this:


Since the combined collections have IObservable based, we don't have the data available at the same time. This makes it very special form of joining. This would keep each element of courseModel2 in memory until OnCompleted is not executed from studentModel2. As the elements are being received in background OnNext for studentModel they are combined with each element of courseModel. This would be creating equal number of Observers for studentModel2 as the total number of elements pushed by courseModel2. As the elements of courseModel2 continue to be received they are added to the list of Observers of studentModel2.


We should be able to safely make a summarized statement about it as follows:

All the elements pushed through OnNext of courseModel2 would be combined with all the elements pushed by studentModel2. If there is any OnNext of either of these IObservable before this LINQ, that would not be considered for joining.

As soon as the OnCompleted is executed for this Observer created at runtime, it would automatically unsubscribe it by disposing itself. As we have seen in previous posts that disposing an unsubscriber would result in its unsubscription. I think this is the default implementation provided for this dynamic Observer by Rx runtime.

This would need updates in StudentViewModel. We need to add a property CourseName as used in MainWindowViewModel. There is one more interesting thing, the datatype of parameter for constructor is specified as dynamic. This is basically because of anonymous type constructed during LINQ operation. In the constructor, we are using it as a duck type datatype, we know what properties would it have but we don't know it's name as it would be assigned any arbitrary name at run-time. This is a special feature of .net framework 4.0 [http://msdn.microsoft.com/en-us/library/dd264736.aspx].

class StudentViewModel : INotifyPropertyChanged
{
    public StudentViewModel(dynamic student)
    {
        this.StudentId = student.StudentId;
        this.StudentName = student.StudentName;
        this.CourseName = student.CourseName;
    }

    private string _courseName;
    public string CourseName
    {
        get { return _courseName; }
        set 
        { 
            _courseName = value;
            OnPropertyChanged("CourseName");
        }
    }

    private string _studentName;
    public string StudentName
    {
        get { return _studentName; }
        set 
        { 
            _studentName = value;
            OnPropertyChanged("StudentName");
        }
    }

    private int _studentId;
    public int StudentId
    {
        get { return _studentId; }
        set
        {
            _studentId = value;
            OnPropertyChanged("StudentId");
        }
    }

    #region INotifyPropertyChanged implementation

        public event PropertyChangedEventHandler  PropertyChanged;
        private void OnPropertyChanged(string propertyName)
        {
            if (PropertyChanged != null)
            {
                PropertyChanged(this, new PropertyChangedEventArgs(propertyName));
            }
        }

        #endregion INotifyPropertyChanged implementation
}

So we need to update Student to add a new property, CourseId. This would be used as a key to join StudentsModel and CourseModel IObservable collections. This is int type.

>
class Student 
{
    public string StudentName { get; set; }
    public int StudentId { get; set; }
    public int CourseId { get; set; }
}

Since we have updated Student type, we need to update the definition for StudentsModel. This is to specify value of CourseId. We are just deriving it based on a algorithm [If _studentId is odd then set CourseId as 2, otherwise, set it as 1]. We are also reducing the number of times that timer event would execute. You can see that we are calling Observer's OnCompleted just when the event has been executed 3 times. As we call OnCompleted on Observer, we are also stopping the timer, this would not keep the timer running on the background.

>
class StudentsModel : IObservable<Student>
{
    private int _studentId = 1;

    //Timer to notify observers about updates in the reactive observable
    private readonly DispatcherTimer _t = new DispatcherTimer();

    public List<IObserver<Student>> Observers = 
        new List<IObserver<Student>>();

    public StudentsModel()
    {
        _t.Interval = TimeSpan.FromMilliseconds(3000);
        _t.Tick += new EventHandler(Tick);
        _t.Start();
    }

    void Tick(object sender, EventArgs e)
    {
        var student = new Student()
                          {
                              StudentId = _studentId,
                              StudentName = string.Format("Student: {0}", _studentId),
                              CourseId = ((int)(_studentId % 2)) + 1
                          };

        if (_studentId <= 3)
        {
            Observers.ForEach((observer) => observer.OnNext(student));
            _studentId++;
        }
        else
        {
            Observers.ForEach((observer) => observer.OnCompleted());
            _t.Stop();
        }
    }

    public IDisposable Subscribe(IObserver<Student> observer)
    {
        IDisposable unSubscriber = new Unsubscriber<Student>(Observers, observer);
        if (!Observers.Contains(observer))
        {
            Observers.Add(observer);
        }
        return unSubscriber;
    }
}

Now Let's run the application. The application runs successfully and we see the elements being added to the list box.



But there is a weird thing. The notification ellipse is still filled as blue. It should have turned Green if OnCompleted of the LINQ result is executed in MainWindowViewModel. It means that OnCompleted ,for this, is not executed. You can prove that by inserting break point in the lambda expression used for OnCompleted. But the main question is why it is not executed??

Basically, we have discussed that equal numbers of Observers would be created for studentModel2 as the number of elements pushed by courseModel2 after the LINQ statment is executed. So there are will be two observers for this. If we insert break-point in the Tick event handler for timer _t in StudentsModel then we realize that OnCompleted is executed only once. We need to execute OnCompleted for all observers to fix this. This is really weird because we are using Foreach on Obsevers List. It should have executed the OnCompleted for all observers. There should have been no Observer in the list before next statement is executed [_t.Stop()]. But if we put a break point then it seems that OnCompleted is executed only once as one Observer is still lingering on.


This seems to get fixed if we update the code as follows:

void Tick(object sender, EventArgs e)
{
    var student = new Student()
                      {
                          StudentId = _studentId,
                          StudentName = string.Format("Student: {0}", _studentId),
                          CourseId = ((int)(_studentId % 2)) + 1
                      };

    if (_studentId <= 3)
    {
        Observers.ForEach((observer) => observer.OnNext(student));
        _studentId++;
    }
    else
    {
        Observers.ToList().ForEach((observer) => observer.OnCompleted());                
        _t.Stop();
    }
}

Now run this and insert the break point appropriately.



We have just called ToList() on a collection which is already a Generic List to fix this. This shows some issue with the deferred execution of Lambdas of foreach here, which is executing OnCompleted for each observer. This would definitely introduce some performance delay but whatever works...

Now since all the OnCompleted have been executed for all the collections involved in the LINQ query, it is safe to executed the OnCompleted of the result. Rx runtime does exactly that. This sets IsLoaded to true and results in filling the Notification ellipse as Green.



Download Code: