如何在rx.net中组合GroupedObservables? [英] How to combine GroupedObservables in rx.net?
问题描述
我可以观察到，我使用GroupBy来获取许多流.我实际上想要每个子流上的Scan结果.假设可观察到的是产品价格之上，而扫描结果是每种产品类型的平均价格.
I have one observable that I use GroupBy on to get a number of streams. I actually want a Scan result over each sub-stream. Let's say the observable is over product prices and the scan result is average price per product type.
我还有另一个与那些产品"有关的事件(比如说显示产品价格"事件)，我想将其与上一类产品的最新产品价格结合起来.因此，需要将每个组的Scan输出与事件流的每个元素结合起来，以获取该事件产品的最新平均价格.
I have another stream of events pertaining to those 'products' (let's say "show product price" events) and I want to combine it with the previous stream's latest product price. So the Scan output per group needs to be combined with each element of the event stream to get the latest average price for that event's product.
由于某种原因，我无法获得正确的语法，并且整天都在不知所措.有人可以帮忙吗?
For some reason I cannot get the right syntax and I have been bashing away at this all day. Can someone please help?
我正在添加以下代码以说明大致意图.
I am adding the code below to illustrate the approximate intent.
public class Node
{
private List<int> Details = new List<int>();
public void AddInfo(int x)
{
Details.Add(x );
}
public Node(int x)
{
Details.Add(x);
}
public int Index => Details[0]%10; //just to simplify the grouping and debugging
public int Latest => Details.Last();
}
public class Message
{
private static Random _random = new Random();
public int MessageNodeInfo { get; private set; }
public Message()
{
MessageNodeInfo = _random.Next();
}
}
public class AccumulatingInfoTest
{
private static Random _random=new Random();
private IObservable<Message> MessageStream()
{
TimeSpan timeSpan = TimeSpan.FromSeconds(0.5);
var ret= Observable.Generate(0,
_ => { return true; },
_ => { return 0; },
_ => { return new Message(); },
_=> timeSpan)
.Publish()
.RefCount();
return ret;
}
public class ArbitraryCommonClass
{
public int K { get; set; }
public Message M { get; set; }
public Node D { get; set; }
public ArbitraryCommonClass Combine(ArbitraryCommonClass a)
{
return new ArbitraryCommonClass()
{
K = this.K,
M = this.M ?? a.M,
D = this.D ?? a.D
};
}
}
public void Start()
{
var inputStream = MessageStream();
inputStream.Subscribe(y => Console.WriteLine("Input: K " + y.MessageNodeInfo % 10 + " V " + y.MessageNodeInfo));
var nodeInfoStream = inputStream
.Select(nodeInfo => new Node(nodeInfo.MessageNodeInfo))
.GroupBy(node => node.Index)
.Select(groupedObservable => new
{
Key = groupedObservable.Key,
Observable = groupedObservable
.Scan(
(nodeAcc, node) => { nodeAcc.AddInfo(node.Latest); return nodeAcc; }
)
.Select(a => new ArbitraryCommonClass() { K = a.Index, M = (Message)null, D = a })
}
);
var groupedMessageStream =
inputStream
.GroupBy(
m => new Node(m.MessageNodeInfo).Index
)
.Select(a => new
{
Key =a.Key,
Observable = a.Select(b => new ArbitraryCommonClass() { K = a.Key, M = b, D = null })
});
var combinedStreams = nodeInfoStream
.Merge(groupedMessageStream)
.GroupBy(s => s.Key)
.Select(grp => grp
.Scan(
(state, next) => new { Key = state.Key, Observable = Observable.CombineLatest(state.Observable, next.Observable, (x, y) => { return x.Combine(y); }) }
)
)
.Merge()
.SelectMany(x => x.Observable.Select(a=>a));
combinedStreams.Where(x=>x.M!=null).Subscribe(x => Console.WriteLine(x.K + " " + x.M.MessageNodeInfo + " " + x.D.Latest));
}
}
推荐答案
假定以下类别:
public class Product
{
public string Type { get; set; } = "Default";
public decimal Price { get; set; }
}
这是GroupBy和Scan的使用(显示按类型分组的平均产品价格).诀窍是在可观察的分组上按Select键进入各个分组，执行任何操作，然后(大概)将它们合并回去.您可以将Select和Merge折叠为一个SelectMany，但分开时更易于阅读:
Here's a use of GroupBy with Scan (shows the average product price grouped by type). The trick is to Select over the grouped observable to get to the individual groupings, do whatever, then (presumably) merge them back together. You could collapse the Select and the Merge into a single SelectMany, but it can be easier to read when separated:
var productSubject = new Subject<Product>();
var printSignal = new Subject<Unit>();
var latestAverages = productSubject.GroupBy(p => p.Type)
.Select(g => g
.Scan((0, 0.0m), (state, item) => (state.Item1 + 1, state.Item2 + item.Price)) //hold in state the count and the running total for each group
.Select(t => (g.Key, t.Item2 / t.Item1)) //divide to get the average
)
.Merge()
.Scan(ImmutableDictionary<string, decimal>.Empty, (state, t) => state.SetItem(t.Key, t.Item2)); //Finally, cache the average by group.
printSignal.WithLatestFrom(latestAverages, (_, d) => d)
.Subscribe(avgs =>
{
foreach (var avg in avgs)
{
Console.WriteLine($"ProductType: {avg.Key}. Average: {avg.Value}");
}
Console.WriteLine();
});
var productsList = new List<Product>()
{
new Product { Price = 1.00m },
new Product { Price = 2.00m },
new Product { Price = 3.00m },
new Product { Price = 2.00m, Type = "Alternate" },
new Product { Price = 4.00m, Type = "Alternate" },
new Product { Price = 6.00m, Type = "Alternate" },
};
productsList.ForEach(p => productSubject.OnNext(p));
printSignal.OnNext(Unit.Default);
productSubject.OnNext(new Product { Price = 4.0m });
printSignal.OnNext(Unit.Default);
productSubject.OnNext(new Product { Price = 8.0m, Type = "Alternate" });
printSignal.OnNext(Unit.Default);
这使用nuget包System.Collections.Immutable.
This uses nuget package System.Collections.Immutable.
这篇关于如何在rx.net中组合GroupedObservables?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
