java - RX Observable - measure execution time (even when nested) -

i'm using appended class measure execution time of observables. works fine, long don't measure nested observables...

working fine

observable<t> obs1 = ...; observable<t> obs2 = ...;  obs1     .compose(rxmeasure.applymeasurement(t.class, "tag1"))     .subscribe();  obs2     .compose(rxmeasure.applymeasurement(t.class, "tag2"))     .subscribe(); 

not working

observable<t> obs3 = ...;  observable<t> obs = obs1     .flatmap(resul1 -> obs2)     .flatmap(result2 -> obs3)     .subscribe(); 

this leads result, observables subscribed @ start , means, measurement not correct anymore. here i'm trying execute 3 observables sequentially...

what want

i want measure function that

• is not breaking chain
• is measuring execution time of observable (something time between start/end of observable instead of subscribe/terminate)

any ideas?

rxmeasure class

public class rxmeasure {     private static boolean menabled = true;      public static void setenabled(boolean enabled)     {         menabled = enabled;     }      public static <t> observable.transformer<t, t> applymeasurement(class clazz, string tag)     {         return observable -> measure(observable, clazz, tag);     }      public static <t> observable<t> measure(observable<t> observable, class clazz, string tag)     {         if (!menabled)             return observable;          longholder start = new longholder(0);         return observable                 .doonsubscribe(() -> start.set(system.currenttimemillis()))                 .doonterminate(() -> l.d(clazz, "[" + tag + "] duration: " + string.valueof(system.currenttimemillis() - start.get()) + "ms"));     } } 

hum... if want run sequentially , not have data dependencies, won't work?

observable.concatmap(   obs1.compose(applymeasurement(t.class, "tag1")),   obs2.compose(applymeasurement(t.class, "tag2")),   obs3.compose(applymeasurement(t.class, "tag3")), ).subscribe(); 

otherwise, need better define term "execution time", observable may subscribed multiple times.

edit: looks plain concat solution:

observable<observable<t>> obss = observable.just(   obs1.compose(applymeasurement(t.class, "tag1")),   obs2.compose(applymeasurement(t.class, "tag2")),   obs3.compose(applymeasurement(t.class, "tag3")), );  observable.concat(obss).subscribe(); 

my understanding docs , rxjava source code concat subscribe after previous observable completes / gets unsibscribed.