IMessageQueue.h#

// Copyright (c) 2024, NVIDIA CORPORATION. All rights reserved.
//
// NVIDIA CORPORATION and its licensors retain all intellectual property
// and proprietary rights in and to this software, related documentation
// and any modifications thereto. Any use, reproduction, disclosure or
// distribution of this software and related documentation without an express
// license agreement from NVIDIA CORPORATION is strictly prohibited.
//

#pragma once

#include "MessageQueueTypes.h"
#include "../../omni/Expected.h"
#include "../IObject.h"

namespace carb
{

namespace eventdispatcher
{

class IMessageQueue : public carb::IObject
{
public:
    using ExpectedType = omni::expected<void, omni::core::Result>;

    using UnexpectedType = omni::unexpected<omni::core::Result>;

    virtual RStringKey getName() const = 0;

    virtual thread::ThreadId getOwningThread() const = 0;

    virtual bool hasMessages() const = 0;

    virtual void internalPeek(ExpectedType& result, MessageFn fn, void* userData) const = 0;

    virtual void internalPop(ExpectedType& result, MessageFn fn, void* userData) = 0;

    virtual void internalAwaitMessage(ExpectedType& result, uint64_t nsToWait, NotifyFn notify, void* userData) const = 0;

    virtual void internalStop(ExpectedType& result) = 0;

    template <class Func>
    ExpectedType peek(Func&& func) const;

    template <class Func>
    ExpectedType pop(Func&& func);

    omni::expected<void, omni::core::Result> awaitMessage() const;

    template <class Rep, class Period>
    omni::expected<void, omni::core::Result> awaitMessageFor(const std::chrono::duration<Rep, Period>& duration) const;

    template <class Clock, class Duration>
    omni::expected<void, omni::core::Result> awaitMessageUntil(const std::chrono::time_point<Clock, Duration>& when) const;

    ExpectedType stop();

    virtual void internalPush(ExpectedType& result, const EventData& data, bool wait, NotifyFn notify, void* userData) = 0;

    template <class... Args>
    ExpectedType pushAsync(RString eventName, Args&&... payload);

    template <class InIter>
    ExpectedType pushAsyncIter(RString eventName, InIter begin, InIter end);

    template <class R CARB_NO_DOC(
        ,
        std::enable_if_t<std::is_same<void, cpp::void_t<decltype(std::begin(std::declval<R>()), std::end(std::declval<R>()))>>::value,
                         bool> = false)>
    ExpectedType pushAsyncRange(RString eventName, R&& range)
    {
        // Implementation needs to be inline because certain compilers get very confused about the SFINAE
        std::vector<NamedVariant> variants;
        for (auto& r : range)
            variants.emplace_back(detail::translate(r));
        std::sort(variants.begin(), variants.end(), detail::NamedVariantLess{});
        CARB_ASSERT(std::adjacent_find(variants.begin(), variants.end(), detail::NamedVariantEqual{}) == variants.end(),
                    "Message has duplicate keys");
        EventData data{ eventName, variants.size(), variants.data() };
        ExpectedType result;
        internalPush(result, data, false, nullptr, nullptr);
        return result;
    }
#ifndef DOXYGEN_BUILD
    // Enable implicit initializer_list since {} cannot be deduced to a type
    template <class T>
    ExpectedType pushAsyncRange(RString eventName, std::initializer_list<std::pair<RStringKey, T>> range)
    {
        return pushAsyncRange<std::initializer_list<std::pair<RStringKey, T>>>(eventName, std::move(range));
    }
#endif

    template <class... Args>
    ExpectedType pushAndWait(RString eventName, Args&&... payload);

    template <class R CARB_NO_DOC(
        ,
        std::enable_if_t<std::is_same<void, cpp::void_t<decltype(std::begin(std::declval<R>()), std::end(std::declval<R>()))>>::value,
                         bool> = false)>
    ExpectedType pushAndWaitRange(RString eventName, R&& range)
    {
        // Implementation needs to be inline because certain compilers get very confused about the SFINAE
        std::vector<NamedVariant> variants;
        for (auto& r : range)
            variants.emplace_back(detail::translate(r));
        std::sort(variants.begin(), variants.end(), detail::NamedVariantLess{});
        CARB_ASSERT(std::adjacent_find(variants.begin(), variants.end(), detail::NamedVariantEqual{}) == variants.end(),
                    "Message has duplicate keys");
        EventData data{ eventName, variants.size(), variants.data() };
        ExpectedType result;
        internalPush(result, data, true, nullptr, nullptr);
        return result;
    }
#ifndef DOXYGEN_BUILD
    // Enable implicit initializer_list since {} cannot be deduced to a type
    template <class T>
    ExpectedType pushAndWaitRange(RString eventName, std::initializer_list<std::pair<RStringKey, T>> range)
    {
        return pushAndWaitRange<std::initializer_list<std::pair<RStringKey, T>>>(eventName, std::move(range));
    }
#endif

    template <class InIter>
    ExpectedType pushAndWaitIter(RString eventName, InIter begin, InIter end);

};

using IMessageQueuePtr = carb::ObjectPtr<IMessageQueue>;

struct IMessageQueueFactory
{
    CARB_PLUGIN_INTERFACE("carb::eventdispatcher::IMessageQueueFactory", 1, 0);

    IMessageQueue*(CARB_ABI* internalCreateMessageQueue)(RStringKey name,
                                                         const MessageQueueParams& params,
                                                         omni::core::Result& result,
                                                         bool& existing);
    IMessageQueue*(CARB_ABI* internalGetMessageQueue)(RStringKey name);

    omni::expected<std::pair<IMessageQueuePtr, bool>, omni::core::Result> createMessageQueue(
        RStringKey name, const MessageQueueParams& params);

    IMessageQueuePtr getMessageQueue(RStringKey name);
};

} // namespace eventdispatcher
} // namespace carb

#ifndef __INTELLISENSE__
#    include "IMessageQueue.inl"
#endif