ControllDroneServerLinux/SafeMessageQueue.hpp - master - vasandi/DroneControl

ControllDroneServerLinux/SafeMessageQueue.hpp

86 строк2 KB

Андрей Васильченко

управление дроном в Airsim

08 июл 2026, 13:29

08 июл 2026, 13:29d1a17d2

#ifndef SAFE_MSG_QUEUE_HPP
#define SAFE_MSG_QUEUE_HPP

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <queue>
#include <string>
#include <atomic>

namespace drone
{
/// <summary>
/// Поточно-безопасная очередь сообщений
/// </summary>
template <typename MessageType>
class SafeMessageQueue
{
private:
    std::mutex _mtx;
    std::condition_variable _cond_var;
    std::queue<MessageType> _messages;
    std::atomic<bool> _shutdown{false};

public:
    /// <summary>
    /// Добавление сообщение в очередь
    /// </summary>
    template <typename T>
    void push(T&& message)
    {
        std::lock_guard<std::mutex> lock(_mtx);
        _messages.emplace(std::forward<T>(message));
        _cond_var.notify_one();
    }

    /// <summary>
    /// Удаление сообщения из очереди (прерывается при shutdown)
    /// </summary>
    bool pop(MessageType &result)
    {
        std::unique_lock<std::mutex> lock(_mtx);
        _cond_var.wait(lock, [this]() { return !_messages.empty() || _shutdown.load(); });
        if (_shutdown.load() && _messages.empty())
            return false;
        result = std::move(_messages.front());
        _messages.pop();
        return true;
    }

    /// <summary>
    /// Удаление сообщения с таймаутом
    /// </summary>
    bool tryPop(MessageType &msg, std::chrono::milliseconds timeout)
    {
        std::unique_lock<std::mutex> lock(_mtx);
        if (!_cond_var.wait_for(lock, timeout, [this]() { return !_messages.empty() || _shutdown.load(); }))
            return false;
        if (_shutdown.load() && _messages.empty())
            return false;
        msg = std::move(_messages.front());
        _messages.pop();
        return true;
    }

    /// <summary>
    /// Установка флага завершения и пробуждение потоков
    /// </summary>
    void shutdown()
    {
        _shutdown.store(true);
        _cond_var.notify_all();
    }

    /// <summary>
    /// Пробудить ожидающие потоки
    /// </summary>
    void wake()
    {
        _cond_var.notify_all();
    }
};
}

#endif