c# - Reactive Extensions .MaxBy -

why

var = observable.interval(timespan.fromseconds(1))     .publish();  a.subscribe(o => {     console.writeline("test"); }); a.connect(); 

fire, not

var = observable.interval(timespan.fromseconds(1))     .maxby(o=>o)     .publish();  a.subscribe(o => {     console.writeline("test"); }); a.connect(); 

i trying use maxby in different scenario can't above work.

this more complex example

var _telemetrybatchobservable = observable.fromeventpattern<devicestatestreameventarg>(         ev => devicestatestreamevent += ev,         ev => devicestatestreamevent -= ev)     .synchronize()     .groupby(o => o.eventargs.deviceid)     .select(o => o.maxby(i => i.eventargs.datetimeoffset))     .selectmany(o => o.select(i => i))     .selectmany(o => o.select(i => i))     .buffer(timespan.frommilliseconds(5000), 100)     .publish(); 

max , maxby emit single value, when source observable terminated. if have non-terminating source, they'll never emit.

try code counter-example:

var = observable.interval(timespan.fromseconds(1))     .take(3) //causes termination after 3 values     .maxby(o => o)     .publish(); 

wiki