我有一个简单的Web API,它返回一个Iobservable.我正在使用HttpClient来获取Observable,以便我可以订阅它.我的问题是订阅时返回的Iobservable发出一个’空’结果.
服务器
public IObservable<DataItem> GetDataItems()
{
return Observable.Generate(0,i => i < 10,i => i + 1,i => new DataItem
{
Id = i,Name = String.Format("Storage{0}",i)
});
}
客户
public IObservable<DataItem> GetDataItems()
{
using (HttpClient apiClient = new HttpClient())
{
apiClient.BaseAddress = new Uri("http://localhost:9001");
apiClient.DefaultRequestHeaders.Add("x-user-authentication","xxxxxx");
return apiClient
.GetAsync("api/xxxx/yyyy").Result.Content
.ReadAsAsync<DataItem>().ToObservable();
}
}
var source = GetDataItems();
List<DataItem> items = new List<DataItem>();
IDisposable consoleSubscription = source.Subscribe(
x => Console.WriteLine("{0}:{1}",x.Id,x.Name),ex => Console.WriteLine("OnError : {0} ",ex.Message),() => Console.WriteLine("Encountered End of Stream")
);
consoleSubscription.Dispose();
我的问题是我没有从服务器获取任何数据.我得到一个’空’可观察.我对我的控制器写了一个单元测试,它确实给了数据项.
有任何建议请帮忙.无法理解我哪里出错了.服务器或客户端上没有错误.
解决方法
你有点雄心勃勃,期待一个IObservable< T>自动流过电线.我担心WebAPI不会为你做那件事.
你看到的是默认的json序列化器输出IObservable< T>的属性的结果. – 没有,所以你得到空的支架.
您的单元测试有效,因为它全部在内存中 – 没有发生序列化/反序列化.
有一些方法可以使用HttpResponseMessage的StreamContent属性来流式传输结果,您可以将结果桥接到IObservable< T>. – 但它并不是真正惯用的WebApi. WebApi的异步支持实际上旨在通过服务器上的单项响应异步处理请求,而不是连续返回流事件.
最重要的是,我认为WebApi(至少在撰写本文时)是错误的技术选择.你最好不要看SignalR是为这种场景而构建的,并且包含在当前的ASP.NET版本中.它具有javascript和.NET客户端支持,您可以桥接到IObservable< T>相当容易.有些人已经看过这个,比如in this post sporting example code.
一些消息传递中间件,如my-Channels Nirvana(编辑：自从Terracotta买断并包装到Universal Messaging中,example code可以在他们的文档中找到.),而像sql Server StreamInsight这样的CEP解决方案也具有开箱即用的IObservable支持.
