c# - RX: how to pass the latest value from observable into ReactiveCommand -

i want latest value of idstream , use in command execute action.

public iobservable<option<guid>> idstream { get; }  idstream = documentstream.oftype<documentopened().select(x => x.document.id.some()) .merge(documentstream.oftype<documentclosed().select(x => option<guid>.none()));  var savecommand = reactivecommand.create(() => save(id), canexecute); 

i had tried use answer https://stackoverflow.com/a/31168822/7779560 , got this:

var savecommand = reactivecommand.create(() => { }, canexecute); savecommand.withlatestfrom(idstream, (_, id) => id)             .subscribe(id => save(id)); 

and works, can't use isexecuting , thrownexceptions command's functionality in case (they act empty action passed execute during command creation).

upd:

execution order:

1. idstream creating
2. command creating
3. documentstream processes documentopened event (get id value - checked it)
4. savecommand execution

how can achieve it?

upd 2: need await methods inside command body (saveasync, example).

does work you? replay retain latest value published. when command executed grab latest value, take(1) unsubscribes after because need 1 value, , pushes save;

    [test]     public void executeusinglastproducedvalue()     {         subject<string> producer = new subject<string>();         iobservable<bool> canexecute = observable.return(true);         iobservable<string> idstream = producer;         string savecalledwith = string.empty;          func<string, task> saveasync = (id) =>         {             savecalledwith = id;             return task.delay(0);         };          // idstream creating          var connectedidstream =             idstream             .replay(1);          connectedidstream             .connect();          //command creating         var command = reactivecommand.createfromobservable(() =>         {             return connectedidstream                 .take(1)                 .do(async id =>                 {                     await saveasync(id);                 });         }         , canexecute);           //alternate way         //command creating         //var command = reactivecommand.createfromobservable(() =>         //{         //    return connectedidstream         //        .take(1)         //        .selectmany(id => saveasync(id).toobservable());         //}         //, canexecute);           //documentstream processes documentopened event (get id value - checked it)         producer.onnext("something random");         producer.onnext("working");          //at point save still hasen't been called verifiyng it's still empty         assert.areequal(string.empty, savecalledwith);          //trigger execution of command         command.execute(unit.default).subscribe();          //verified saved called called         assert.areequal(savecalledwith, "working");     }