Switch to Cold Observables

InterfaceStubGenerator/Program.cs

@@ -37,6 +37,7 @@ static void Main(string[] args)
             using (var of = File.OpenWrite(target.FullName)) {
                 var bytes = Encoding.UTF8.GetBytes(template);
                 of.Write(bytes, 0, bytes.Length);
+                of.Flush();
             }
         }
 

Refit-Tests/GitHubApi.cs

@@ -65,6 +65,9 @@ public interface IGitHubApi
         [Get("/")]
         Task<HttpResponseMessage> GetIndex();
 
+        [Get("/")]
+        IObservable<string> GetIndexObservable();
+
         [Get("/give-me-some-404-action")]
         Task NothingToSeeHere();
     }

Refit-Tests/RefitStubs.cs

@@ -75,6 +75,12 @@ public virtual Task<HttpResponseMessage> GetIndex()
             return (Task<HttpResponseMessage>) methodImpls["GetIndex"](Client, arguments);
         }
 
+        public virtual IObservable<string> GetIndexObservable()
+        {
+            var arguments = new object[] {  };
+            return (IObservable<string>) methodImpls["GetIndexObservable"](Client, arguments);
+        }
+
         public virtual Task NothingToSeeHere()
         {
             var arguments = new object[] {  };

Refit-Tests/RequestBuilder.cs

@@ -287,10 +287,12 @@ public class SomeRequestData
     public class TestHttpMessageHandler : HttpMessageHandler
     {
         public HttpRequestMessage RequestMessage { get; private set; }
+        public int MessagesSent { get; set; }
 
         protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
         {
             RequestMessage = request;
+            MessagesSent++;
             return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("test") });
         }
     }

Refit-Tests/RestService.cs

@@ -100,12 +100,33 @@ public async Task HitTheGitHubUserApiAsObservableAndSubscribeAfterTheFact()
             // NB: We're gonna await twice, so that the 2nd await is definitely
             // after the result has completed.
             await obs;
-
             var result2 = await obs;
             Assert.AreEqual("octocat", result2.Login);
             Assert.IsFalse(String.IsNullOrEmpty(result2.AvatarUrl));
         }
 
+        [Test]
+        public async Task TwoSubscriptionsResultInTwoRequests()
+        {
+            var input = new TestHttpMessageHandler();
+            var client = new HttpClient(input) { BaseAddress = new Uri("http://foo") };
+            var fixture = RestService.For<IGitHubApi>(client);
+
+            Assert.AreEqual(0, input.MessagesSent);
+
+            var obs = fixture.GetIndexObservable()
+                .Timeout(TimeSpan.FromSeconds(10));
+
+            await obs;
+            Assert.AreEqual(1, input.MessagesSent);
+
+            var result = await obs;
+            Assert.AreEqual(2, input.MessagesSent);
+
+            // NB: TestHttpMessageHandler returns 'test' for every request
+            Assert.IsTrue(result.Contains("test"));
+        }
+
         [Test]
         public async Task ShouldRetHttpResponseMessage()
         {

Refit/RequestBuilderImplementation.cs

@@ -203,10 +203,19 @@ Func<HttpClient, object[], Task> buildVoidTaskFuncForMethod(RestMethodInfo restM
         Func<HttpClient, object[], Task<T>> buildTaskFuncForMethod<T>(RestMethodInfo restMethod)
             where T : class
         {
-            return async (client, paramList) => {
+            var ret = buildCancellableTaskFuncForMethod<T>(restMethod);
+            return (client, paramList) => ret(client, CancellationToken.None, paramList);
+        }
+
+        Func<HttpClient, CancellationToken, object[], Task<T>> buildCancellableTaskFuncForMethod<T>(RestMethodInfo restMethod)
+            where T : class
+        {
+            return async (client, ct, paramList) => {
                 var factory = BuildRequestFactoryForMethod(restMethod.Name, client.BaseAddress.AbsolutePath);
                 var rq = factory(paramList);
-                var resp = await client.SendAsync(rq);
+
+                var resp = await client.SendAsync(rq, HttpCompletionOption.ResponseHeadersRead, ct);
+
                 if (restMethod.SerializedReturnType == typeof(HttpResponseMessage)) {
                     return resp as T;
                 }
@@ -215,7 +224,12 @@ Func<HttpClient, object[], Task<T>> buildTaskFuncForMethod<T>(RestMethodInfo res
                     throw await ApiException.Create(resp);
                 }
 
-                var content = await resp.Content.ReadAsStringAsync();
+                var ms = new MemoryStream();
+                var fromStream = await resp.Content.ReadAsStreamAsync();
+                await fromStream.CopyToAsync(ms, 4096, ct);
+
+                var bytes = ms.ToArray();
+                var content = Encoding.UTF8.GetString(bytes, 0, bytes.Length);
                 if (restMethod.SerializedReturnType == typeof(string)) {
                     return content as T;
                 }
@@ -227,88 +241,42 @@ Func<HttpClient, object[], Task<T>> buildTaskFuncForMethod<T>(RestMethodInfo res
         Func<HttpClient, object[], IObservable<T>> buildRxFuncForMethod<T>(RestMethodInfo restMethod)
             where T : class
         {
-            var taskFunc = buildTaskFuncForMethod<T>(restMethod);
+            var taskFunc = buildCancellableTaskFuncForMethod<T>(restMethod);
 
-            return (client, paramList) => {
-                var ret = new FakeAsyncSubject<T>();
-
-                taskFunc(client, paramList).ContinueWith(t => {
-                    if (t.Exception != null) {
-                        ret.OnError(t.Exception);
-                    } else {
-                        ret.OnNext(t.Result);
-                        ret.OnCompleted();
-                    }
-                });
-
-                return ret;
-            };
+            return (client, paramList) => 
+                new TaskToObservable<T>(ct => taskFunc(client, ct, paramList));
         }
 
-        class FakeAsyncSubject<T> : IObservable<T>, IObserver<T>
+        class TaskToObservable<T> : IObservable<T>
         {
-            Tuple<bool, T> resultBox;
-            Tuple<bool, Exception> completionResult;
+            Func<CancellationToken, Task<T>> taskFactory;
 
-            List<IObserver<T>> subscriberList = new List<IObserver<T>>();
-
-            public void OnNext(T value)
-            {
-                if (completionResult != null) return;
-
-                var result = Tuple.Create(true, value);
-                Interlocked.Exchange(ref resultBox, result);
-
-                var currentList = default(IObserver<T>[]);
-                lock (subscriberList) { currentList = subscriberList.ToArray(); }
-                foreach (var v in currentList) v.OnNext(value);
-            }
-
-            public void OnError(Exception error)
-            {
-                var final = Interlocked.CompareExchange(ref completionResult, Tuple.Create(true, error), null);
-                if (final != null) return;
-
-                var currentList = default(IObserver<T>[]);
-                lock (subscriberList) { currentList = subscriberList.ToArray(); }
-                foreach (var v in currentList) v.OnError(error);
-
-                subscriberList = null;
-            }
-
-            public void OnCompleted()
+            public TaskToObservable(Func<CancellationToken, Task<T>> taskFactory) 
             {
-                var final = Interlocked.CompareExchange(ref completionResult, Tuple.Create(true, default(Exception)), null);
-                if (final != null) return;
-
-                var currentList = default(IObserver<T>[]);
-                lock (subscriberList) { currentList = subscriberList.ToArray(); }
-                foreach (var v in currentList) v.OnCompleted();
-
-                subscriberList = null;
+                this.taskFactory = taskFactory;
             }
 
             public IDisposable Subscribe(IObserver<T> observer)
             {
-                if (completionResult != null) {
-                    if (resultBox != null) observer.OnNext(resultBox.Item2);
+                var cts = new CancellationTokenSource();
+                taskFactory(cts.Token).ContinueWith(t => {
+                    if (cts.IsCancellationRequested) return;
 
-                    if (completionResult.Item2 != null) {
-                        observer.OnError(completionResult.Item2);
-                    } else { 
-                        observer.OnCompleted();
+                    if (t.Exception != null) {
+                        observer.OnError(t.Exception.InnerExceptions.First());
+                        return;
                     }
-
-                    return new AnonymousDisposable(() => {});
-                }
-
-                lock (subscriberList) { 
-                    subscriberList.Add(observer);
-                }
 
-                return new AnonymousDisposable(() => {
-                    lock (subscriberList) { subscriberList.Remove(observer); }
+                    try {
+                        observer.OnNext(t.Result);
+                    } catch (Exception ex) {
+                        observer.OnError(ex);
+                    }
+
+                    observer.OnCompleted();
                 });
+
+                return new AnonymousDisposable(cts.Cancel);
             }
         }
     }