From 513f19370a8c4284b8448d8f4c5a063a1a9e26d1 Mon Sep 17 00:00:00 2001
From: Chris Eager <chris@signal.org>
Date: Thu, 26 Sep 2024 09:38:39 -0500
Subject: [PATCH] Add Experiment#compareMonoResult

---
 .../textsecuregcm/experiment/Experiment.java             | 9 +++++++++
 .../textsecuregcm/storage/MessagesCache.java             | 3 +--
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/experiment/Experiment.java b/service/src/main/java/org/whispersystems/textsecuregcm/experiment/Experiment.java
index f049a84a4..4cf24bc8d 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/experiment/Experiment.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/experiment/Experiment.java
@@ -18,6 +18,7 @@ import java.util.concurrent.Executor;
 import java.util.function.Supplier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Mono;
 
 /**
  * An experiment compares the results of two operations and records metrics to assess how frequently they match.
@@ -70,6 +71,14 @@ public class Experiment {
     this.experimentNullMismatchTimer = experimentNullMismatchTimer;
   }
 
+  public <T> void compareMonoResult(final T expected, final Mono<T> experimentMono) {
+    final Timer.Sample sample = Timer.start();
+
+    experimentMono.subscribe(
+        actual -> recordResult(expected, actual, sample),
+        cause -> recordError(cause, sample));
+  }
+
   public <T> void compareFutureResult(final T expected, final CompletionStage<T> experimentStage) {
     final Timer.Sample sample = Timer.start();
 
diff --git a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
index a943be541..d26bc0b7d 100644
--- a/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
+++ b/service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesCache.java
@@ -463,8 +463,7 @@ public class MessagesCache extends RedisClusterPubSubAdapter<String, String> imp
             }
           });
 
-      experiment.compareFutureResult(mrmMessage.toBuilder().clearSharedMrmKey().build(),
-          mrmMessageMono.toFuture());
+      experiment.compareMonoResult(mrmMessage.toBuilder().clearSharedMrmKey().build(), mrmMessageMono);
 
       return mrmMessageMono;
     } else {