DataFeed.h

Go to the documentation of this file.

// Copyright Onix Solutions Limited [OnixS]. All rights reserved.
//
// This software owned by Onix Solutions Limited [OnixS] and is protected by copyright law
// and international copyright treaties.
//
// Access to and use of the software is governed by the terms of the applicable OnixS Software
// Services Agreement (the Agreement) and Customer end user license agreements granting
// a non-assignable, non-transferable and non-exclusive license to use the software
// for it's own data processing purposes under the terms defined in the Agreement.
//
// Except as otherwise granted within the terms of the Agreement, copying or reproduction of any part
// of this source code or associated reference material to any other location for further reproduction
// or redistribution, and any amendments to this copyright notice, are expressly prohibited.
//
// Any reproduction or redistribution for sale or hiring of the Software not in accordance with
// the terms of the Agreement is a violation of copyright law.
//
 
#pragma once
 
#include <array>
#include <type_traits>
#include <utility>
#include <cstddef>
 
#include <OnixS/CME/MDH/FeedAttributes.h>
#include <OnixS/CME/MDH/Watch.h>
#include <OnixS/CME/MDH/testing/MessageOperation.h>
#include <OnixS/CME/MDH/testing/PacketHolder.h>
#include <OnixS/CME/MDH/testing/Utils.h>
#include <OnixS/CME/MDH/threading/Future.h>
 
ONIXS_CME_MDH_TESTING_NAMESPACE_BEGIN

namespace details
{
    template <typename Feed, typename... Arguments>
    struct IsDataFeedSelfConstruction : std::false_type{};
 
    template <typename Feed, typename Argument>
    struct IsDataFeedSelfConstruction<Feed, Argument>
        : std::is_base_of<Feed, typename std::decay<Argument>::type>{};
 
    template <typename Message>
    struct IsSpanCompatibleMessage
    {
    private:
        template <typename T>
        static auto test(int) -> decltype(asSpan(std::declval<T>()), std::true_type{});
 
        template <typename>
        static std::false_type test(...);
 
    public:
        static constexpr bool value = decltype(test<Message>(0))::value;
    };
 
    template <typename Message>
    struct IsMaterializableMessage : std::integral_constant<bool,
        IsMessageRecipe<typename std::decay<Message>::type>::value ||
        IsSpanCompatibleMessage<Message>::value>
    {};
 
    template <typename Message, bool = IsMessageRecipe<typename std::decay<Message>::type>::value>
    class MaterializedMessage
    {
    public:
        explicit MaterializedMessage(Message&& message) noexcept
            : message_(std::forward<Message>(message))
        {
        }
 
        ByteSpan span()
        {
            return asSpan(std::forward<Message>(message_));
        }
 
    private:
        Message&& message_;
    };
 
    template <typename Message>
    class MaterializedMessage<Message, true>
    {
        using Recipe = typename std::decay<Message>::type;
 
    public:
        explicit MaterializedMessage(const Recipe& recipe)
        {
            recipe.applyTo(holder_);
        }
 
        ByteSpan span() noexcept
        {
            return asSpan(holder_);
        }
 
    private:
        typename Recipe::HolderType holder_;
    };
}

template <typename Delivery>
class DataFeed : private Delivery
{
    template <bool...> struct BooleanPack{};
    template <bool... Values> using All = std::is_same<BooleanPack<Values..., true>, BooleanPack<true, Values...>>;
 
protected:
    enum : std::size_t { SendMaxSpansCount = Delivery::MaxSpansCount };

    template <typename... Messages>
    using MessageSpans = std::array<ByteSpan, sizeof...(Messages)>;

    template <typename Result, typename... Messages>
    using EnableSend = typename std::enable_if<
        (sizeof...(Messages) > 0)
            && (sizeof...(Messages) < SendMaxSpansCount)
            && All<!std::is_base_of<PacketHeader, typename std::decay<Messages>::type>::value...>::value
            && All<details::IsMaterializableMessage<Messages>::value...>::value,
        Result>::type;

    template <typename Result, typename... Messages>
    using EnablePacketHolderSend = typename std::enable_if<(sizeof...(Messages) < SendMaxSpansCount), Result>::type;
 
public:
    enum : std::size_t { MaxSpansCount = Delivery::MaxSpansCount };
    static_assert(MaxSpansCount >= 2, "must include the packet header and at least one message.");

    template <typename... Arguments, typename std::enable_if<
        !details::IsDataFeedSelfConstruction<DataFeed, Arguments...>::value, int>::type = 0>
    explicit DataFeed(Arguments&&... arguments)
        : Delivery(std::forward<Arguments>(arguments)...)
    {
    }

    virtual ~DataFeed() = default;

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(SequenceNumber seqNum, Messages&&... messages) -> EnableSend<Threading::Event, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendMessages consumer{*this, seqNum, Messaging::Timestamp{}, true};
        return materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(
        SequenceNumber seqNum, Messaging::Timestamp sendingTime,
        Messages&&... messages) -> EnableSend<Threading::Event, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendMessages consumer{*this, seqNum, sendingTime, false};
        return materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    ONIXS_CMEMDH_NODISCARD
    Threading::Event send(SequenceNumber seqNum, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        return send(seqNum, UtcWatch::now(), checked.data, checked.count);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    ONIXS_CMEMDH_NODISCARD
    Threading::Event send(SequenceNumber seqNum, Messaging::Timestamp sendingTime, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        return send(seqNum, sendingTime, checked.data, checked.count);
    }

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<Threading::Event, Messages...>
    {
        return sendPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<Threading::Event, Messages...>
    {
        return sendPacketHolder(seqNum, sendingTime, packet);
    }

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(SequenceNumber seqNum, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<Threading::Event, Messages...>
    {
        return sendPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    ONIXS_CMEMDH_NODISCARD
    auto send(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<Threading::Event, Messages...>
    {
        return sendPacketHolder(seqNum, sendingTime, packet);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, Messages&&... messages) -> EnableSend<void, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendPrimaryMessages consumer{*this, seqNum, Messaging::Timestamp{}, true};
        materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, Messages&&... messages)
        -> EnableSend<void, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendPrimaryMessages consumer{*this, seqNum, sendingTime, false};
        materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    void sendPrimary(SequenceNumber seqNum, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        sendPrimary(seqNum, UtcWatch::now(), checked.data, checked.count);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    void sendPrimary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        sendPrimary(seqNum, sendingTime, checked.data, checked.count);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendPrimaryPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendPrimaryPacketHolder(seqNum, sendingTime, packet);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendPrimaryPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    auto sendPrimary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendPrimaryPacketHolder(seqNum, sendingTime, packet);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, Messages&&... messages) -> EnableSend<void, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendSecondaryMessages consumer{*this, seqNum, Messaging::Timestamp{}, true};
        materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, Messages&&... messages)
        -> EnableSend<void, Messages...>
    {
        validateMessageCount<Messages...>();
        MessageSpans<Messages...> messageSpans{};
        SendSecondaryMessages consumer{*this, seqNum, sendingTime, false};
        materializeMessages<0>(consumer, messageSpans, std::forward<Messages>(messages)...);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    void sendSecondary(SequenceNumber seqNum, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        sendSecondary(seqNum, UtcWatch::now(), checked.data, checked.count);
    }

    template <class Collection, details::EnableIfByteSpanCollection<Collection> = 0>
    void sendSecondary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, Collection&& messages)
    {
        const auto checked = details::checkedSpans(messages, MaxSpansCount - 1, 1);
        sendSecondary(seqNum, sendingTime, checked.data, checked.count);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendSecondaryPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendSecondaryPacketHolder(seqNum, sendingTime, packet);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendSecondaryPacketHolder(seqNum, packet);
    }

    template <typename... Messages>
    auto sendSecondary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>&& packet)
        -> EnablePacketHolderSend<void, Messages...>
    {
        sendSecondaryPacketHolder(seqNum, sendingTime, packet);
    }

    ONIXS_CMEMDH_NODISCARD
    NetFeedRole::Enum primaryFeed() const noexcept
    {
        return Delivery::primaryFeed();
    }

    ONIXS_CMEMDH_NODISCARD
    NetFeedRole::Enum secondaryFeed() const noexcept
    {
        return Delivery::secondaryFeed();
    }

    ONIXS_CMEMDH_NODISCARD
    bool hasSecondary() const noexcept
    {
        return Delivery::hasSecondary();
    }
 
private:
    DataFeed(const DataFeed&) = delete;
    DataFeed& operator=(const DataFeed&) = delete;
    DataFeed(DataFeed&&) = delete;
    DataFeed& operator=(DataFeed&&) = delete;
 
    struct SendMessages
    {
        using Result = Threading::Event;
 
        DataFeed& feed;
        SequenceNumber sequenceNumber;
        Messaging::Timestamp sendingTime;
        bool useCurrentTime;
 
        Threading::Event operator()(const ByteSpan* spans, size_t count) const
        {
            return feed.send(
                sequenceNumber, useCurrentTime ? UtcWatch::now() : sendingTime,
                spans, count);
        }
    };
 
    struct SendPrimaryMessages
    {
        using Result = void;
 
        DataFeed& feed;
        SequenceNumber sequenceNumber;
        Messaging::Timestamp sendingTime;
        bool useCurrentTime;
 
        void operator()(const ByteSpan* spans, size_t count) const
        {
            feed.sendPrimary(sequenceNumber, useCurrentTime ? UtcWatch::now() : sendingTime, spans, count);
        }
    };
 
    struct SendSecondaryMessages
    {
        using Result = void;
 
        DataFeed& feed;
        SequenceNumber sequenceNumber;
        Messaging::Timestamp sendingTime;
        bool useCurrentTime;
 
        void operator()(const ByteSpan* spans, size_t count) const
        {
            feed.sendSecondary(sequenceNumber, useCurrentTime ? UtcWatch::now() : sendingTime, spans, count);
        }
    };
 
    template <std::size_t Index, typename Consumer, std::size_t Count>
    static typename std::enable_if<Index == Count, typename Consumer::Result>::type
        materializeMessages(Consumer& consumer, std::array<ByteSpan, Count>& spans)
    {
        return consumer(spans.data(), spans.size());
    }
 
    template <std::size_t Index, typename Consumer, std::size_t Count, typename Message, typename... Messages>
    static typename Consumer::Result
        materializeMessages(Consumer& consumer, std::array<ByteSpan, Count>& spans, Message&& message, Messages&&... messages)
    {
        details::MaterializedMessage<Message> materialized(std::forward<Message>(message));
        spans[Index] = materialized.span();
        return materializeMessages<Index + 1>(consumer, spans, std::forward<Messages>(messages)...);
    }
 
    template <typename... Messages>
    Threading::Event sendPacketHolder(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        return send(seqNum, UtcWatch::now(), messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    Threading::Event sendPacketHolder(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        return send(seqNum, sendingTime, messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    void sendPrimaryPacketHolder(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        sendPrimary(seqNum, UtcWatch::now(), messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    void sendPrimaryPacketHolder(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        sendPrimary(seqNum, sendingTime, messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    void sendSecondaryPacketHolder(SequenceNumber seqNum, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        sendSecondary(seqNum, UtcWatch::now(), messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    void sendSecondaryPacketHolder(SequenceNumber seqNum, Messaging::Timestamp sendingTime, PacketHolder<Messages...>& packet)
    {
        const auto messageSpans = makeMessageSpans(packet);
        sendSecondary(seqNum, sendingTime, messageSpans.data(), messageSpans.size());
    }
 
    template <typename... Messages>
    static typename PacketHolder<Messages...>::MessageSpans makeMessageSpans(PacketHolder<Messages...>& packet) noexcept
    {
        validateMessageCount<Messages...>();
        return packet.toNetworkMessageSpans();
    }
 
    template <typename... Messages>
    static void validateMessageCount() noexcept
    {
        static_assert(sizeof...(Messages) < MaxSpansCount, "packet contains too many messages.");
    }
 
    template <typename Sender>
    static auto sendPacket(SequenceNumber seqNum, Messaging::Timestamp sendingTime, const ByteSpan messages[], size_t messagesCnt, Sender&& sender)
        -> decltype(sender(std::declval<const ByteSpan*>(), std::size_t{}))
    {
        PacketHeader packetHeader;
        packetHeader.seqNumber(seqNum);
        packetHeader.sendingTime(sendingTime);
 
        std::array<ByteSpan, MaxSpansCount> packet{};
        packet[0] = asSpan(packetHeader);
 
        for (std::size_t index = 0; index < messagesCnt; ++index)
            packet[index + 1] = messages[index];
 
        return sender(packet.data(), messagesCnt + 1);
    }
 
    Threading::Event send(SequenceNumber seqNum, Messaging::Timestamp sendingTime, const ByteSpan messages[], std::size_t messagesCnt)
    {
        auto sender =
            [this](const ByteSpan packet[], std::size_t spansCount) { return Delivery::send(packet, spansCount);};
 
        return sendPacket(seqNum, sendingTime, messages, messagesCnt, sender);
    }
 
    void sendPrimary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, const ByteSpan messages[], std::size_t messagesCnt)
    {
        auto sender =
            [this](const ByteSpan packet[], std::size_t spansCount) { Delivery::sendPrimary(packet, spansCount);};
 
        sendPacket(seqNum, sendingTime, messages, messagesCnt, sender);
    }
 
    void sendSecondary(SequenceNumber seqNum, Messaging::Timestamp sendingTime, const ByteSpan messages[], std::size_t messagesCnt)
    {
        auto sender =
            [this](const ByteSpan packet[], std::size_t spansCount) { Delivery::sendSecondary(packet, spansCount);};
 
        sendPacket(seqNum, sendingTime, messages, messagesCnt, sender);
    }
};
 
ONIXS_CME_MDH_TESTING_NAMESPACE_END