Is it bad to keep subscribing?

Question

I have this method in my View Model:

    private void GetCustomers()
    {
      // Call the Model to get the Customers
      // Passing in 0 to get the first page
      // Paging could easily be done here
      // You could also pass in other criteria
      Model.GetCustomers(0).Subscribe(p = >
      {
        // Check for an error in the Service
        if (p.EventArgs.Error == null)
        {
          // loop thru each item in the
          // DataServiceCollection< CustomerRecord >
          // Collection
          foreach (CustomerRecord Customer in
            (DataServiceCollection< CustomerRecord >)p.Sender)
          {
            // Add to the Customer to the colCustomerRecord 
            // Collection so the View can bind to it
            colCustomerRecord.Add(Customer);
          }
        }
      });

    }

That is subscribing to this method in my Model:

    public static IObservable < IEvent < LoadCompletedEventArgs> > 
      GetCustomers(int intPage)
    {
      // Create a URI that points to the OData Service
      Uri objUri = new Uri(GetBaseAddress(), UriKind.RelativeOrAbsolute);

      // Set up oData service call
      SampleDataSource SDS = new SampleDataSource(objUri);

      // Construct a Query
      var query = (from SampleCustomerData in SDS.SampleCustomerData
             where SampleCustomerData.CustomerNotes.Contains("3")
             select SampleCustomerData).Skip(intPage).Take(10);

      // Set up a DataServiceCollection to hold the results
      DataServiceCollection< CustomerRecord > CustomerRecords = 
        new DataServiceCollection< CustomerRecord >();

      // Set up a Rx Observable (in a variable called observable) 
      // that will contain the results of
      // the "LoadCompleted" event that CustomerRecords will fire
      // When LoadAsync(query) is fired in the following statement
      IObservable< IEvent < LoadCompletedEventArgs> > observable =
        Observable.FromEvent< LoadCompletedEventArgs >(CustomerRecords,
        "LoadCompleted");

      // Execute the LoadAsync on CustomerRecords passing
      // the query that was constructed earlier
      CustomerRecords.LoadAsync(query);

      // Return observable
      return observable;
    }

A reader pointed out that it was bad that I was resubscribing each time I needed to pull up a new page. 

My question is, what is so bad about resubscribing ?

The way it is now, I am simply saving a LOT of code that I would need if I had to do the async the normal way.

Answer 1

Hi,

Re-subscribing isn't bad, unless of course it's happening more than necessary and it will hurt performance (doubtful in your case) or if there are side-effects that are expense to compute or that you don't want to happen more than once (also doubtful in your case).

Regardless, there are ways around having to subscribe each time a page is needed, and it may prove to be more manageable for you.  Have you considered storing a reference to a single DataServiceCollection, subscribing once and resetting the collection for each page?  Whether that's even possible depends upon the implementation of DataServiceCollection.

Alternatively, you could encapsulate your requirements and then consumers only must subscribe once.  Expose a Customers property of type IObservable<IEvent<LoadCompletedEventArgs>>.  Back it with a field of type Subject<IEvent<LoadCompletedEventArgs>>.  Then define a method named GetCustomersAsync to invoke the query, which would be the same code you have except that instead of returning the observable you would subscribe the subject; e.g., Observable.FromEvent<...>(CustomerRecords, "..").Subscribe(subject);

- Dave

---

http://davesexton.com/blog

Answer 2

Hi,

Just thought of another thing to point out.  Observables have an error channel, so instead of exposing the clunky IEvent type, it would probably be much better to expose the type of the object being returned and send errors to subject.OnError instead.

To do this you shouldn't subscribe the subject, or if you're sticking with your original re-subscribing approach then you shouldn't return the observable directly.  Instead, for the subject approach, subscribe a shim method that explicitly calls either subject.OnNext or OnError, depending upon the result of the query.  Alternatively, if you're using the re-subscribing approach, then return an observable using Observable.CreateWithDisposable.  The disposable will be a subscription to the observable you would normally return, and then you can use the observer parameter like a subject.

Here's a rough example of what the latter approach would look like:

public static IObservable<Customer> WhenCustomerLoaded(int intPage)
{
  return Observable.CreateWithDisposable<Customer>(
    observer =>
    {
      var uri = new Uri(GetBaseAddress(), UriKind.RelativeOrAbsolute);

      var sds = new SampleDataSource(uri);

      var query = (from data in sds.SampleCustomerData
            where data.CustomerNotes.Contains("3")
            select data).Skip(intPage).Take(10);

      var results = new DataServiceCollection<CustomerRecord>();

      var whenLoaded = Observable.FromEvent<LoadCompletedEventArgs>(results, "LoadCompleted");

      var disposable = whenLoaded.Subscribe(
        value =>
        {
          if (value.EventArgs.Error != null)
            observer.OnError(value.EventArgs.Error);
          else
          {
            results.Run(observer.OnNext);
            observer.OnCompleted();
          }
        }, 
      observer.OnError);

      results.LoadAsync(query);

      return disposable;
    });
}

- Dave

---

http://davesexton.com/blog

Answer 3

Wow thank you very much. Your version (of course) is much cleaner, and what you are exposing  IObservable<Customer> is WAY cleaner and I think others will feel more comfortable consuming it in their View Models than what I had. I will use your approach moving forward.

RX is great!

Answer 4

It appears I have a problem on this line:

results.Run(observer.OnNext);

There is no "Run".

Answer 5

I spent a few hours playing with this. I was unable to come up with a solution that did not involve adding a lot more code. Also it appears that you have to subscribe to a IObservable to get anything out of it.

So, even if I did get:

IObservable<Customer> WhenCustomerLoaded(int intPage)

working, I would still need a subscribe to the result, right?

Answer 6

Hi,

> There is no "Run".

I haven't used the Silverlight version, so if that's what you're using then perhaps EnumerableEx.Run wasn't included.  In the full version it's in the System.Linq namespace.  Try adding a using statement to your code file if you haven't already.  Alternatively, Run is just a shortcut for a very simple foreach loop that passes each value to the specified function.  You could replace it with the following code:

foreach (var value in results)
  observer.OnNext(value);

> I was unable to come up with a solution that did not involve adding a lot more code.

If you want to be more specific about the issues you've encountered someone will probably be able to help.

> I would still need a subscribe to the result, right?

Yes.  The point of the example was to create an abstraction.  Consuming it isn't much different than how you originally consumed the observable in your example:

  Model.GetCustomers(0).Subscribe(p => ...

When consuming my example, p would be a Customer.

Also, just to be clear, I don't have much info about the types that you're using so I can only make assumptions.  I did mention that my code was a "rough example", since I entered it into the forum editor without even using VS, let alone trying to compile and test it.  You're definitely going to have to make changes so that it works with your types.

- Dave

---

http://davesexton.com/blog

Answer 7

Dave,

First, let me thank you so much for your help. I wasn't even close to where I wanted to be without the code you provided. I have learned so much from your example I hope I did not come off as ungrateful. I was up to late and craky :). The thing that I was unable to get past was:

instead of:

results.Run(observer.OnNext);
observer.OnCompleted();

I need to use:

foreach (var item in results)
{
 observer.OnNext(item);
}

I was unable to figure out that "foreach" part. That is what had me stuck. Thank you for providing the answer :) I also thought I needed to get rid of all the "Subscribe" parts, but the examples on http://rxwiki.wikidot.com/101samples showed how it was used to get values from Observable.

So if I just set as a goal to create an abstraction that is easily consumed by my View Model, the code that works is:

using System;
using System.Linq;
using System.Collections.Generic;
using SilverlightODataSample.wsSampleCustomerData;
using System.Data.Services.Client;
using System.Collections.ObjectModel;

namespace SilverlightODataSample
{
 public class Model
 {
  public static ObservableCollection<CustomerRecord> WhenCustomerLoaded(int intPage)
  {
   ObservableCollection<CustomerRecord> FinalCollection = new ObservableCollection<CustomerRecord>();

   var myVal = Observable.CreateWithDisposable<CustomerRecord>(
    observer =>
    {
     var uri = new Uri(GetBaseAddress(), UriKind.RelativeOrAbsolute);

     var sds = new SampleDataSource(uri);

     var query = (from data in sds.SampleCustomerData
        select data).Skip(intPage).Take(10);

     var results = new DataServiceCollection<CustomerRecord>();

     var whenLoaded = Observable.FromEvent<LoadCompletedEventArgs>(results, "LoadCompleted");

     var disposable = whenLoaded.Subscribe(
      value =>
      {
       if (value.EventArgs.Error != null)
       {
        observer.OnError(value.EventArgs.Error);
       }
       else
       {
        foreach (var item in results)
        {
         observer.OnNext(item);
        }
       }
      },
      observer.OnError);

     results.LoadAsync(query);

     return disposable;
    });

   myVal.Subscribe(x =>
   {
    FinalCollection.Add(x);
   });

   return FinalCollection;
  }

  #region GetBaseAddress
  private static string GetBaseAddress()
  {
   // This gets the address of the webservice by 
   // getting the AbsoluteUri and then stripping out the 
   // name of the .xap file
   string strXapFile = @"/ClientBin/SilverlightODataSample.xap";
   string strBaseWebAddress =
    App.Current.Host.Source.AbsoluteUri.Replace(strXapFile, "");
   return string.Format(@"{0}/{1}", strBaseWebAddress, @"Service.svc");
  }
  #endregion
 }
}

To consume it, just one beautiful line of code!:

 private void GetCustomers(int intPage)
 {
  colCustomerRecord = Model.WhenCustomerLoaded((intPage * 10));
 }

If I am still missing something (for example the code I added "myVal.Subscribe..") please do not hesitate to point out any mistakes in your opinion.

Thank you again!

Answer 8

Hi,

It seems that you've combined your two original methods into a single method, but that's not what my intention was for creating the abstraction; instead, it was actually to decouple these two methods further.

Returning ObservableCollection means that subscribers lose the ability to handle exceptions.  Recall that observables have an error channel, which subscribers can listen to if they'd like to handle exceptions.  ObservableCollection doesn't provide such a mechanism, so you must handle them inside of the WhenCustomerLoaded method.  Clearly this is a problem if, for example, you'd like to display the error message in the UI.  If the code is in your view model or UI already, then you could just put all of your UI-related code into this method or expose an Error event on the view model (I was trying to help you away from this through the abstraction).  However, if this code isn't running in the UI or a view model, then you have to decide how you want to propagate error information to callers; e.g., your method could accept an Action<Exception> parameter.  But that seems a bit silly to me because observables already provide this in the form of the onError parameter on Subscribe.

Another thing that you lose with ObservableCollection is laziness.  As soon as your method is called the query is executed.

Yet another thing lost is composibility.  If in the future you have the need to create reactive queries on top of the collection, for example to filter or project the values in different ways, you'll probably want to use IObservable instead because it allows you to use reactive LINQ, whereas ObservableCollection doesn't.

Edit: Another thing lost with ObservableCollection is cancellation.  Notice in the example below that I'm not using disposable.

Therefore, I still recommend returning IObservable<CustomerRecord> from this method and moving it into your business layer.  Retain your original GetCustomers method in your view model or UI and populate an ObservableCollection by subscribing, as you have.  You may also listen to the error channel at the same time so that errors can be handled properly in the UI.  This will keep your method encapsulated, reusable, cancellable, lazy and composable.

But if you're sure that you want to return ObservableCollection from this method anyway, instead of returning IObservable, then you no longer need my example.  For one thing, you can remove the error-related code and as I mentioned handle the error explicitly.  In my example below I'm just throwing the error, which will cause the application to crash because it will be on a background thread; i.e., it cannot be caught from calling code.  Furthermore, there's no longer any need to create a custom observable since you can populate your collection from the LoadCompleted subscription, similar to what you had in your original post.

Here's another rough example, although to be clear this is not the approach that I recommend.  I'm posting it for completeness and to illustrate the difference in error handling when using ObservableCollection instead of IObservable. 

public static ObservableCollection<CustomerRecord> WhenCustomerLoaded(int intPage)
{
 var collection = new ObservableCollection<CustomerRecord>();

 var uri = new Uri(GetBaseAddress(), UriKind.RelativeOrAbsolute);

 var sds = new SampleDataSource(uri);

 var query = (from data in sds.SampleCustomerData
  select data).Skip(intPage).Take(10);

 var results = new DataServiceCollection<CustomerRecord>();

 var whenLoaded = Observable.FromEvent<LoadCompletedEventArgs>(results, "LoadCompleted");

 var disposable = whenLoaded.Subscribe(
 value =>
 {
 if (value.EventArgs.Error != null)
 {
 throw value.EventArgs.Error;
 }
 else
 {
 foreach (var item in results)
 {
  collection.Add(item);
 }
 }
 });

 results.LoadAsync(query);

 return collection;
}

- Dave

---

http://davesexton.com/blog

Answer 9

Man that code is great. I plan to use it in an upcoming http://CodeProject.com article (I will give you credit in the source code). The "use case" is the desire to easily call a web service with paging. Right now you have that with a 90% reduction in code vs. other methods. As a bonus, it only takes one line of code to call the method.

Thousands of people will probably use this code. A bit ironic that you don't recommend it :)

Thanks for everything!

Answer 10

Hi,

What I recommend is separating the IObservable implementation from the ObsevableCollection implementation.  IObservable is much more useful.  Also it's more natural to go from IObservable to ObservableCollection than it is to go the other way around.

Here's an example of why it would be useful to separate your business logic from your presentation logic.  To be honest, I'd even move the paging implementation and the LINQ query into the business layer, but this is just an example :)

Note: I haven't tried compiling this code.

class PaidCustomersViewModel
{
  public int ClientId
  {
    get;
    set;
  }

  public ObservableCollection<CustomerRecord> Customers
  {
    get
    {
      return customers;
    }
  }

  private readonly ObservableCollection<CustomerRecord> customers = new ObservableCollection<CustomerRecord>();
  private int currentPage = -1;

  public void NextPage()
  {
    currentPage = (currentPage == -1) ? 0 : currentPage + 10;

    IObservable<CustomerRecord> whenCustomerLoaded =
      from customer in CustomerRecord.WhenCustomerLoaded(currentPage)
      where customer.Paid == true
      where customer.ClientId == ClientId
      select customer;

    customers.Clear();

    whenCustomerLoaded.Subscribe(
     customers.Add,
     ex => LogAndShowErrorToUser(ex.Message),
     () => NotifyUIThatAllCustomersAreLoaded());
  }

  public void PreviousPage() { ... }
  ...
}

- Dave

---

http://davesexton.com/blog

Answer 11

I will point people toward this thread so they can see each version and your explanation.

... but the the Observabale collection version is so perfect I want to frame it... :)

Answer 12

I posted an article that uses the code here:

http://www.codeproject.com/KB/silverlight/ODataPaging.aspx