c# - Using reactive extensions how I can create a dynamic list with values who can expire -

scenario

i'm receiving differents notification ids every 100 ms (source1) , need put every id in cache specific received date, if id came twice update date. after need search information ids invoking service, when receive information on app, need show ordered received date, updating screen every 5 seconds. if id not refreshed in range of 10 seconds source1, needs change of state display in different category or state

problem

i'm trying use reactive extensions solve problem, i'm not sure if it's correct technology because:

• i don't know should have cache , how manage states
• how best way manage concurrency in general invoke external service in meantime can receive more ids new or old

at end have clean list result of information can see elements being updated , of them not.

can me? thanks

it sounds .scan operator might meet needs.

try this:

var source = new subject<int>();  var query =     source         .scan(new dictionary<int, datetime>(), (a, x) =>         {             a[x] = datetime.now;             return new dictionary<int, datetime>(a);         })         .select(x => x.orderbydescending(y => y.value)); 

you can test following code:

var values = new [] { 1, 2, 1, 3, 2, 1 };  observable     .interval(timespan.fromseconds(5.0))     .take(values.length)     .select(x => values[x])     .subscribe(source); 

i get:

it's better though use immutabledictionary query looks this:

var query =     source         .scan(             new dictionary<int, datetime>().toimmutabledictionary(),             (a, x) => a.setitem(x, datetime.now))         .select(x => x.orderbydescending(y => y.value)); 

---

var query =     source         .scan(immutabledictionary<int, datetime>.empty, (a, x) => a.setitem(x, datetime.now))         .select(x => observable.interval(timespan.fromseconds(5.0)).select(y => x).startwith(x))         .switch()         .select(x => x.orderbydescending(y => y.value)); 

try query - continues produce values when source does, every 5 seconds after latest value come out repeats last item (unless source produces value , reset 5 second timer).