恢复多播可观察量 [英] Resurrect multicast observables

问题描述

这是使用Rx的商业系统中相当常见的模式（根据我的经验）。

This is a fairly common pattern (in my experience) in business systems using Rx.

class Cache
{
	private readonly Dictionary<string, IObservable<int>> _cache = new Dictionary<string, IObservable<int>>();
	
	public IObservable<int> Get<T>(string key)
	{
		lock(_cache)
		{
			IObservable<int> stream;
			if(!_cache.TryGetValue(key, out stream))
			{
				stream = Create(key).Multicast(new Subject<int>()).RefCount();
				_cache[key] = stream;
			}			
			return stream;
		}
	}
	
	private IObservable<int> Create(string key)
	{
		// get some observable based on the key
		throw new NotImplementedException();
	}
}

推荐答案

解决方案1 ​​

class Cache
{
	private readonly Dictionary<string, IObservable<int>> _cache = new Dictionary<string, IObservable<int>>();
	
	public IObservable<int> Get<T>(string key)
	{
		lock(_cache)
		{
			IObservable<int> stream;
			if(!_cache.TryGetValue(key, out stream))
			{
				stream = Create(key)
				.Do(x => { }, ex => 
				{
					// remove observable on error
					lock(_cache)
					{
						_cache.Remove(key);
					}
				})
				.Multicast(new Subject<int>()).RefCount();
				_cache[key] = stream;
			}			
			return stream;
		}
	}
	
	private IObservable<int> Create(string key)
	{
		// get some observable based on the key
		throw new NotImplementedException();
	}
}

这篇关于恢复多播可观察量的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！