Asynchronous cache of values

MyPermissionRepo.cs

void Main()
{
	var repo = new MyPermissionRepo();
	
	//These will all get batched up and sent together
	repo.IsPermissioned("Marcus").Dump("Marcus");
	repo.IsPermissioned("Lee").Dump("Lee");
	repo.IsPermissioned("Merc").Dump("Merc");
	repo.IsPermissioned("Si").Dump("Si");
	
	Thread.Sleep(1000);
	repo.IsPermissioned("Lee").Dump("Lee");	//Already in cache
	Thread.Sleep(1000);
	repo.IsPermissioned("Marcus").Dump("Marcus");	//Already in cache
	Thread.Sleep(1000);
	repo.IsPermissioned("Melinda").Dump("Melinda");	//New request
	
}

// Define other methods and classes here
public class MyPermissionRepo
{
	private readonly ConcurrentDictionary<string, AsyncSubject<bool>> _cache = new ConcurrentDictionary<string, AsyncSubject<bool>>();
	private readonly Subject<string> _keyRequests = new Subject<string>();
	private readonly IObservable<IGroupedObservable<string, bool>> _values;
	
	public MyPermissionRepo()
	{
		_keyRequests.Buffer(TimeSpan.FromMilliseconds(100))
					.Where(keys=>keys.Count>0)
					.Select(keys=>Observable.Start(()=>GetPermissions(keys)))
					.SelectMany(asyncResults=>asyncResults)
					.SelectMany(results=>results)
					.GroupBy(result=>result.Item1, result=>result.Item2)
					.Subscribe(grp=>grp.Take(1).Subscribe(_cache[grp.Key]));
	}
	
	public IObservable<bool> IsPermissioned(string key)
	{
		return _cache.GetOrAdd(key, CreatePermissionFuture);
	}
	
	private AsyncSubject<bool> CreatePermissionFuture(string key)
	{
		//Yuck. How to do without Sujects?
		_keyRequests.OnNext(key);
		return new AsyncSubject<bool>();
	}
	
	private IList<Tuple<string, bool>> GetPermissions(IList<string> keys)
	{
		string.Join(", ", keys.ToArray()).Dump("GetPermissions");
		//Values like M ercedes and M arcus return true.
		return keys.Select(k => Tuple.Create(k, k.StartsWith("M"))).ToList();
	}	
}

Should look at Rxx and see if this is what the Observable Dictionary does