MessageBatch.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 <vector>
 
#include <OnixS/ICE/BOE/messaging/MessageHolder.h>
#include <OnixS/ICE/BOE/messaging/SbeMessage.h>
#include <OnixS/ICE/BOE/messaging/NetworkMessage.h>
 
ONIXS_ICEBOE_NAMESPACE_BEGIN
ONIXS_ICEBOE_EXPORTED_CLASS Session;
ONIXS_ICEBOE_NAMESPACE_END
 
ONIXS_ICEBOE_MESSAGING_NAMESPACE_BEGIN
 
typedef std::vector<NetworkMessage> NetMessages;
typedef std::pair<NetworkMessage*, NetworkMessage*> NetMessagesRange;
class MessageBatchChecker;

template <typename MessageTypeT,
size_t MaxMessageSize = GetMaxMessageSize<typename HeldAdapter<MessageTypeT>::SbeType, DefaultMaxGroupItems>::Size,
typename MessageInitializer = FieldsInitPolicy>
class MessageBatch
{
public:
 
    using MsgHolder = MessageHolder<MessageTypeT, MaxMessageSize, MessageInitializer>;
    using MsgHolders = std::vector<MsgHolder>;
 
    using iterator = typename MsgHolders::iterator;
    using const_iterator = typename MsgHolders::const_iterator;

    MessageBatch() = default;

    explicit MessageBatch(size_t size) : batch_(size)
    {
        updateHeaders();
    }

    MessageBatch(std::initializer_list<MsgHolder> list) : batch_(list)
    {
        updateHeaders();
    }
 
    MessageBatch(const MessageBatch &) = delete;
    MessageBatch & operator=(const MessageBatch &) = delete;

    void add(const MsgHolder & msg)
    {
        batch_.push_back(msg);
    }

    void updateHeaders()
    {
        netMsgs_.resize(batch_.size());
 
        for (size_t counter = 0; counter < batch_.size(); ++counter)
            netMsgs_[counter] = batch_[counter].toNetworkMessage();
    }
 
    MsgHolder & operator[](size_t index) { return batch_[index]; }
    const MsgHolder & operator[](size_t index) const { return batch_[index]; }
 
    size_t size() const { return batch_.size(); }
 
    const_iterator begin() const { return batch_.begin(); }
    iterator begin() { return batch_.begin(); }
 
    const_iterator end() const { return batch_.end(); }
    iterator end() { return batch_.end(); }
 
private:
 
    friend class ONIXS_ICEBOE_NAMESPACE::Session;
    friend class ONIXS_ICEBOE_MESSAGING_NAMESPACE::MessageBatchChecker;
 
    MsgHolders batch_;
    NetMessages netMsgs_;
};

class MessageBatchCombiner
{
public:

    MessageBatchCombiner() = default;

    template <typename MessageTypeT,
    size_t MaxMessageSize = GetMaxMessageSize<typename HeldAdapter<MessageTypeT>::SbeType, DefaultMaxGroupItems>::Size,
    typename MessageInitializer = FieldsInitPolicy>
    void add(MessageBatch<MessageTypeT, MaxMessageSize, MessageInitializer> & batch)
    {
        netMsgs_.reserve(batch.size());
        for (auto & holder : batch)
            netMsgs_.push_back(holder.toNetworkMessage());
    }
 
    template <typename MessageTypeT,
    size_t MaxMessageSize = GetMaxMessageSize<typename HeldAdapter<MessageTypeT>::SbeType, DefaultMaxGroupItems>::Size,
    typename MessageInitializer = FieldsInitPolicy>
    void add(MessageBatch<MessageTypeT, MaxMessageSize, MessageInitializer> && batch) = delete;

    template <typename MessageTypeT,
    size_t MaxMessageSize = GetMaxMessageSize<typename HeldAdapter<MessageTypeT>::SbeType, DefaultMaxGroupItems>::Size,
    typename MessageInitializer = FieldsInitPolicy>
    void add(MessageHolder<MessageTypeT, MaxMessageSize, MessageInitializer> & holder)
    {
        netMsgs_.push_back(holder.toNetworkMessage());
    }
 
    template <typename MessageTypeT,
    size_t MaxMessageSize = GetMaxMessageSize<typename HeldAdapter<MessageTypeT>::SbeType, DefaultMaxGroupItems>::Size,
    typename MessageInitializer = FieldsInitPolicy>
    void add(MessageHolder<MessageTypeT, MaxMessageSize, MessageInitializer> && holder) = delete;

    void clear()
    {
        netMsgs_.clear();
    }

    NetMessagesRange getRange(size_t index) noexcept
    {
        assert(!netMsgs_.empty());
        return {&*netMsgs_.begin(), &*(netMsgs_.begin() + index)};
    }
 
    MessageBatchCombiner(const MessageBatchCombiner &) = delete;
    MessageBatchCombiner & operator=(const MessageBatchCombiner &) = delete;
 
private:
 
    friend class ONIXS_ICEBOE_NAMESPACE::Session;
    friend class ONIXS_ICEBOE_MESSAGING_NAMESPACE::MessageBatchChecker;
 
    NetMessages netMsgs_;
};
 
class MessageBatchChecker
{
public:
    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    static size_t calculateBatchSize(
        MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs) noexcept
    {
        return calculateBatchSize(msgs.netMsgs_);
    }

    template <typename SbeMessageType, size_t MaxMessageSize, typename MessageInitializer>
    static size_t fitSize(
        MessageBatch<SbeMessageType, MaxMessageSize, MessageInitializer> & msgs, size_t size = defaultMaxPacketSize()) noexcept
    {
        return calculateBatchSize(msgs) <= size;
    }

    static size_t calculateBatchSize(MessageBatchCombiner & msgs) noexcept
    {
        return calculateBatchSize(msgs.netMsgs_);
    }

    static bool fitSize(MessageBatchCombiner & msgs, size_t size = defaultMaxPacketSize()) noexcept
    {
        return calculateBatchSize(msgs) <= size;
    }
 
private:
    ONIXS_ICEBOE_EXPORTED
    static size_t calculateBatchSize(const NetMessages& msgs) noexcept;
 
    ONIXS_ICEBOE_EXPORTED
    static size_t defaultMaxPacketSize() noexcept;
};
 
ONIXS_ICEBOE_MESSAGING_NAMESPACE_END