c# - Using Rx Framework for async calls using the void AsyncMethod(Action<T> callback) pattern -

i've seen tons of examples on how use observable.fromasyncpattern() in rx framework simplify async calls, i'm using interface doesn't use standard async pattern of iasyncresult beginxxx/endxxx(iasyncresult), doesn't work me.

the library i'm working exposes async functions callback patter:

void getallobjects(action<list<object>> callback) 

in ideal world i'd turn this:

var isloadingusers = true; var isloadingsystems = true; var isloadingcustomers = true; var isloadingrules = true;  mclient.getallusers(userscallback); mclient.getallcustomers(customerscallback); mclient.getallrules(rulescallback);  // set isloadingxxx variables false in callbacks // once false  mclient.getallsystems(systemscallback); 

into this:

var o = observable.forkjoin(                      observable.start(getallusers()),                      observable.start(getallcustomers()),                      observable.start(getallrules())                     ).finally(() => getallsystems); 

how 1 go turning pattern returns iobservable?

func<iobservable<tret>> fromlistcallbackpattern(action<action<list<tret>>> function) {     return () => {         // use replaysubject if people subscribe *after*         // real method finishes, they'll still items         ret = new replaysubject<tret>();          function((list) => {             // we're going "rebroadcast" list onto subject             // isn't rx'iest way this,             // comprehensible :)             foreach(var v in list) {                 ret.onnext(v);             }             ret.oncompleted();         });          return ret;     }; } 

now, can like:

var getallusers = fromlistcallbackpattern(mclient.getallusers); getallusers().subscribe(x => /* ... */);