Catching exceptions in Changes.ObserveContinuous()

Handles negative responses and server side errors. (#176)

Author: SamuelWeibel

source/projects/MyCouch/Contexts/Changes.cs

@@ -97,11 +97,13 @@ public virtual IObservable<string> ObserveContinuous(GetChangesRequest request,
 
             Task.Factory.StartNew(async () =>
             {
-                using (var httpResponse = await SendAsync(httpRequest, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ForAwait())
+                try
                 {
-                    var response = await ContinuousChangesResponseFactory.CreateAsync(httpResponse).ForAwait();
-                    if (response.IsSuccess)
+                    using (var httpResponse = await SendAsync(httpRequest, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ForAwait())
                     {
+                        var response = await ContinuousChangesResponseFactory.CreateAsync(httpResponse).ForAwait();
+                        if (!response.IsSuccess)
+                            throw new MyCouchResponseException(response);
                         using (var content = await httpResponse.Content.ReadAsStreamAsync().ForAwait())
                         {
                             using (var reader = new StreamReader(content, MyCouchRuntime.DefaultEncoding))
@@ -117,6 +119,10 @@ public virtual IObservable<string> ObserveContinuous(GetChangesRequest request,
                         }
                     }
                 }
+                catch (Exception e)
+                {
+                    ob.Error(e);
+                }
             }, cancellationToken).ForAwait();
 
             return ob;

source/projects/MyCouch/MyObservable.cs

@@ -38,6 +38,15 @@ public virtual void Complete()
                 observer.OnCompleted();
         }
 
+        public virtual void Error(Exception error)
+        {
+            var obs = _observers.ToArray();
+            _observers.Clear();
+
+            foreach (var observer in obs)
+                observer.OnError(error);
+        }
+
         private class Unsubscriber : IDisposable
         {
             private readonly IList<IObserver<T>> _observers;