IpcChannel.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <cstddef>
#include <string>
#include <utility>
 
#include "corium/Events.hpp"
#include "corium/EventSink.hpp"
#include "corium/ipc/SharedMemory.hpp"
#include "corium/ipc/ShmMpscQueue.hpp"
 
namespace corium::ipc {
 
template <
    typename EventVariant = DefaultEvents,
    std::size_t Capacity = 256
>
class IpcChannel {
    static_assert(std::is_trivially_copyable_v<EventVariant>,
        "IpcChannel EventVariant must be trivially copyable for zero-copy shared memory IPC.");
 
public:
    using QueueType = ShmMpscQueue<EventVariant, Capacity>;
 
    IpcChannel() = default;
 
    bool create(const std::string& channelName) noexcept
    {
        if (!_shm.open(channelName, QueueType::requiredMemorySize(), SharedMemory::AccessMode::CreateOrOpen)) {
            return false;
        }
        _queue.bind(_shm.data(), true);
        return _queue.isValid();
    }
 
    bool attach(const std::string& channelName) noexcept
    {
        if (!_shm.open(channelName, QueueType::requiredMemorySize(), SharedMemory::AccessMode::OpenReadWrite)) {
            return false;
        }
        _queue.bind(_shm.data(), false);
        return _queue.isValid();
    }
 
    bool bindRaw(void* buffer, bool isCreator = true) noexcept
    {
        _queue.bind(buffer, isCreator);
        return _queue.isValid();
    }
 
    template <typename EventType>
    bool post(EventType&& event) noexcept
    {
        return _queue.tryPush(EventVariant{std::forward<EventType>(event)});
    }
 
    bool tryPop(EventVariant& outEvent) noexcept
    {
        return _queue.tryPop(outEvent);
    }
 
    template <typename SinkType>
    std::size_t pumpInto(const SinkType& sink, std::size_t maxEvents = 0)
    {
        std::size_t count = 0;
        EventVariant ev;
        while (_queue.tryPop(ev)) {
            sink.post(std::move(ev));
            count++;
            if (maxEvents > 0 && count >= maxEvents) {
                break;
            }
        }
        return count;
    }
 
    void unlink() noexcept
    {
        if (_shm.isValid()) {
            std::string name = _shm.name();
            _shm.close();
            SharedMemory::unlink(name);
        }
    }
 
    [[nodiscard]] bool isValid() const noexcept
    {
        return _shm.isValid() && _queue.isValid();
    }
 
    [[nodiscard]] bool empty() const noexcept
    {
        return _queue.empty();
    }
 
    [[nodiscard]] constexpr std::size_t capacity() const noexcept
    {
        return Capacity;
    }
 
private:
    SharedMemory _shm;
    QueueType _queue;
};
 
} // namespace corium::ipc