How to use Micrometer Timer to record duration of async method (returns Mono or Flux)
You could use reactor.util.context.Context
import io.micrometer.core.instrument.Clock;
import io.micrometer.core.instrument.Timer;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import static org.hamcrest.Matchers.is;
public class TestMonoTimer {
private static final Logger LOG = LoggerFactory.getLogger(TestMonoTimer.class);
private static final String TIMER_SAMPLE = "TIMER_SAMPLE";
private static final Timer TIMER = new SimpleMeterRegistry().timer("test");
private static final AtomicBoolean EXECUTION_FLAG = new AtomicBoolean();
@Test
public void testMonoTimer() {
Mono.fromCallable(() -> {
Thread.sleep(1234);
return true;
}).transform(timerTransformer(TIMER))
.subscribeOn(Schedulers.parallel())
.subscribe(EXECUTION_FLAG::set);
Awaitility.await().atMost(2, TimeUnit.SECONDS).untilAtomic(EXECUTION_FLAG, is(true));
Assert.assertTrue(TIMER.totalTime(TimeUnit.SECONDS) > 1);
}
private static <T> Function<Mono<T>, Publisher<T>> timerTransformer(Timer timer) {
return mono -> mono
.flatMap(t -> Mono.subscriberContext()
.flatMap(context -> Mono.just(context.<Timer.Sample>get(TIMER_SAMPLE).stop(timer))
.doOnNext(duration -> LOG.info("Execution time is [{}] seconds",
duration / 1000000000D))
.map(ignored -> t)))
.subscriberContext(context -> context.put(TIMER_SAMPLE, Timer.start(Clock.SYSTEM)));
}
}