reactiveui · alexandru-calinoiu · Jun 24, 2014 · Jun 24, 2014 · Jun 26, 2014 · Jun 28, 2014
diff --git a/Refit-Tests/Refit-Tests-Net45.csproj b/Refit-Tests/Refit-Tests-Net45.csproj
@@ -36,6 +36,10 @@
     <Reference Include="Castle.Core">
       <HintPath>..\ext\Net45\Castle.Core.dll</HintPath>
     </Reference>
+    <Reference Include="Microsoft.Reactive.Testing">
+      <HintPath>..\packages\Rx-Testing.2.2.4\lib\net45\Microsoft.Reactive.Testing.dll</HintPath>
+    </Reference>
+    <Reference Include="Microsoft.VisualStudio.QualityTools.UnitTestFramework, Version=10.0.0.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL" />
     <Reference Include="Newtonsoft.Json">
       <HintPath>..\ext\Net45\Newtonsoft.Json.dll</HintPath>
     </Reference>
@@ -52,6 +56,18 @@
       <HintPath>..\packages\Microsoft.Net.Http.2.2.13\lib\net45\System.Net.Http.Primitives.dll</HintPath>
     </Reference>
     <Reference Include="System.Net.Http.WebRequest" />
+    <Reference Include="System.Reactive.Core">
+      <HintPath>..\packages\Rx-Core.2.2.4\lib\net45\System.Reactive.Core.dll</HintPath>
+    </Reference>
+    <Reference Include="System.Reactive.Interfaces">
+      <HintPath>..\packages\Rx-Interfaces.2.2.4\lib\net45\System.Reactive.Interfaces.dll</HintPath>
+    </Reference>
+    <Reference Include="System.Reactive.Linq">
+      <HintPath>..\packages\Rx-Linq.2.2.4\lib\net45\System.Reactive.Linq.dll</HintPath>
+    </Reference>
+    <Reference Include="System.Reactive.PlatformServices">
+      <HintPath>..\packages\Rx-PlatformServices.2.2.4\lib\net45\System.Reactive.PlatformServices.dll</HintPath>
+    </Reference>
     <Reference Include="System.Xml.Linq" />
     <Reference Include="System.Data.DataSetExtensions" />
     <Reference Include="Microsoft.CSharp" />
@@ -61,6 +77,7 @@
   <ItemGroup>
     <Compile Include="RequestBuilder.cs" />
     <Compile Include="RestService.cs" />
+    <Compile Include="support\TestableObserver.cs" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\Refit\Refit-Net45.csproj">
@@ -73,6 +90,7 @@
     </ProjectReference>
   </ItemGroup>
   <ItemGroup>
+    <None Include="app.config" />
     <None Include="packages.config" />
   </ItemGroup>
   <ItemGroup>

diff --git a/Refit-Tests/RequestBuilder.cs b/Refit-Tests/RequestBuilder.cs
@@ -1,12 +1,12 @@
 using System;
-using System.IO;
 using System.Net.Http;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using System.Net;
 using NUnit.Framework;
 using System.Threading;
+using Refit.Tests.support;
 
 namespace Refit.Tests
 {
@@ -211,6 +211,9 @@ public interface IDummyHttpApi
         [Get("/foo/bar/{id}")]
         Task<string> FetchSomeStuff(int id);
 
+        [Get("/foo/bar/{id}")]
+        IObservable<string> FetchSomeStuffObservable(int id);
+
         [Get("/foo/bar/{id}?baz=bamf")]
         Task<string> FetchSomeStuffWithHardcodedQueryParameter(int id);
 
@@ -258,9 +261,17 @@ public class TestHttpMessageHandler : HttpMessageHandler
     {
         public HttpRequestMessage RequestMessage { get; private set; }
 
+        public bool WasSend { get; set; }
+
+        public TestHttpMessageHandler()
+        {
+            WasSend = false;
+        }
+
         protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
         {
             RequestMessage = request;
+            WasSend = true;
             return Task.FromResult(new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent("test") });
         }
     }
@@ -481,5 +492,35 @@ public void SuchFlexibleContentTypeWow()
             Assert.IsNotNull(output.Content.Headers.ContentType, "Headers include Content-Type header");
             Assert.AreEqual("text/dson", output.Content.Headers.ContentType.MediaType, "Content-Type header has the expected value");
         }
+
+        [Test]
+        public void ObservableMethodDoesNotGetDispachedOnCreation()
+        {
+            var fixture = new RequestBuilderImplementation(typeof(IDummyHttpApi));
+            var factory = fixture.BuildRestResultFuncForMethod("FetchSomeStuffObservable");
+            var testHttpMessageHandler = new TestHttpMessageHandler();
+
+            factory(new HttpClient(testHttpMessageHandler) { BaseAddress = new Uri("http://api/") }, new object[] { 42 });
+
+            Assert.IsFalse(testHttpMessageHandler.WasSend);
+        }
+
+        [Test]
+        public void ObservableMethodDoesGetDispachedOnSubscription()
+        {
+            var fixture = new RequestBuilderImplementation(typeof(IDummyHttpApi));
+            var factory = fixture.BuildRestResultFuncForMethod("FetchSomeStuffObservable");
+            var testHttpMessageHandler = new TestHttpMessageHandler();
+
+            var observable = (IObservable<string>) factory(new HttpClient(testHttpMessageHandler) { BaseAddress = new Uri("http://api/") }, new object[] { 42 });
+            var semaphore = new Semaphore(0, 1);
+            var testableObserver = new TestableObserver<string>(() => semaphore.Release());
+            observable.Subscribe(testableObserver);
+            semaphore.WaitOne(100);
+
+            Assert.IsTrue(testHttpMessageHandler.WasSend);
+            Assert.IsTrue(testableObserver.OnNextWasCalled);
+        }
+
     }
 }
diff --git a/Refit-Tests/RestService.cs b/Refit-Tests/RestService.cs
@@ -2,9 +2,12 @@
 using System.Net.Http;
 using NUnit.Framework;
 using System.Threading.Tasks;
+using System.Threading;
+using Refit.Tests.support;
 
 namespace Refit.Tests
 {
+
     public class User
     {
         public string login { get; set; }
@@ -48,6 +51,16 @@ public interface IGitHubApi
         Task<HttpResponseMessage> GetIndex();
     }
 
+    [Headers("User-Agent: Refit Integration Tests")]
+    public interface IObservableGitHubApi
+    {
+        [Get("/users/{username}")]
+        IObservable<User> GetUser(string userName);
+
+        [Get("/")]
+        IObservable<HttpResponseMessage> GetIndex();    
+    }
+
     public class RootObject
     {
         public string _id { get; set; }
@@ -86,6 +99,34 @@ public void ShouldRetHttpResponseMessage()
             Assert.IsTrue(result.Result.IsSuccessStatusCode);
         }
 
+        [Test]
+        public void HitTheGitHubUserApiObservable()
+        {
+            var fixture = RestService.For<IObservableGitHubApi>("https://api.github.com");
+            var result = fixture.GetUser("octocat");
+            var semaphore = new Semaphore(0, 2);
+            var testableObserver = new TestableObserver<User>(() => semaphore.Release());
+            result.Subscribe(testableObserver);
+            semaphore.WaitOne(3000);
+
+            Assert.IsTrue(testableObserver.OnNextWasCalled);
+            Assert.IsTrue(testableObserver.OnCompleteWasCalled);
+        }
+
+        [Test]
+        public void HitTheGitHubUserApiErrorObservable()
+        {
+            var fixture = RestService.For<IObservableGitHubApi>("https://api.github.com");
+            var result = fixture.GetUser("some_random_user_that_I_hope_does_not_exist");
+
+            var semaphore = new Semaphore(0, 1);
+            var testableObserver = new TestableObserver<User>(() => semaphore.Release());
+            result.Subscribe(testableObserver);
+            semaphore.WaitOne(1000);
+
+            Assert.IsTrue(testableObserver.OnErrorWasCalled);
+        }
+
         [Test]
         public void HitTheNpmJs()
         {

diff --git a/Refit-Tests/app.config b/Refit-Tests/app.config
@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="utf-8"?>
+<configuration>
+  <runtime>
+    <assemblyBinding xmlns="urn:schemas-microsoft-com:asm.v1">
+      <dependentAssembly>
+        <assemblyIdentity name="System.Net.Http.Primitives" publicKeyToken="b03f5f7f11d50a3a" culture="neutral" />
+        <bindingRedirect oldVersion="0.0.0.0-4.2.13.0" newVersion="4.2.13.0" />
+      </dependentAssembly>
+    </assemblyBinding>
+  </runtime>
+</configuration>
diff --git a/Refit-Tests/packages.config b/Refit-Tests/packages.config
@@ -4,4 +4,10 @@
   <package id="Microsoft.Bcl.Build" version="1.0.8" targetFramework="net45" />
   <package id="Microsoft.Net.Http" version="2.2.13" targetFramework="net45" />
   <package id="NUnit" version="2.6.2" targetFramework="net45" />
+  <package id="Rx-Core" version="2.2.4" targetFramework="net45" />
+  <package id="Rx-Interfaces" version="2.2.4" targetFramework="net45" />
+  <package id="Rx-Linq" version="2.2.4" targetFramework="net45" />
+  <package id="Rx-Main" version="2.2.4" targetFramework="net45" />
+  <package id="Rx-PlatformServices" version="2.2.4" targetFramework="net45" />
+  <package id="Rx-Testing" version="2.2.4" targetFramework="net45" />
 </packages>
diff --git a/Refit-Tests/support/TestableObserver.cs b/Refit-Tests/support/TestableObserver.cs
@@ -0,0 +1,47 @@
+namespace Refit.Tests.support
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Reactive;
+
+    using Microsoft.Reactive.Testing;
+
+    public class TestableObserver<T> : ITestableObserver<T>
+    {
+        private readonly Action _afterAction;
+
+        public bool OnNextWasCalled { get; set; }
+
+        public bool OnErrorWasCalled { get; set; }
+
+        public bool OnCompleteWasCalled { get; set; }
+
+        public TestableObserver(Action afterAction)
+        {
+            _afterAction = afterAction;
+            >
+            >
+            >
+        }
+
+        public void OnNext(T value)
+        {
+            >
+            _afterAction();
+        }
+
+        public void OnError(Exception error)
+        {
+            >
+            _afterAction();
+        }
+
+        public void OnCompleted()
+        {
+            >
+            _afterAction();
+        }
+
+        public IList<Recorded<Notification<T>>> Messages { get; private set; }
+    }
+}
diff --git a/Refit/RequestBuilderImplementation.cs b/Refit/RequestBuilderImplementation.cs
@@ -216,91 +216,41 @@ Func<HttpClient, object[], IObservable<T>> buildRxFuncForMethod<T>(RestMethodInf
             var taskFunc = buildTaskFuncForMethod<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();
-                    }
-                });
+                var ret = new FakeAsyncSubject<T>(taskFunc, client, paramList);
 
                 return ret;
             };
         }
 
-        class CompletionResult 
+        class FakeAsyncSubject<T> : IObservable<T>
         {
-            public bool IsCompleted { get; set; }
-            public Exception Error { get; set; }
-        }
-
-        class FakeAsyncSubject<T> : IObservable<T>, IObserver<T>
-        {
-            bool resultSet;
-            T result;
-            CompletionResult completion;
-            List<IObserver<T>> subscriberList = new List<IObserver<T>>();
-
-            pub
10000
lic void OnNext(T value)
-            {
-                if (completion == null) return;
-
-                result = value;
-                resultSet = true;
-
-                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 completion, new CompletionResult() { IsCompleted = false, Error = error }, null);
-                if (final.IsCompleted) return;
-
-                var currentList = default(IObserver<T>[]);
-                lock (subscriberList) { currentList = subscriberList.ToArray(); }
-                foreach (var v in currentList) v.OnError(error);
-
-                final.IsCompleted = true;
-            }
+            Func<HttpClient, object[], Task<T>> taskFunc;
+            HttpClient client;
+            object[] paramList;
 
-            public void OnCompleted()
+            public FakeAsyncSubject(Func<HttpClient, object[], Task<T>> taskFunc, HttpClient client, object[] paramList)
             {
-                var final = Interlocked.CompareExchange(ref completion, new CompletionResult() { IsCompleted = false, Error = null }, null);
-                if (final.IsCompleted) return;
-
-                var currentList = default(IObserver<T>[]);
-                lock (subscriberList) { currentList = subscriberList.ToArray(); }
-                foreach (var v in currentList) v.OnCompleted();
-
-                final.IsCompleted = true;
+                this.taskFunc = taskFunc;
+                this.client = client;
+                this.paramList = paramList;
             }
 
             public IDisposable Subscribe(IObserver<T> observer)
             {
-                if (completion != null) {
-                    if (completion.Error != null) {
-                        observer.OnError(completion.Error);
-                        return new AnonymousDisposable(() => {});
+                taskFunc(client, paramList).ContinueWith(t =>
+                {
+                    if (t.Exception != null)
+                    {
+                        observer.OnError(t.Exception);
+                    }
+                    else
+                    {
+                        observer.OnNext(t.Result);
+                        observer.OnCompleted();
                     }
-
-                    if (resultSet) observer.OnNext(result);
-                    observer.OnCompleted();
-
-                    return new AnonymousDisposable(() => {});
-                }
-
-                lock (subscriberList) { 
-                    subscriberList.Add(observer);
-                }
-
-                return new AnonymousDisposable(() => {
-                    lock (subscriberList) { subscriberList.Remove(observer); }
                 });
+
+                return new AnonymousDisposable(() => { });
             }
         }
     }

diff --git a/packages/Rx-Core.2.2.4/Rx-Core.2.2.4.nupkg b/packages/Rx-Core.2.2.4/Rx-Core.2.2.4.nupkg