DaemonMessaging.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 "qrmci/CommunicationHandler.h"
#include "qrmci/Error.h"
#include "qrmci/Runners.h"
 
#include <atomic>
#include <chrono>
#include <functional>
#include <spdlog/spdlog.h>
#include <string>
#include <string_view>
#include <utility>
 
namespace mqss::qrmci {

template <class Message, class Handler>
bool receiveNext(CommunicationHandler &comms, const std::string &queueName,
                 std::chrono::milliseconds timeout,
                 const std::atomic<bool> &terminationFlag, Handler &&handle) {
  auto received = comms.receive<Message>(queueName, timeout, terminationFlag);
  if (!received) {
    spdlog::warn("Could not read from queue '{}': {}", queueName,
                 received.error().detail);
    return false;
  }
  if (!received->has_value()) {
    return false;
  }
  std::invoke(std::forward<Handler>(handle), std::move(**received));
  return true;
}

template <class Message, class Handler>
bool drainQueue(CommunicationHandler &comms, const std::string &queueName,
                const std::atomic<bool> &terminationFlag, Handler &&handle) {
  bool handledAny = false;
  // A single non-blocking poll per turn: the queue is drained by looping,
  // not by waiting, so an empty queue costs one round trip rather than a
  // timeout.
  while (receiveNext<Message>(comms, queueName, std::chrono::milliseconds(0),
                              terminationFlag, handle)) {
    handledAny = true;
  }
  return handledAny;
}

template <class Message>
void send(CommunicationHandler &comms, const Message &message,
          const std::string &queueName) {
  if (auto sent = comms.send(message, queueName); !sent) {
    spdlog::warn("{}", sent.error().detail);
  }
}

inline void sendFailure(CommunicationHandler &comms,
                        const mqss::QuantumTask &task, const Error &failure,
                        std::string_view stage) {
  send(comms, cancelQuantumTask(task, failure.detail),
       task.result_destination());
  spdlog::warn("Task {} {} [{}]: {}", task.task_id(), stage,
               toString(failure.kind), failure.detail);
}

inline void sendResult(CommunicationHandler &comms,
                       const mqss::QuantumTask &task,
                       const mqss::QuantumResult &result) {
  send(comms, result, task.result_destination());
  spdlog::info("Job Executed successfully. Results for task {} sent to '{}'.",
               task.task_id(), task.result_destination());
}
 
} // namespace mqss::qrmci