请求完成后，通知可观察者完成

Question

我对反应式编程还比较陌生，即使在阅读了Rx之后，我也在努力思考它。网络在行动。

我有一个服务类，它有两个公共方法Get和GetStream。

GetStream将创建一个可观察的，该可观察的将连接store和update可观察的以便消费者无问题地接收未来的更新。

update可观察到的每分钟执行一次来自providers的符号更新-轮询api请求

Get将创建一个订阅store项目的可观察项，并在更新时与update可观察项连接。问题是，一旦更新完成，就不会调用OnComplete，并且使用者会无限期挂起。我把这个可观察到的东西当作一些地方的阻塞调用和其他地方的非阻塞调用。

我想我可以用ResetEvent逃脱惩罚，但这似乎不是一种非常被动的处理方式。有没有更好的反应性替代方案？也许我试图将一个可观察的模式放在一个应该只返回IEnumerable的地方？

目标是拥有一个可观察的，它在每个指定的时间间隔更新存储。有一种方法可以获取商店的当前状态，包括当前更新。使用另一种方法来获取商店的状态和任何未来的更新。

以下是我目前要做的事情的代码片段。

public class SymbolService : ISymbolService, IDisposable
{
private readonly ILogger _logger = Logger.Create(nameof(SymbolService));

private readonly IObservable<(string Provider, Symbol Symbol)> _update;
private readonly ISymbolStore _store;
private readonly IDisposable _token;
private bool _updating = true;
public SymbolService(SymbolOptions options, ISymbolStore store,
IEnumerable<ISymbolProvider> providers)
{
_store = store;
// Request and aggregate symbols from providers and put
// into store.
var source = providers
.Select(provider => provider
.GetSymbols()
.Select(symbol => (provider.Name, symbol))
.Take(10))
.Merge()
.Do(entry =>
{
_updating = true;
_store.Put(entry);
},
() =>
_updating = false)
.Log(_logger, "Symbol");
// Perform update of symbols every interval
var update = Observable
.Timer(TimeSpan.Zero, options.RefreshInterval)
.SelectMany(_ => source)
.Publish();
_update = update;
_token = update
.Connect();
}
/// <summary>
/// Gets all symbols for a given provider.
/// </summary>
/// <param name="provider">The specified provider.</param>
/// <returns>The observable stream.</returns>
public IObservable<Symbol> Get(string provider)
{
// Get symbols from store by provider and concat stream
// if update is being performed
return _store
.Get(provider)
.Concat(Observable
.If(() => _updating, _update
.TakeWhile(_ => _updating)
.Where(x => x.Provider.Equals(provider))
.Select(x => x.Symbol)));
}
/// <summary>
/// Gets all symbols and future symbols for a given provider.
/// </summary>
/// <param name="provider">The specified provider.</param>
/// <returns>The observable stream.</returns>
public IObservable<Symbol> GetStream(string provider)
{
// Get symbols from store by provider and concat stream
// of all future symbols
return _store
.Get(provider)
.Concat(_update
.Where(x => x.Provider.Equals(provider))
.Select(x => x.Symbol))
.Distinct();
}
public void Dispose()
{
_token?.Dispose();
}
}

非常感谢在这方面的任何帮助。

Answer 1

在再次浏览了这本书之后，我认为BehaviourSubject将非常适合解决方案。我只包含了回答我问题的相关代码片段。

订阅CCD_ 15，该CCD_。

private readonly BehaviorSubject<bool> _updating;
public SymbolService(SymbolOptions options, ISymbolStore store,
IEnumerable<ISymbolProvider> providers)
{
_store = store;
_updating = new BehaviorSubject<bool>(true);
// Request and aggregate symbols from providers and put
// into store.
var source = providers
.Select(provider => provider
.GetSymbols()
.Select(symbol => (provider.Name, symbol)))
.Merge();
// Perform update of symbols every interval
var update = Observable
.Timer(TimeSpan.Zero, options.RefreshInterval)
.Log(_logger, "Updating", LogEventLevel.Information)
.Do(_ => _updating.OnNext(true))
.SelectMany(_ =>
{
return source
.GetSymbols()
.Do(_store.Put)
.Finally(() => _updating.OnNext(false));
})
.Publish();
_update = update;
_token = update
.Connect();
}
public IObservable<Symbol> Get(string provider)
{
// Get symbols from store by provider and concat stream
// if update is being performed
return _store
.Get(provider)
.Concat(_update
.Where(x => x.Provider.Equals(provider))
.Select(x => x.Symbol))
.TakeUntil(_updating.Where(x => x == false));
}