如何从Web API获取IObservable

Question

我有一个简单的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();

我的问题是我没有从服务器获取任何数据.我得到一个'空'可观察.我对我的控制器写了一个单元测试,它确实给了数据项.

有任何建议请帮忙.无法理解我哪里出错了.服务器或客户端上没有错误.

Answer 1

你是一个有点雄心勃勃的人,希望IObservable<T>自动通过电线传输.我担心WebAPI不会为你做那件事.

你看到的是默认的json序列化器输出属性的结果IObservable<T>- 并且没有,所以你得到空的大括号.

您的单元测试有效,因为它全部在内存中 - 没有发生序列化/反序列化.

有一些方法可以使用HttpResponseMessage的StreamContent属性来传输可以桥接到/来自的结果IObservable<T>- 但它不是真正的惯用WebApi.WebApi的异步支持实际上是针对服务器上具有单项响应的请求的异步处理,而不是连续返回流事件.

最重要的是,我认为WebApi(至少在撰写本文时)是错误的技术选择.你最好看一下为这种场景而构建的SignalR,它包含在当前的ASP.NET版本中.它具有javascript和.NET客户端支持,您可以IObservable<T>非常轻松地桥接.有些人已经看过这个,比如在这篇帖子中运动示例代码.

一些消息传递中间件,比如my-Channels Nirvana(编辑:自从Terracotta收购并包装到Universal Messaging中.一些示例代码可以在他们的文档中找到.),而SQL Server StreamInsight等CEP解决方案也有开箱即用的IObservable支持.