petefisher78/spark
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix TableUpdate 'SelectKeyValue' extensions (subscriptions cannot share the same cache)

Browse Source
This commit is contained in:
John Colvin committed 2019-07-25 17:46:13 -05:00
1 parent adac484f25
commit 6430f6e2df
1 file changed
+34 -33
Download Patch File Download Diff File Expand all files Collapse all files
src/client/spark_api/api/Update.cs
+34 -33
View File
@@ -719,45 +719,46 @@ namespace bts.api.core {
// Constructs a new dictionary so not as cheap as Select on the Update<> type
public static IObservable<TableUpdate<T2, TKey>>
Select<T1, T2, TKey>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey> key) where TKey : IEquatable<TKey> {
var cache = ImmutableDictionary<TKey, T2>.Empty;
return src.Select(update => {
var added = update.added.Select(map);
var removed = update.removed.Select(map);
var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
var builder = cache.ToBuilder();
builder.RemoveRange(removed.Select(key));
builder.AddRange(added.Select(t => new KeyValuePair<TKey,T2>(key(t), t)));
foreach (var (o,n) in changes) {
Debug.Assert(key(o).Equals(key(n)));
builder[key(o)] = n;
}
cache = builder.ToImmutable();
Select<T1, T2, TKey>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey> key) where TKey : IEquatable<TKey> =>
Observable.Create<TableUpdate<T2, TKey>>(observer => {
var cache = ImmutableDictionary<TKey, T2>.Empty;
return src.Subscribe(update => {
var added = update.added.Select(map);
var removed = update.removed.Select(map);
var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
var builder = cache.ToBuilder();
builder.RemoveRange(removed.Select(key));
builder.AddRange(added.Select(t => new KeyValuePair<TKey, T2>(key(t), t)));
foreach (var (o, n) in changes) {
Debug.Assert(key(o).Equals(key(n)));
builder[key(o)] = n;
}
cache = builder.ToImmutable();
return new TableUpdate<T2, TKey>(cache, added, changes, removed, key);
observer.OnNext(new TableUpdate<T2, TKey>(cache, added, changes, removed, key));
});
});
}
// Constructs a new dictionary so not as cheap as Select on the Update<> type
public static IObservable<TableUpdate<T2, TKey2>>
SelectKeyValue<T1, T2, TKey, TKey2>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey2> key) where TKey : IEquatable<TKey> where TKey2 : IEquatable<TKey2> {
var cache = ImmutableDictionary<TKey2, T2>.Empty;
return src.Select(update => {
var added = update.added.Select(map);
var removed = update.removed.Select(map);
var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
var builder = cache.ToBuilder();
builder.RemoveRange(removed.Select(key));
builder.AddRange(added.Select(t => new KeyValuePair<TKey2,T2>(key(t), t)));
foreach (var (o,n) in changes) {
Debug.Assert(key(o).Equals(key(n)));
builder[key(o)] = n;
}
cache = builder.ToImmutable();
return new TableUpdate<T2, TKey2>(cache, added, changes, removed, key);
SelectKeyValue<T1, T2, TKey, TKey2>(this IObservable<TableUpdate<T1, TKey>> src, Func<T1, T2> map, Func<T2, TKey2> key) where TKey : IEquatable<TKey> where TKey2 : IEquatable<TKey2> =>
Observable.Create<TableUpdate<T2, TKey2>>(observer => {
var cache = ImmutableDictionary<TKey2, T2>.Empty;
return src.Subscribe(update => {
var added = update.added.Select(map);
var removed = update.removed.Select(map);
var changes = update.changes.Select(ch => new Change<T2>(map(ch.Old), map(ch.New)));
var builder = cache.ToBuilder();
builder.RemoveRange(removed.Select(key));
builder.AddRange(added.Select(t => new KeyValuePair<TKey2, T2>(key(t), t)));
foreach (var (o, n) in changes) {
Debug.Assert(key(o).Equals(key(n)));
builder[key(o)] = n;
}
cache = builder.ToImmutable();
observer.OnNext(new TableUpdate<T2, TKey2>(cache, added, changes, removed, key));
});
});
}
public static IObservable<TableUpdate<(TLeft left, TRight right), TKey>> LeftJoin<TLeft, TRight, TKey>(this IObservable<TableUpdate<TLeft, TKey>> left_source, IObservable<TableUpdate<TRight, TKey>> right_source,
Func<TLeft, TKey> left_key, Func<TRight, TKey> right_key, TRight right_default = default)