CommunicationHandler.h

Go to the documentation of this file.

/*
 * Copyright (c) 2026 MQSS Maintainers
 * All rights reserved.
 *
 * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 */

 
#pragma once
 
#include "mqss/Messenger.hpp"
#include "mqss/Protocol.hpp"
#include "mqss/Transport.hpp"
#include "qrmci/Config.h"
#include "qrmci/Error.h"
 
#include <atomic>
#include <chrono>
#include <expected>
#include <optional>
#include <spdlog/spdlog.h>
#include <string>
#include <utility>
 
namespace mqss::qrmci {

inline constexpr std::chrono::milliseconds DefaultReceiveTimeout{500};

inline constexpr std::chrono::milliseconds WaitForever{-1};

[[nodiscard]] mqss::TransportOptions<mqss::RabbitMqSimple>
makeTransportOptions(const RabbitMqConnectionConfig &config);

inline std::string messageLabel(const mqss::QuantumTask &task) {
  return "quantum task " + std::to_string(task.task_id());
}

inline std::string messageLabel(const mqss::QuantumResult &result) {
  return "quantum result for task " + std::to_string(result.task_id());
}

inline std::string messageLabel(const mqss::Backend &backend) {
  return "backend status for backend " + backend.name();
}

template <class Message>
std::string messageLabel(const Message & /*unused*/) = delete;

class CommunicationHandler {
public:
  explicit CommunicationHandler(const RabbitMqConnectionConfig &config);

  CommunicationHandler(const CommunicationHandler &) = delete;
  CommunicationHandler &operator=(const CommunicationHandler &) = delete;
  CommunicationHandler(CommunicationHandler &&) = delete;
  CommunicationHandler &operator=(CommunicationHandler &&) = delete;

  template <class Message>
  [[nodiscard]] std::expected<void, Error> send(const Message &message,
                                                const std::string &queueName) {
    auto sendStatus = messenger.send<Message>({queueName}, message);
 
    if (!sendStatus.ok()) {
      return std::unexpected(Error{Error::Kind::MessagingFailed,
                                   "Failed to send " + messageLabel(message) +
                                       ": " + sendStatus.reason()});
    }
    return {};
  }

  template <class Message>
  [[nodiscard]] std::expected<std::optional<Message>, Error>
  receive(const std::string &queueName, std::chrono::milliseconds timeout,
          const std::atomic<bool> &terminationFlag) {
    const bool waitIndefinitely = (timeout == WaitForever);
    const auto pollInterval =
        waitIndefinitely ? DefaultReceiveTimeout : timeout;
 
    // The flag is the loop condition rather than a second conjunct alongside
    // the timeout, so it is honoured at every timeout instead of only at 0.
    while (!terminationFlag.load(std::memory_order_acquire)) {
      // Manual ack keeps the broker-side no_ack flag false, so RabbitMQ's
      // per-consumer prefetch actually throttles delivery instead of pushing
      // every queued message onto this call's short-lived connection at once.
      // The message is acked right after a successful decode.
      auto received = messenger.receiveExtended<Message>(
          {queueName}, mqss::ReceiveArgs{
                           .timeout = pollInterval,
                           .ack_mode = mqss::AckMode::Manual,
                       });
      if (received.has_value()) {
        auto &[message, delivery] = *received;
        if (auto ackStatus = delivery.ack(); !ackStatus.ok()) {
          spdlog::warn("Failed to ack {}: {}", messageLabel(message),
                       ackStatus.reason());
        }
        return std::optional<Message>(std::move(message));
      }
      if (received.error().code() != mqss::StatusCode::Timeout) {
        return std::unexpected(
            Error{Error::Kind::MessagingFailed,
                  "Receive error: " + received.error().reason()});
      }
      if (!waitIndefinitely) {
        break;
      }
    }
    return std::optional<Message>{};
  }

  template <class Message>
  [[nodiscard]] std::expected<std::optional<Message>, Error>
  receive(const std::string &queueName,
          std::chrono::milliseconds timeout = DefaultReceiveTimeout) {
    static const std::atomic<bool> NeverTerminates{false};
    return receive<Message>(queueName, timeout, NeverTerminates);
  }
 
private:
  mqss::Messenger<mqss::RabbitMqSimple, mqss::ProtoJson> messenger;
};
} // namespace mqss::qrmci