Merge branch 'main' into resume_consumer

bloomberg · Jan 12, 2024 · 25fe41f · 25fe41f
2 parents 6ba18e2 + 20ca5fb
commit 25fe41f
Showing 1 changed file with 19 additions and 7 deletions.
diff --git a/examples/rmqperftest/rmqperftest_runner.cpp b/examples/rmqperftest/rmqperftest_runner.cpp
@@ -40,6 +40,10 @@
 namespace BloombergLP {
 namespace rmqperftest {
 namespace {
+
+typedef bsl::pair<bsl::shared_ptr<rmqa::Producer>, bsl::string>
+    ProducerRoutingKeyPair;
+
 class ConsumerCallback {
   public:
     ConsumerCallback(const ConsumerArgs& args,
@@ -96,6 +100,18 @@ bsl::string_view queueNameOrRoutingKey(bsl::string_view routingKey,
     }
 }
 
+void finalizeProducers(bsl::vector<ProducerRoutingKeyPair>& producers)
+{
+    bsl::cout << producers.size() << " producers finished, exiting producers"
+              << bsl::endl;
+    for (bsl::vector<ProducerRoutingKeyPair>::iterator it = producers.begin();
+         it != producers.end();
+         ++it) {
+        it->first->waitForConfirms();
+    }
+    producers.clear();
+}
+
 } // namespace
 
 int Runner::run(const PerfTestArgs& args)
@@ -179,8 +195,6 @@ int Runner::run(const PerfTestArgs& args)
     }
 
     // Create producer
-    typedef bsl::pair<bsl::shared_ptr<rmqa::Producer>, bsl::string>
-        ProducerRoutingKeyPair;
     bsl::vector<ProducerRoutingKeyPair> producers;
     {
         bsl::vector<rmqt::QueueHandle>::const_iterator queues;
@@ -311,12 +325,10 @@ int Runner::run(const PerfTestArgs& args)
 
             producerMessagesSent += 1;
 
-            if ((int)producerMessagesSent ==
+            if ((int)producerMessagesSent >=
                 args.producer.producerMessageCount) {
-                bsl::cout << producers.size()
-                          << " producers finished, exiting producers"
-                          << bsl::endl;
-                producers.clear();
+
+                finalizeProducers(producers);
             }
         }
     }