From 9a901270988633d67bf70e19ae10497bf6460cfe Mon Sep 17 00:00:00 2001
From: Marcin Grzejszczak <mgrzejszczak@vmware.com>
Date: Tue, 14 Dec 2021 13:19:53 +0100
Subject: [PATCH] Fixed benchmarks

---
 .../SleuthBenchmarkingStreamApplication.java  | 12 +++++-----
 .../jmh/stream/MicroBenchmarkStreamTests.java | 22 +++++++++----------
 2 files changed, 17 insertions(+), 17 deletions(-)

diff --git a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java
index 6c176269d..65fc4288f 100644
--- a/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java
+++ b/benchmarks/src/main/java/org/springframework/cloud/sleuth/benchmarks/app/stream/SleuthBenchmarkingStreamApplication.java
@@ -134,7 +134,7 @@ public class SleuthBenchmarkingStreamApplication {
 	@Bean(name = "myFlux")
 	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "DECORATE_ON_EACH",
 			matchIfMissing = true)
-	public Function<Flux<String>, Flux<String>> onEachFunction() {
+	public Function<Flux<Message<String>>, Flux<Message<String>>> onEachFunction() {
 		log.info("on each function");
 		return new SleuthFunction();
 	}
@@ -142,14 +142,14 @@ public class SleuthBenchmarkingStreamApplication {
 	@Bean(name = "myFlux")
 	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "DECORATE_QUEUES",
 			matchIfMissing = true)
-	public Function<Flux<String>, Flux<String>> decorateQueuesFunction() {
+	public Function<Flux<Message<String>>, Flux<Message<String>>> decorateQueuesFunction() {
 		log.info("decorate queues function");
 		return new SleuthFunction();
 	}
 
 	@Bean(name = "myFlux")
 	@ConditionalOnProperty(value = "spring.sleuth.function.type", havingValue = "DECORATE_ON_LAST")
-	public Function<Flux<String>, Flux<String>> onLastFunction() {
+	public Function<Flux<Message<String>>, Flux<Message<String>>> onLastFunction() {
 		log.info("on last function");
 		return new SleuthFunction();
 	}
@@ -269,18 +269,18 @@ class SleuthNonReactiveFunction implements Function<String, String> {
 
 }
 
-class SleuthFunction implements Function<Flux<String>, Flux<String>> {
+class SleuthFunction implements Function<Flux<Message<String>>, Flux<Message<String>>> {
 
 	private static final Logger log = LoggerFactory.getLogger(SleuthFunction.class);
 
 	static final Scheduler SCHEDULER = Schedulers.newParallel("sleuthFunction");
 
 	@Override
-	public Flux<String> apply(Flux<String> input) {
+	public Flux<Message<String>> apply(Flux<Message<String>> input) {
 		return input.doOnEach(signal -> log.info("Got a message"))
 				.flatMap(s -> Mono.delay(Duration.ofMillis(1), SCHEDULER).map(aLong -> {
 					log.info("Logging [{}] from flat map", s);
-					return s.toUpperCase();
+					return MessageBuilder.withPayload(s.getPayload().toUpperCase()).build();
 				}));
 	}
 
diff --git a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java
index d8e2fdf36..1465af3fd 100644
--- a/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java
+++ b/benchmarks/src/test/java/org/springframework/cloud/sleuth/benchmarks/jmh/stream/MicroBenchmarkStreamTests.java
@@ -56,9 +56,9 @@ import org.springframework.messaging.support.MessageBuilder;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-@Measurement(iterations = 10, time = 1)
-@Warmup(iterations = 10, time = 1)
-@Fork(4)
+@Measurement(iterations = 1, time = 1)
+@Warmup(iterations = 1, time = 1)
+@Fork(1)
 @BenchmarkMode(Mode.SampleTime)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
 @Microbenchmark
@@ -156,14 +156,14 @@ public class MicroBenchmarkStreamTests {
 		public enum Instrumentation {
 
 			// @formatter:off
-			noSleuthSimple(Pair.noSleuth(), function("simple")),
-			sleuthSimpleOnQueues(function("simple"), Pair.onHook()),
-			sleuthSimpleManual(function("simple_manual"), Pair.manual(), functionDisabled(), integrationDisabled()),
-			sleuthSimpleNoFunctionInstrumentationManual(function("simple_manual"), Pair.manual(), functionDisabled(), integrationDisabled()),
-			sleuthSimpleOnEach(function("simple"), Pair.onEach()),
-			sleuthSimpleOnLast(function("simple"), Pair.onLast()),
-			sleuthSimpleWithAroundOnQueues(function("simple_function_with_around")),
-			noSleuthReactiveSimple(function("reactive_simple"), Pair.noSleuth()),
+//			noSleuthSimple(Pair.noSleuth(), function("simple")),
+//			sleuthSimpleOnQueues(function("simple"), Pair.onHook()),
+//			sleuthSimpleManual(function("simple_manual"), Pair.manual(), functionDisabled(), integrationDisabled()),
+//			sleuthSimpleNoFunctionInstrumentationManual(function("simple_manual"), Pair.manual(), functionDisabled(), integrationDisabled()),
+//			sleuthSimpleOnEach(function("simple"), Pair.onEach()),
+//			sleuthSimpleOnLast(function("simple"), Pair.onLast()),
+//			sleuthSimpleWithAroundOnQueues(function("simple_function_with_around")),
+//			noSleuthReactiveSimple(function("reactive_simple"), Pair.noSleuth()),
 			sleuthReactiveSimpleOnQueues(function("DECORATE_QUEUES"), Pair.decorateQueues()),
 			sleuthReactiveSimpleOnEach(function("DECORATE_ON_EACH"), Pair.onEach()),
 			sleuthReactiveSimpleManual(function("reactive_simple_manual"), Pair.manual()),