### RMQ C++ Client Usage Example

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

Demonstrates setting up a RabbitMQ context, defining topology, creating producers and consumers, sending messages with confirmations, and handling received messages. Ensure the RabbitContext lives longer than objects created from it. The producer's send method returns immediately unless maxOutstandingConfirms is reached, in which case it waits for a confirm callback.

```cpp
// Holds required threads, and other resources.
// The context must live longer than objects created from it.
rmqa::RabbitContext rabbit;

// Create topology to be declared on every reconnection
rmqa::Topology topology;
rmqt::QueueHandle q1    = topology.addQueue("queue-name");
rmqt::ExchangeHandle e1 = topology.addExchange("exch-name");

// Bind e1 and q1 using binding key 'key'
topology.bind(e1, q1, "key");

// To create an auto-generated queue
rmqt::QueueHandle q2 = topology.addQueue();
topology.bind(e1, q2, "key2");

// Each `Producer`/`Consumer` has a reference to a `Topology` object
// which can be updated with updateTopology.
bsl::shared_ptr<rmqa::VHost> vhost = rabbit.createVHostConnection(
    "my-connection",
    bsl::make_shared<rmqt::SimpleEndpoint>(
        "localhost", "rmqcpp", 5762),
    bsl::make_shared<rmqt::PlainCredentials>(
        "guest", "guest")); // returns immediately

// Get a producer
// How many messages can be awaiting confirmation before `send` blocks
const uint16_t maxOutstandingConfirms = 10;

rmqt::Result<rmqa::Producer> producerResult = vhost->createProducer(topology, e1, maxOutstandingConfirms);

if (!producerResult) {
    // handle errors.
    bsl::cout << "Error creating connection: " << producerResult.error();
    return -1;
}

bsl::shared_ptr<rmqa::Producer> producer = producerResult.value();

void receiveConfirmation(const rmqt::Message& message,
                         const bsl::string& routingKey,
                         const rmqt::ConfirmResponse& response)
{
    if (response.status() == rmqt::ConfirmResponse::ACK) {
        // Message is now guaranteed to be safe with the broker
    }
    else {
        // REJECT / RETURN indicate problem with the send request (bad routing
        // key?)
    }
}

bsl::string json    = "[5, 3, 1]";
rmqt::Message message(
    bsl::make_shared<bsl::vector<uint8_t> >(json.cbegin(), json.cend()));

// `send` returns immediately unless there are `maxOutstandingConfirms`
// oustanding messages already. In which case it waits until at least one
// confirm comes back.
// User must wait until the confirm callback is executed before considering
// the send to be committed.
const rmqp::Producer::SendStatus sendResult =
    producer->send(message, "key", &receiveConfirmation);

if (sendResult != rmqp::Producer::SENDING) {
    // Unable to enqueue this send
    return -1;
}

// Consumer callback
class MessageConsumer {
  private:
    bool processMessage(const rmqt::Message& message)
    {
        // process Message here
        return true;
    }

  public:
    void operator()(rmqp::MessageGuard& messageGuard)
    {
        if (processMessage(messageGuard.message())) {
            messageGuard.ack();
        }
        else {
            messageGuard.nack();
            // Would automatically nack if it goes out of scope
        }
    }
};

rmqt::Result<rmqa::Consumer> consumerResult =
    vhost->createConsumer(
        topology,            // topology
        q1,                  // queue
        MessageConsumer(),   // Consumer callback invoked on each message
        "my consumer label", // Consumer Label (shows in Management UI)
        500                  // prefetch count
    );

if (!consumerResult) {
    // An argument passed to the consumer was bad, retrying will have no effect
    return -1;
}

bsl::shared_ptr<rmqa::Consumer> consumer = consumerResult.value();

// Shutdown

// Blocks until `timeout` expires or all confirmations have been received
// Note this could block forever if a separate thread continues publishing
if (!producer->waitForConfirms(/* timeout */)) {
    // Timeout expired
}

consumer->cancelAndDrain();

```

--------------------------------

### Setup Python Virtual Environment for Tests

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Configures a Python virtual environment if not already defined. Installs dependencies from requirements.txt and optionally telnetlib3 on non-SunOS systems.

```cmake
if (NOT DEFINED ENV{PYTHON_BINARY})
    find_package (Python3 COMPONENTS Interpreter REQUIRED)
    execute_process (COMMAND "${Python3_Executable}" -m venv ${CMAKE_CURRENT_BINARY_DIR}/.venv)
    set (ENV{VIRTUAL_ENV} ${CMAKE_CURRENT_BINARY_DIR}/.venv)
    execute_process(COMMAND ${CMAKE_CURRENT_BINARY_DIR}/.venv/bin/pip install  -r ${CMAKE_CURRENT_LIST_DIR}/requirements.txt)
    if(NOT CMAKE_SYSTEM_NAME STREQUAL "SunOS")
        execute_process(COMMAND ${CMAKE_CURRENT_BINARY_DIR}/.venv/bin/pip install telnetlib3)
    endif()
    set(PYTHON_BINARY ${CMAKE_CURRENT_BINARY_DIR}/.venv/bin/python)
else()
    set(PYTHON_BINARY $ENV{PYTHON_BINARY})
endif()
```

--------------------------------

### Install Producer Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/producer/CMakeLists.txt

Installs the RMQ producer executable to the bin directory as part of the rmqtools component.

```cmake
install(
    TARGETS librmq_producer
    DESTINATION bin
    COMPONENT rmqtools)

```

--------------------------------

### Install Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/consumemessages/CMakeLists.txt

Installs the 'librmq_consumemessages' executable to the 'bin' directory under the 'rmqtools' component. This ensures the executable is placed in the correct location after building.

```cmake
install(
    TARGETS librmq_consumemessages
    DESTINATION bin
    COMPONENT rmqtools)
)
```

--------------------------------

### Install Queue Delete Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/queuedelete/CMakeLists.txt

Installs the queue delete executable to the bin directory as part of the rmqtools component.

```cmake
install(
    TARGETS librmq_queuedelete
    DESTINATION bin
    COMPONENT rmqtools)

```

--------------------------------

### RabbitMQ Consumer Callback Implementation

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Example implementation of a consumer callback using a class with an `operator()`. Messages must be acknowledged only after complete processing to avoid data loss.

```cpp
// Consumer callback
class MessageConsumer {
  public:
    void operator()(rmqp::MessageGuard& guard)
    {
        BALL_LOG_INFO << "Received message: " << guard.message() << " Content: "
                      << bsl::string((const char*)guard.message().payload(),
                                     guard.message().payloadSize());
        guard.ack();
    }
};


```

--------------------------------

### Initialize RabbitContext with Options

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Demonstrates minimal and custom initialization of RabbitContext. Custom options include setting error callbacks, message processing timeouts, connection error thresholds, shuffling endpoints, and annotating client properties.

```cpp
#include <rmqa_rabbitcontext.h>
#include <rmqa_rabbitcontextoptions.h>
#include <rmqt_result.h>

using namespace BloombergLP;

// --- Minimal context (default thread pool, no metrics) ---
rmqa::RabbitContext rabbit;

// --- Context with custom options ---
rmqa::RabbitContextOptions opts;

// Custom thread pool (must outlive RabbitContext)
// bdlmt::ThreadPool myPool(...);
// opts.setThreadpool(&myPool);

// Error callback: invoked when the broker closes a channel/connection
opts.setErrorCallback(
    [](const bsl::string& errorText, int errorCode) {
        std::cerr << "RabbitMQ error [" << errorCode << "]: "
                  << errorText << "\n";
    });

// Warn if a consumer callback takes longer than 30 s
opts.setMessageProcessingTimeout(bsls::TimeInterval(30));

// Fire the error callback if no connection is established within 60 s
opts.setConnectionErrorThreshold(bsls::TimeInterval(60));

// Shuffle DNS-resolved endpoints to spread load
opts.setShuffleConnectionEndpoints(true);

// Annotate the connection visible in the RabbitMQ Management UI
opts.setClientProperty("product",
                       rmqt::FieldValue(bsl::string("my-service")));
opts.setClientProperty("version",
                       rmqt::FieldValue(bsl::string("1.2.3")));

rmqa::RabbitContext rabbitWithOpts(opts);
```

--------------------------------

### Sample CMakeLists.txt for Git Submodule Integration

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

This CMakeLists.txt demonstrates how to find a required package (bal), add the rmqcpp submodule, and link it to your executable. Ensure 'bal' is also available or managed appropriately.

```cmake
cmake_minimum_required(VERSION 3.25)
project(myapp)

find_package(bal REQUIRED)
add_subdirectory(rmqcpp)

add_executable(myapplication main.cpp)

target_link_libraries(myapplication rmq bal)

```

--------------------------------

### Set up Project with Git Submodule

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

Integrate rmqcpp into your project by adding it as a git submodule. This approach allows you to manage rmqcpp as a separate component within your larger project repository.

```bash
$ tree
.
`-- myapplication
|   |-- CMakeLists.txt
|   |-- main.cpp
|   `-- rmqcpp
|   |   |-- CMakeLists.txt
|   |   |-- CMakePresets.json
|   |   |-- CONTRIBUTING.md
|   |   |-- LICENSE
|   |   |-- Makefile
|   |   |-- README.md
|   |   |-- ...
|   |-- vcpkg.json

```

--------------------------------

### Build RMQ Hello World Producer with CMake

Source: https://github.com/bloomberg/rmqcpp/blob/main/examples/helloworld/CMakeLists.txt

Configures the build system to create an executable for the RMQ hello world producer. It links against the RMQ and BSL libraries.

```cmake
find_package(bsl REQUIRED)

add_executable(rmqhelloworld_producer
    producer.m.cpp
)

target_link_libraries(rmqhelloworld_producer PUBLIC
    rmq
    bsl
)
```

--------------------------------

### rmqt::Message

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Represents an AMQP message, encapsulating a byte-vector payload along with AMQP properties such as delivery mode, message ID, and headers. All messages default to persistent delivery mode, and a unique GUID is automatically assigned.

```APIDOC
## Message — AMQP message data type

`rmqt::Message` wraps a byte-vector payload together with AMQP properties (delivery mode, message ID, headers, priority). All messages default to **persistent** delivery mode. A unique GUID is assigned automatically on construction.

```cpp
#include <rmqt_message.h>
#include <rmqt_fieldvalue.h>
#include <rmqt_properties.h>

using namespace BloombergLP;

// --- Basic message ---
std::string json = R"({\"event\":\"order.created\",\"id\":\"xyz\"})";
rmqt::Message basicMsg(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()));

// --- With a custom message ID ---
rmqt::Message msgWithId(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    "my-idempotency-key-42");

// --- With headers (for headers exchange routing) ---
rmqt::FieldTable headers{
    {"format", rmqt::FieldValue(bsl::string("json"))},
    {"type",   rmqt::FieldValue(bsl::string("event"))}};
rmqt::Message msgWithHeaders(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    /*messageId=*/ "",
    bsl::make_shared<rmqt::FieldTable>(headers));

// --- With full Properties ---
rmqt::Properties props;
props.deliveryMode = rmqt::DeliveryMode::PERSISTENT;
props.priority     = 5;
rmqt::Message msgWithProps(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    props);

// Access payload on the consume side
void processMessage(const rmqt::Message& msg) {
    bsl::string body(
        reinterpret_cast<const char*>(msg.payload()),
        msg.payloadSize());
    std::cout << "ID=" << msg.messageId()
              << " guid=" << msg.guid()
              << " persistent=" << msg.isPersistent()
              << " payload=" << body << "\n";
}
```
```

--------------------------------

### Build and Test Commands with vcpkg

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

This section provides commands for configuring, building, and testing a project using CMake with the vcpkg toolchain. It also includes alternative Makefile targets for common build tasks and Docker-based builds.

```bash
# Configure with vcpkg toolchain
cmake -DCMAKE_TOOLCHAIN_FILE="${VCPKG_ROOT}/scripts/buildsystems/vcpkg.cmake" \
      -DCMAKE_CXX_STANDARD=17 \
      -B build .

# Build
cmake --build build

# Or use the provided Makefile targets:
make init    # initialise build environment
make build   # incremental build
make unit    # run unit tests

# Docker alternative (no local vcpkg needed):
make docker-setup
make docker-build
make docker-unit
make docker-shell
```

--------------------------------

### Create Async Consumer with Configuration - C++

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Creates an asynchronous consumer that invokes an `onMessage` callback for each delivered message. Messages must be acked or nacked before the `rmqp::MessageGuard` goes out of scope. Uses `rmqt::ConsumerConfig` for settings like prefetch count and priority.

```cpp
#include <rmqa_vhost.h>
#include <rmqa_consumer.h>
#include <rmqp_consumer.h>
#include <rmqt_consumerconfig.h>

using namespace BloombergLP;

// --- Using ConsumerConfig (preferred) ---
rmqt::ConsumerConfig config;
config.setConsumerTag("order-processor-1")
      .setPrefetchCount(50)
      .setConsumerPriority(5);

rmqt::Result<rmqa::Consumer> consumerResult =
    vhost->createConsumer(
        topology,
        orderQueue,
        [](rmqp::MessageGuard& guard) {
            const rmqt::Message& msg = guard.message();

            // Access payload
            bsl::string body(
                reinterpret_cast<const char*>(msg.payload()),
                msg.payloadSize());

            std::cout << "Received [" << msg.messageId() << "]: "
                      << body << "\n";

            // Ack on success, nack to requeue on failure
            if (body.find("orderId") != bsl::string::npos) {
                guard.ack();
            } else {
                guard.nack(); // auto-nacks if guard leaves scope
            }
        },
        config);

if (!consumerResult) {
    std::cerr << "Consumer creation failed: "
              << consumerResult.error() << "\n";
    return 1;
}
bsl::shared_ptr<rmqa::Consumer> consumer = consumerResult.value();

// --- Graceful shutdown ---
// cancel() tells broker to stop delivering; outstanding messages still arrive
consumer->cancel();

// cancelAndDrain() cancels AND waits for all in-flight callbacks to complete
rmqt::Result<> drained = consumer->cancelAndDrain(bsls::TimeInterval(10));
if (!drained) {
    std::cerr << "Drain timed out\n";
}

```

--------------------------------

### Create a RabbitMQ Consumer

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Instantiate a `rmqa::Consumer` object. Requires topology, queue, a consumer callback, a label, and a prefetch count.

```cpp

rmqt::Result<rmqa::Consumer> consumerResult = 
    vhost->createConsumer(
        topology,            // topology
        q1,                  // queue
        MessageConsumer(),   // consumer callback
        "my consumer label", // Consumer Label (shows in Management UI)
        5                    // prefetch count
    );


```

--------------------------------

### Configure Application Build with CMake

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

Use this CMake command to configure your application's build, specifying the toolchain file, library directory, and C++ standard. Ensure the VCPKG_ROOT environment variable is set correctly.

```bash
$ cmake -DCMAKE_TOOLCHAIN_FILE=${VCPKG_ROOT}/scripts/buildsystems/vcpkg.cmake -DCMAKE_INSTALL_LIBDIR=lib64 -DCMAKE_CXX_STANDARD=17 ..

```

--------------------------------

### CMakeLists.txt for rmqcpp Project

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

This CMakeLists.txt file configures a C++ project to use the rmqcpp library. It sets the C++ standard to 17 and links the rmqcpp::rmq target to the main executable.

```cmake
cmake_minimum_required(VERSION 3.25)
project(my-service CXX)

set(CMAKE_CXX_STANDARD 17)

find_package(rmqcpp REQUIRED)

add_executable(my-service main.cpp)
target_link_libraries(my-service PUBLIC rmqcpp::rmq)
```

--------------------------------

### Send a Message with Publisher Confirmation

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Construct an `rmqt::Message` and send it using the producer. The `send` method returns immediately unless the unconfirmed message limit is reached. User must wait for the confirm callback.

```cpp
// Work in progress -- the interface for constructing message payloads is still subject to change
bsl::string messageText    = "Hello RabbitMQ!";
bsl::vector<uint8_t> > payload(messageText.cbegin(), messageText.cend());
rmqt::Message message(bsl::make_shared<bsl::vector<uint8_t> >(payload));

// Method `send` returns immediately unless there are `maxOutstandingConfirms` unacknowledged
// messages already, in which case it waits until at least one confirm comes back.
// User must wait until the confirm callback is executed before considering
// the send to be committed.
const rmqp::Producer::SendStatus sendResult =
    producer->send(message, "key", &receiveConfirmation);

if (sendResult != rmqp::Producer::SENDING) {
    // handle errors
    return -1;
}

// Blocks until `timeout` expires or all confirmations have been received
// Note this could block forever if a separate thread continues publishing
if (!producer->waitForConfirms(/* timeout */)) {
    // Timeout expired
}
```

--------------------------------

### Link rmqcpp Library in CMakeLists.txt

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

After finding the rmqcpp package in your CMakeLists.txt, link it to your executable. This makes the rmqcpp library and its functionalities available to your application.

```cmake
find_package(rmqcpp REQUIRED)

add_executable(myapplication main.cpp)

target_link_libraries(myapplication rmqcpp::rmq)
```

--------------------------------

### Build RMQ Topology Headers Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/examples/topology/CMakeLists.txt

Defines an executable target 'rmqtopology_headers' and links it with the 'rmq' and 'bsl' libraries. Ensure 'bsl' is found using find_package.

```cmake
find_package(bsl REQUIRED)

add_executable(rmqtopology_headers
    headers.m.cpp
)

target_link_libraries(rmqtopology_headers PUBLIC
    rmq
    bsl
)
```

--------------------------------

### Integrate rmqcpp with CMake and vcpkg

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Link against the 'rmqcpp::rmq' target and set the C++ standard to 17 or later when building with CMake and vcpkg. Add 'rmqcpp' to your project's dependencies in vcpkg.json.

```json
{
    "$schema": "https://raw.githubusercontent.com/microsoft/vcpkg-tool/main/docs/vcpkg.schema.json",
    "name": "my-service",
    "version": "1.0.0",
    "dependencies": ["rmqcpp"]
}
```

```cmake

```

--------------------------------

### Add rmqcpp Dependency with vcpkg

Source: https://github.com/bloomberg/rmqcpp/blob/main/README.md

Specify rmqcpp as a dependency in your vcpkg.json file. This ensures that vcpkg will fetch and build rmqcpp along with its own dependencies when you build your project.

```json
{
    "$schema": "https://raw.githubusercontent.com/microsoft/vcpkg-tool/main/docs/vcpkg.schema.json",
    "name": "testproj",
    "version": "1.0.0",
    "dependencies": [
      "rmqcpp"
    ]
}
```

--------------------------------

### Publish Message with Confirmation - C++

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Enqueues a message for delivery and provides a callback for broker confirmations (ACK, REJECT, RETURN). Blocks if the in-flight window is full. Use `waitForConfirms` for graceful shutdown.

```cpp
#include <rmqa_producer.h>
#include <rmqt_message.h>
#include <rmqt_confirmresponse.h>
#include <rmqp_producer.h>

using namespace BloombergLP;

// Build message (persistent delivery mode by default)
std::string payload = R"({\"orderId\":\"abc-123\",\"amount\":42.00})";
rmqt::Message message(
    bsl::make_shared<bsl::vector<uint8_t>>(payload.begin(), payload.end()));

// Confirmation callback (invoked on the rmqcpp thread pool)
auto onConfirm = [](const rmqt::Message&        msg,
                    const bsl::string&           routingKey,
                    const rmqt::ConfirmResponse& response) {
    if (response.status() == rmqt::ConfirmResponse::ACK) {
        // Safe to commit DB transaction, reply to client, etc.
        std::cout << "Message " << msg.guid() << " confirmed\n";
    } else if (response.status() == rmqt::ConfirmResponse::RETURN) {
        // Mandatory flag triggered — no queue matched the routing key
        std::cerr << "Message returned (no binding for '"
                  << routingKey << "'): " << response << "\n";
    } else {
        // REJECT — broker could not process the message
        std::cerr << "Message rejected: " << response << "\n";
    }
};

rmqp::Producer::SendStatus status =
    producer->send(message, "new-order", onConfirm);

if (status == rmqp::Producer::DUPLICATE) {
    std::cerr << "Duplicate GUID — create a new rmqt::Message\n";
    return 1;
} else if (status != rmqp::Producer::SENDING) {
    std::cerr << "Unexpected send failure\n";
    return 1;
}

// Graceful shutdown: wait for all pending confirms (indefinitely)
producer->waitForConfirms();

// Or with a timeout:
// rmqt::Result<> drained = producer->waitForConfirms(bsls::TimeInterval(5));
// if (!drained) { /* timeout */ }

```

--------------------------------

### Create RabbitMQ vHost Connection

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Establishes a connection to a RabbitMQ virtual host. The actual handshake occurs in the background. Accepts either parsed URI info or explicit endpoint and credentials. The VHost object must outlive any derived Producer or Consumer objects.

```cpp
#include <rmqa_rabbitcontext.h>
#include <rmqa_connectionstring.h>
#include <rmqt_simpleendpoint.h>
#include <rmqt_secureendpoint.h>
#include <rmqt_plaincredentials.h>

using namespace BloombergLP;

rmqa::RabbitContext rabbit;

// --- Option A: from a parsed URI ---
bsl::optional<rmqt::VHostInfo> vhostInfo =
    rmqa::ConnectionString::parse("amqp://guest:guest@localhost:5672/");
bsl::shared_ptr<rmqa::VHost> vhostA =
    rabbit.createVHostConnection("my-service-connection", vhostInfo.value());

// --- Option B: explicit endpoint + credentials (plain AMQP) ---
bsl::shared_ptr<rmqa::VHost> vhostB = rabbit.createVHostConnection(
    "my-service-connection",
    bsl::make_shared<rmqt::SimpleEndpoint>("localhost", "/", 5672),
    bsl::make_shared<rmqt::PlainCredentials>("guest", "guest"));

// --- Option C: TLS (amqps) ---
// bsl::shared_ptr<rmqt::SecurityParameters> tlsParams = ...;
// bsl::shared_ptr<rmqa::VHost> vhostC = rabbit.createVHostConnection(
//     "tls-connection",
//     bsl::make_shared<rmqt::SecureEndpoint>("broker.example.com", "/", 5671, tlsParams),
//     bsl::make_shared<rmqt::PlainCredentials>("user", "secret"));

```

--------------------------------

### Link RMQT Tests Libraries

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqt/CMakeLists.txt

Links the RMQT test executable with its dependencies, including the RMQT library, its link libraries, and Google Test/Mock.

```cmake
target_link_libraries(rmqt_tests PUBLIC 
    rmqt
    $<TARGET_PROPERTY:rmqt,LINK_LIBRARIES>
    GTest::gtest
    GTest::gmock
)
```

--------------------------------

### Configure rmqperftest Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/examples/rmqperftest/CMakeLists.txt

Defines the rmqperftest executable, links necessary libraries, and sets include directories. This is used to build the performance testing tool for RMQ.

```cmake
find_package(bsl REQUIRED)
find_package(bal REQUIRED)

add_executable(rmqperftest
    rmqperftest.m.cpp
    rmqperftest_args.cpp
    rmqperftest_consumerargs.cpp
    rmqperftest_runner.cpp
)

target_compile_definitions(rmqperftest PRIVATE USES_LIBRMQ_EXPERIMENTAL_FEATURES)

target_link_libraries(rmqperftest PUBLIC
    bsl
    bal
    rmq
    rmqintegration
)
target_include_directories(rmqperftest PRIVATE .)

install(
    TARGETS rmqperftest
    DESTINATION bin
    COMPONENT rmqtools
)
```

--------------------------------

### Create a RabbitMQ Producer

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Instantiate a `rmqa::Producer` object. Requires a topology, exchange, and maximum number of unconfirmed messages before `send` blocks.

```cpp
// How many messages can be awaiting confirmation before `send` blocks
const uint16_t maxOutstandingConfirms = 10;

// Create a producer object
rmqt::Result<rmqa::Producer> producerResult =
    vhost->createProducer(topology, e1, maxOutstandingConfirms);

if (!producerResult) {
    // Handle error
    bsl::cout << "Error creating connection: " << producerResult.error();
    return -1;
}

bsl::shared_ptr<rmqa::Producer> producer = producerResult.value();
```

--------------------------------

### Configure Pytest Command

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Sets up the pytest command using the determined Python binary and defines common arguments for test execution.

```cmake
set(PYTHON_BINARY ${CMAKE_CURRENT_BINARY_DIR}/.venv/bin/python)
set( PYTEST ${PYTHON_BINARY} -mpytest)
set( PYTEST_ARGS "-q" "-rap" "--maxfail=2" "--tb=short" "--durations=1" "--log-level=error" "-s" )
```

--------------------------------

### Create VHost Connection

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Create a VHost object from a RabbitContext to manage producers and consumers on a specific RabbitMQ vhost. The actual connection to the broker is established lazily when creating producers or consumers.

```cpp
rmqt::VHostInfo vhostInfo(
    bsl::make_shared<rmqt::SimpleEndpoint>("rabbit", "vhost-name"),
    bsl::make_shared<rmqt::PlainCredentials>(guest, guest));
    
bsl::shared_ptr<rmqa::VHost> vhost = context.createVHostConnection(
    "my-producer-connection", vhostInfo); // returns immediately
```

--------------------------------

### Create RabbitContext

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Instantiate a RabbitContext object, which manages threadpools, metric publishers, and error callbacks. This object must outlive all other library objects. Custom options can be provided to override default components.

```cpp
rmqa::RabbitContext context;

// Only if necessary, create an options object to provide custom components for the library
rmqa::RabbitContextOptions options;
options.setThreadpool(/* threadpool */);
options.setMetricPublisher(/* metric publisher */);
options.setErrorCallback(/* error callback */);

rmqa::RabbitContext contextWithOptions(options);
```

--------------------------------

### Link Libraries for RMQ Test Mocks Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqtestmocks/CMakeLists.txt

Links the necessary libraries to the rmqtestmocks_tests executable, including the rmqtestmocks library and Google Test/Mock.

```cmake
target_link_libraries(rmqtestmocks_tests PUBLIC 
    rmqtestmocks
    GTest::gtest
    GTest::gmock
)
```

--------------------------------

### Define Executable and Link Libraries

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/consumemessages/CMakeLists.txt

Defines the executable target 'librmq_consumemessages' and links it with necessary libraries. Ensure all listed libraries are available in your build environment.

```cmake
add_executable(librmq_consumemessages 
    librmq_consumemessages.m.cpp
)

target_link_libraries(librmq_consumemessages PUBLIC 
    bsl
    bal
    rmq
    rmqtestutil
    rmqintegration
)
```

--------------------------------

### Link Dependencies for Integration Library

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Links necessary libraries (bsl, bdl, bal, rmq) to the integration test library.

```cmake
target_link_libraries(rmqintegration PUBLIC
    bsl
    bdl
    bal
    rmq
)
```

--------------------------------

### Set Include Directories for Integration Library

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Specifies the include directories for the integration test library.

```cmake
target_include_directories(rmqintegration PUBLIC .)
```

--------------------------------

### Parse AMQP Connection String

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Parses an AMQP URI into a rmqt::VHostInfo object. Returns an empty optional on failure. The parsed info can be used with createVHostConnection.

```cpp
#include <rmqa_connectionstring.h>
#include <rmqt_vhostinfo.h>

using namespace BloombergLP;

// amqp://user:pass@host:5672/vhost
bsl::optional<rmqt::VHostInfo> info =
    rmqa::ConnectionString::parse("amqp://guest:guest@localhost:5672/");

if (!info) {
    std::cerr << "Bad URI\n";
    return 1;
}

// Use info.value() with createVHostConnection (see below)
```

--------------------------------

### Define RabbitMQ Topology

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Manages the declaration of exchanges, queues, and bindings that will be established before producers or consumers become active. Handles are lightweight references. Supports direct, topic, and headers exchanges, as well as passive declarations and queue arguments like TTL.

```cpp
#include <rmqa_topology.h>
#include <rmqt_exchange.h>
#include <rmqt_queue.h>
#include <rmqt_exchangetype.h>
#include <rmqt_fieldvalue.h>

using namespace BloombergLP;

rmqa::Topology topology;

// --- Direct exchange + named durable queue ---
rmqt::ExchangeHandle directExch = topology.addExchange("orders");
rmqt::QueueHandle    orderQueue = topology.addQueue("orders.processing");
topology.bind(directExch, orderQueue, "new-order");

// --- Topic exchange ---
rmqt::ExchangeHandle topicExch =
    topology.addExchange("events", rmqt::ExchangeType::TOPIC);
rmqt::QueueHandle alertQueue = topology.addQueue("alerts");
topology.bind(topicExch, alertQueue, "payment.#");

// --- Headers exchange ---
rmqt::ExchangeHandle headersExch =
    topology.addExchange("docs", rmqt::ExchangeType::HEADERS);
rmqt::QueueHandle pdfQueue = topology.addQueue("docs.pdf");
rmqt::FieldTable  bindArgs{
    {"x-match", rmqt::FieldValue(bsl::string("all"))},
    {"format",  rmqt::FieldValue(bsl::string("pdf"))}};
topology.bind(headersExch, pdfQueue, "", bindArgs);

// --- Auto-named (server-generated) queue ---
rmqt::QueueHandle tempQueue = topology.addQueue(); // name assigned by broker
topology.bind(directExch, tempQueue, "broadcast");

// --- Passive declarations (assert pre-existing resources) ---
rmqt::ExchangeHandle existingExch = topology.addPassiveExchange("legacy-exch");
rmqt::QueueHandle    existingQ    = topology.addPassiveQueue("legacy-queue");

// --- Queue with TTL via arguments ---
rmqt::FieldTable queueArgs{
    {"x-message-ttl", rmqt::FieldValue(int64_t(60000))}};
rmqt::QueueHandle ttlQueue =
    topology.addQueue("ttl-queue",
                      rmqt::AutoDelete::OFF,
                      rmqt::Durable::ON,
                      queueArgs);

```

--------------------------------

### Define RMQT Tests Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqt/CMakeLists.txt

Defines the executable for the RMQT tests and lists its source files.

```cmake
add_executable(rmqt_tests
    rmqt.m.cpp
    rmqt_consumerconfig.t.cpp
    rmqt_envelope.t.cpp
    rmqt_exchange.t.cpp
    rmqt_fieldvalue.t.cpp
    rmqt_future.t.cpp
    rmqt_message.t.cpp
    rmqt_plaincredentials.t.cpp
    rmqt_secureendpoint.t.cpp
    rmqt_simpleendpoint.t.cpp
)
```

--------------------------------

### Link Libraries for Multithread Consumer

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/multithread_consumer/CMakeLists.txt

Links necessary libraries to the multithread consumer executable. Ensures all dependencies are available at build time.

```cmake
target_link_libraries(librmq_multithread_consumer PUBLIC 
    bsl
    bal
    rmq
    rmqintegration
)
```

--------------------------------

### Set Include Directories for RMQ Test Mocks

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqtestmocks/CMakeLists.txt

Specifies private include directories for the rmqtestmocks_tests executable.

```cmake
target_include_directories(rmqtestmocks_tests PRIVATE .)
```

--------------------------------

### Link Libraries for Queue Delete

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/queuedelete/CMakeLists.txt

Links the necessary libraries for the queue delete executable, including bsl, bal, rmq, and rmqintegration.

```cmake
target_link_libraries(librmq_queuedelete PUBLIC 
    bsl
    bal
    rmq
    rmqintegration
)
```

--------------------------------

### Set RMQT Tests Include Directories

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqt/CMakeLists.txt

Specifies private include directories for the RMQT tests executable.

```cmake
target_include_directories(rmqt_tests PRIVATE .)
```

--------------------------------

### Shutting Down RMQ C++ Client

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Call `waitForConfirms()` to ensure all sent messages are confirmed before proceeding. Then, `cancelAndDrain()` to stop consuming messages and `close()` the connection to the broker. After `vhost->close()`, the library will not attempt to reconnect.

```cpp
// Close the producer. The method `waitForConfirms()` will block until all sent messages are confirmed by the broker.
producer->waitForConfirms();

// Cancel the consumer.
consumer->cancelAndDrain();

// Close the connections to the broker. After calling `rmqa::VHost::close()`, the library will not reconnect to the broker anymore.
vhost->close();

```

--------------------------------

### rmqa::ConnectionString::parse

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Parses a standard AMQP or AMQPS URI into a rmqt::VHostInfo value. This value can then be used with createVHostConnection. Returns an empty bsl::optional if the URI fails to parse.

```APIDOC
## rmqa::ConnectionString::parse — Parse an AMQP URI

Pareses a standard `amqp://` or `amqps://` URI into a `rmqt::VHostInfo` value that can be passed directly to `createVHostConnection`. Returns an empty `bsl::optional` on parse failure.

```cpp
#include <rmqa_connectionstring.h>
#include <rmqt_vhostinfo.h>

using namespace BloombergLP;

// amqp://user:pass@host:5672/vhost
bsl::optional<rmqt::VHostInfo> info =
    rmqa::ConnectionString::parse("amqp://guest:guest@localhost:5672/");

if (!info) {
    std::cerr << "Bad URI\n";
    return 1;
}

// Use info.value() with createVHostConnection (see below)
```
```

--------------------------------

### RabbitMQ Publisher Confirmation Callback

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Define a callback function to handle publisher confirmations (ACK/REJECT/RETURN) received from the broker after sending a message. Essential for ensuring message safety.

```cpp
void receiveConfirmation(const rmqt::Message& message,
                         const bsl::string& routingKey,
                         const rmqt::ConfirmResponse& response)
{
    if (response.status() == rmqt::ConfirmResponse::ACK) {
        // Message is now guaranteed to be safe with the broker
    }
    else {
        // REJECT / RETURN indicate problem with the send request (bad routing
        // key?)
        // NB: users must handle the send failure to avoid dropping the message!
        BALL_LOG_WARN << "Sending message failed: " << response;
    }
}
```

--------------------------------

### Create and Use rmqt::Message Objects

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Construct rmqt::Message with a byte-vector payload and optional AMQP properties like message ID, headers, and delivery mode. Access payload and properties on the consume side.

```cpp
#include <rmqt_message.h>
#include <rmqt_fieldvalue.h>
#include <rmqt_properties.h>

using namespace BloombergLP;

// --- Basic message ---
std::string json = R"({\"event\":\"order.created\",\"id\":\"xyz\"})";
rmqt::Message basicMsg(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()));

// --- With a custom message ID ---
rmqt::Message msgWithId(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    "my-idempotency-key-42");

// --- With headers (for headers exchange routing) ---
rmqt::FieldTable headers{
    {"format", rmqt::FieldValue(bsl::string("json"))},
    {"type",   rmqt::FieldValue(bsl::string("event"))}};
rmqt::Message msgWithHeaders(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    /*messageId=*/ "",
    bsl::make_shared<rmqt::FieldTable>(headers));

// --- With full Properties ---
rmqt::Properties props;
props.deliveryMode = rmqt::DeliveryMode::PERSISTENT;
props.priority     = 5;
rmqt::Message msgWithProps(
    bsl::make_shared<bsl::vector<uint8_t>>(json.begin(), json.end()),
    props);

// Access payload on the consume side
void processMessage(const rmqt::Message& msg) {
    bsl::string body(
        reinterpret_cast<const char*>(msg.payload()),
        msg.payloadSize());
    std::cout << "ID=" << msg.messageId()
              << " guid=" << msg.guid()
              << " persistent=" << msg.isPersistent()
              << " payload=" << body << "\n";
}
```

--------------------------------

### Define Executable for Exit Both Test

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/exitboth/CMakeLists.txt

Defines the C++ executable for the exit both integration test. This is the primary binary that will be built and linked.

```cmake
add_executable(librmq_exitboth 
    librmq_exitboth.m.cpp
)
```

--------------------------------

### Set Compile Definitions

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/queuedelete/CMakeLists.txt

Sets a compile definition for the queue delete executable, enabling experimental features.

```cmake
target_compile_definitions(librmq_queuedelete PRIVATE USES_LIBRMQ_EXPERIMENTAL_FEATURES)
```

--------------------------------

### Link Libraries for rmqio_tests

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqio/CMakeLists.txt

Specifies the libraries to be linked with the rmqio_tests executable. This includes various rmq libraries, utility libraries, and Google Test components.

```cmake
target_link_libraries(rmqio_tests PUBLIC
    rmqamqpt
    rmqt
    rmqio
    rmqamqp
    $<TARGET_PROPERTY:rmqt,LINK_LIBRARIES>
    $<TARGET_PROPERTY:rmqamqpt,LINK_LIBRARIES>
    $<TARGET_PROPERTY:rmqio,LINK_LIBRARIES>
    $<TARGET_PROPERTY:rmqamqp,LINK_LIBRARIES>
    rmqtestutil
    GTest::gtest
    GTest::gmock
)
```

--------------------------------

### rmqa::VHost::createConsumer — Create an async message consumer

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Creates an asynchronous message consumer for a given topology and queue. The consumer uses a callback to process incoming messages and returns a `rmqt::Result<rmqa::Consumer>`.

```APIDOC
## rmqa::VHost::createConsumer — Create an async message consumer

### Description
Returns `rmqt::Result<rmqa::Consumer>`. The `onMessage` callback is invoked on the `RabbitContext` thread pool for every delivered message. `rmqp::MessageGuard` must be acked or nacked before it goes out of scope (the guard auto-nacks on destruction).

### Method
`rmqt::Result<rmqa::Consumer> createConsumer(const Topology& topology, const Queue& queue, std::function<void(rmqp::MessageGuard&)> onMessage, const rmqt::ConsumerConfig& config)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
- **topology** (Topology) - The exchange topology.
- **queue** (Queue) - The queue to consume from.
- **onMessage** (std::function) - Callback function invoked for each received message.
- **config** (rmqt::ConsumerConfig) - Configuration for the consumer (e.g., prefetch count, priority).

### Request Example
```cpp
#include <rmqa_vhost.h>
#include <rmqa_consumer.h>
#include <rmqp_consumer.h>
#include <rmqt_consumerconfig.h>

using namespace BloombergLP;

// --- Using ConsumerConfig (preferred) ---
rmqt::ConsumerConfig config;
config.setConsumerTag("order-processor-1")
      .setPrefetchCount(50)
      .setConsumerPriority(5);

rmqt::Result<rmqa::Consumer> consumerResult =
    vhost->createConsumer(
        topology,
        orderQueue,
        [](rmqp::MessageGuard& guard) {
            const rmqt::Message& msg = guard.message();

            // Access payload
            bsl::string body(
                reinterpret_cast<const char*>(msg.payload()),
                msg.payloadSize());

            std::cout << "Received [" << msg.messageId() << "]: "
                      << body << "\n";

            // Ack on success, nack to requeue on failure
            if (body.find("orderId") != bsl::string::npos) {
                guard.ack();
            } else {
                guard.nack(); // auto-nacks if guard leaves scope
            }
        },
        config);

if (!consumerResult) {
    std::cerr << "Consumer creation failed: "
              << consumerResult.error() << "\n";
    return 1;
}
bsl::shared_ptr<rmqa::Consumer> consumer = consumerResult.value();

// --- Graceful shutdown ---
// cancel() tells broker to stop delivering; outstanding messages still arrive
consumer->cancel();

// cancelAndDrain() cancels AND waits for all in-flight callbacks to complete
rmqt::Result<> drained = consumer->cancelAndDrain(bsls::TimeInterval(10));
if (!drained) {
    std::cerr << "Drain timed out\n";
}
```

### Response
#### Success Response (200)
Returns a `rmqt::Result<rmqa::Consumer>` containing a shared pointer to the created consumer on success.

#### Response Example
`rmqt::Result<rmqa::Consumer>`
```

--------------------------------

### Add Subdirectories for Test Suites

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Includes subdirectories that contain specific test suites.

```cmake
add_subdirectory(rmqperftest)
add_subdirectory(consumemessages)
add_subdirectory(producer)
add_subdirectory(multithread_producer)
add_subdirectory(multithread_consumer)
add_subdirectory(exitboth)
add_subdirectory(queuedelete)
```

--------------------------------

### Define RMQ Test Mocks Executable

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqtestmocks/CMakeLists.txt

Defines the executable for the RMQ test mocks, specifying its source files.

```cmake
add_executable(rmqtestmocks_tests
    rmqtestmocks.m.cpp
    rmqtestmocks_mocks.t.cpp
)
```

--------------------------------

### Adding Subdirectories to Build

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/CMakeLists.txt

Includes other CMake build configurations from specified subdirectories into the main build process.

```cmake
add_subdirectory(rmqtestutil)
```

```cmake
add_subdirectory(integration)
```

```cmake
add_subdirectory(performance)
```

```cmake
add_subdirectory(rmqamqp)
```

```cmake
add_subdirectory(rmqamqpt)
```

```cmake
add_subdirectory(rmqio)
```

```cmake
add_subdirectory(rmqt)
```

```cmake
add_subdirectory(rmqa)
```

```cmake
add_subdirectory(rmqtestmocks)
```

--------------------------------

### Define AMQP Topology

Source: https://github.com/bloomberg/rmqcpp/blob/main/docs/doxygen.md

Define the AMQP topology by adding exchanges and queues, and then binding them together. This sets up the message routing paths within RabbitMQ.

```cpp
// Create topology
rmqa::Topology topology;
rmqt::QueueHandle q1    = topology.addQueue("queue-name");
rmqt::ExchangeHandle e1 = topology.addExchange("exch-name");

// Bind e1 and q1 using binding key 'key'
topology.bind(e1, q1, "key1");

// To create an auto-generated queue
rmqt::QueueHandle q2 = topology.addQueue();
topology.bind(e1, q2, "key2");

```

--------------------------------

### Define Integration Library

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/integration/CMakeLists.txt

Defines an object library for integration tests, specifying the source files.

```cmake
add_library(rmqintegration OBJECT
    rmqintegration_testparameters.cpp
)
```

--------------------------------

### Compiler Definitions for SunPro

Source: https://github.com/bloomberg/rmqcpp/blob/main/src/tests/rmqio/CMakeLists.txt

Applies specific compile definitions for the SunPro compiler to ensure compatibility with Boost.Asio, particularly regarding standard allocator and socket definitions.

```cmake
if( "${CMAKE_CXX_COMPILER_ID}" STREQUAL "SunPro" )
    # _RWSTD_ALLOCATOR tells the solaris <memory> header to define a std::allocator
    # which conforms better to the C++ standard, which is expected by Boost. Without
    # this, the library does not build due to missing std::allocator<..>::rebind

    # "-D_XOPEN_SOURCE=500" "-D__EXTENSIONS__" tells the solaris sys/socket.h to conform
    # better to what boost asio expects (to have a msg_flags member)
    target_compile_definitions(rmqio_tests PRIVATE _RWSTD_ALLOCATOR "-D_XOPEN_SOURCE=500" "-D__EXTENSIONS__")
endif()
```

--------------------------------

### rmqa::TopologyUpdate

Source: https://context7.com/bloomberg/rmqcpp/llms.txt

Allows live binding changes for Producers and Consumers without requiring a reconnection. It enables adding or removing bindings dynamically, which are then confirmed by the broker and integrated into the topology for future re-declarations.

```APIDOC
## TopologyUpdate — Live binding changes

`TopologyUpdate` lets a running `Producer` or `Consumer` add or remove bindings without reconnecting. Changes are sent to the broker and, if confirmed, are folded into the topology that will be re-declared on future reconnections.

```cpp
#include <rmqa_topologyupdate.h>

using namespace BloombergLP;

rmqa::TopologyUpdate update;
update.bind(topicExch, alertQueue, "payment.failed");

rmqt::Result<> bindResult =
    producer->updateTopology(update, bsls::TimeInterval(5));
if (!bindResult) {
    std::cerr << "Topology update failed: " << bindResult.error() << "\n";
}

// Remove a binding
rmqa::TopologyUpdate removal;
removal.unbind(topicExch, alertQueue, "payment.failed");
consumer->updateTopology(removal, bsls::TimeInterval(5));
```
```