SBE Customization Sample

This sample demonstrates how to customize SBE encoding/decoding process.

Source code:

#include "../../Common/PrecisionTimer.h"
#include "../../Common/Helpers.h"
#include "../../Common/Settings.h"

using namespace Settings;

namespace {
#pragma pack(push, 1)

struct MessageHeaderComposite
{
    UInt16 blockLength;
    UInt16 templateId;
    UInt16 schemaId;
    UInt16 version;
};

struct PacketHeaderComposite
{
    UInt32 sequenceNumber;
    UInt64 sendingTime;
};

struct PreMessageHeaderComposite
{
    UInt16 messageSize;
};

#pragma pack(pop)

class CustomCodersLibrary : public ISbeCustomCoderLibrary
{
public:
    void queryCodingControl(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED,
        SbeFieldMetaData field ONIXS_FIXENGINE_UNUSED,
        SbeFieldCodingSettings manageCoding) ONIXS_FIXENGINE_OVERRIDE {
        manageCoding.useFormattedDateTime(false); // Lets suppress formatting
        manageCoding.useValueNamesExternally(false);
    }

    ISbeFieldDecoder* queryDecoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED, SbeFieldMetaData field) ONIXS_FIXENGINE_OVERRIDE {
        if (!field.typeMetaData().name().compare("userVarString"))
            return &userVarStringDecoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    ISbeFieldEncoder* queryEncoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED, SbeFieldMetaData field) ONIXS_FIXENGINE_OVERRIDE {
        if (!field.typeMetaData().name().compare("userVarString"))
            return &userVarStringEncoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    ISbeGroupDimensionDecoder* queryGroupDimensionDecoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED,
        SbeFieldSetMetaData field) ONIXS_FIXENGINE_OVERRIDE {
        if (field.isGroup() && !field.groupDimensionType().name().compare("userGroupSizeEncoding"))
            return &userGroupSizeDecoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    ISbeGroupDimensionEncoder* queryGroupDimensionEncoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED, SbeFieldSetMetaData field) ONIXS_FIXENGINE_OVERRIDE {
        if (field.isGroup() && !field.groupDimensionType().name().compare("userGroupSizeEncoding"))
            return &userGroupSizeEncoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    ISbeCompositeDecoder* queryDecoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED, SbeTypeMetaData type) ONIXS_FIXENGINE_OVERRIDE {
        if (!type.name().compare("userMessageHeader"))
            return &userMessageHeaderDecoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    ISbeCompositeEncoder* queryEncoder(CoderOwnerId owner ONIXS_FIXENGINE_UNUSED, SbeTypeMetaData field) ONIXS_FIXENGINE_OVERRIDE {
        if (!field.name().compare("userMessageHeader"))
            return &userMessageHeaderEncoder_;
        return ONIXS_FIXENGINE_NULLPTR;
    }

    static unsigned getUnsigned(const char* src, size_t length) {
        unsigned result = 0;
        for (const char* current = src; (current - src) < (int)length; ++current) {
            const char c = *current;
            if (c >= '0' && c <= '9') {
                result *= 10;
                result += (c - '0');
            }
            else {
                break;
            }
        }
        return result;
    }

    // Coders for userVarString type:
    //
    //<composite name="userVarString" description="User-defined variable length string">
    //     <type name="length" primitiveType="char" length="10" semanticType="Length"/>
    //     <type name="data" length="0" primitiveType="uint8" semanticType="data" characterEncoding="UTF-16"/>
    //</composite>

    static const size_t VarStringLengthFieldSize = 10;

    class UserVarStringDecoder : public ISbeFieldDecoder
    {
    public:

        size_t decode(SbeFieldDecoderOutput output, unsigned /* version */, const void* data, size_t available) ONIXS_FIXENGINE_OVERRIDE
        {
            if (available < VarStringLengthFieldSize)
                throw std::domain_error("Not enough data to decode data length");

            // Read unsigned number
            const char* chars = static_cast<const char*>(data);
            unsigned length = getUnsigned(chars, VarStringLengthFieldSize);

            if ((available - VarStringLengthFieldSize) < length)
                throw std::domain_error("Not enough data to complete decoding");

            output.set(chars + VarStringLengthFieldSize, (SbeWireSize)length);
            return VarStringLengthFieldSize + length;
        }
    };

    UserVarStringDecoder userVarStringDecoder_;

    class UserVarStringEncoder : public ISbeFieldEncoder
    {
    public:

        SbeWireSize queryFixedSize(unsigned /* version */) ONIXS_FIXENGINE_OVERRIDE {
            return 0;
        }
        void encode(SbeBinaryOutputStream out, unsigned /* version */, const SbeConstFieldAccess& field) ONIXS_FIXENGINE_OVERRIDE {
            std::string s;
            if (!field.toString(s))
                throw std::domain_error("Unable to convert value to string");

            char lengthBuffer[VarStringLengthFieldSize];
            memset(lengthBuffer, 0, VarStringLengthFieldSize);
            sprintf(lengthBuffer, "%d", (int)s.length());

            out.put(lengthBuffer, VarStringLengthFieldSize);
            out.put(s.c_str(), (SbeWireSize)s.length());
        }
    };

    UserVarStringEncoder userVarStringEncoder_;


    // Coders for customized group dimension format:
    //
    //<composite name="userGroupSizeEncoding" description="Customized repeating group dimensions">
    //    <type name="blockLength" primitiveType="uint16" semanticType="Length"/>
    //    <type name="numInGroup" primitiveType="char" length="6" semanticType="NumInGroup"/>
    //</composite>

    static const size_t NumInGroupFieldSize = 6;

#pragma pack(push, 1)
    struct UserGroupSizeLayout
    {
        UInt16 blockLength;
        char numInGroup[NumInGroupFieldSize];
    };
#pragma pack(pop)

    class UserGroupSizeDecoder : public ISbeGroupDimensionDecoder
    {
    public:
        size_t decode(SbeGroupDimensionOutput output, unsigned /* version */, const void* data, size_t available) ONIXS_FIXENGINE_OVERRIDE
        {
            SbeFieldSetMetaData metadata = output.group();
            std::cout << "Decodes data for "
                << (metadata.isGroup() ? "group" : "fieldset") << ": "
                << "<" << metadata.tag() << "> "
                << "that uses type '" << metadata.groupDimensionType().name() << "' to store dimension."
                << std::endl;

            if (available < sizeof(UserGroupSizeLayout))
                throw std::domain_error("Not enough data to decode group dimension");

            // Map raw data to the plain C-structure:
            const UserGroupSizeLayout* groupSize = static_cast<const UserGroupSizeLayout*>(data);

            // Read unsigned number
            unsigned numInGroups = getUnsigned(groupSize->numInGroup, NumInGroupFieldSize);

            // Send decoded data back to the main SBE decoder.
            output.set(groupSize->blockLength, (SbeWireSize)numInGroups);

            return sizeof(UserGroupSizeLayout);
        }
    };

    UserGroupSizeDecoder userGroupSizeDecoder_;

    class UserGroupSizeEncoder : public ISbeGroupDimensionEncoder
    {
    public:
        void encode(SbeBinaryOutputStream out, unsigned /* version */, SbeGroupDimensionInput input) ONIXS_FIXENGINE_OVERRIDE
        {
            SbeFieldSetMetaData metadata = input.groupMetaData();
            std::cout << "Encodes data for "
                << (metadata.isGroup() ? "group" : "fieldset") << ": "
                << "<" << metadata.tag() << "> "
                << "that uses type '" << metadata.groupDimensionType().name() << "' to store dimension."
                << std::endl;

            UserGroupSizeLayout groupSize;
            memset(&groupSize, 0, sizeof(groupSize));

            SbeWireSize blockLength = 0;
            size_t numInGroups = 0;
            input.getGroupDimensionData(&blockLength, &numInGroups);

            groupSize.blockLength = (UInt16)blockLength;
            memset(groupSize.numInGroup, 0, NumInGroupFieldSize);
            sprintf(groupSize.numInGroup, "%d", (int)numInGroups);

            out.put(&groupSize, sizeof(groupSize));
        }
    };

    UserGroupSizeEncoder userGroupSizeEncoder_;

#pragma pack(push, 1)
    struct UserMessageHeaderLayout
    {
        UInt16 blockLength;
        UInt16 templateId;
        unsigned char schemaId;
        unsigned char version;
    };
#pragma pack(pop)

    class UserMessageHeaderDecoder : public ISbeCompositeDecoder
    {
    public:
        size_t decode(SbeSpecialCompositeOutput output, unsigned /* version */, const void* data, size_t available) ONIXS_FIXENGINE_OVERRIDE
        {
            SbeTypeMetaData metadata = output.type();
            std::cout << "Decodes data for free composite \"" << metadata.name() << "\"" << std::endl;

            if (available < sizeof(UserMessageHeaderLayout))
                throw std::domain_error("Not enough data to decode group dimension");

            // Map raw data to the plain C-structure:
            const UserMessageHeaderLayout* messageHeader = static_cast<const UserMessageHeaderLayout*>(data);

            // Send decoded data back to the main SBE decoder.
            output.setMessageHeader(messageHeader->blockLength, messageHeader->templateId, messageHeader->schemaId, messageHeader->version);

            return sizeof(UserMessageHeaderLayout);
        }
    };

    UserMessageHeaderDecoder userMessageHeaderDecoder_;

    class UserMessageHeaderEncoder : public ISbeCompositeEncoder
    {
    public:
        void encode(SbeBinaryOutputStream out, unsigned /* version */, SbeSpecialCompositeInput input) ONIXS_FIXENGINE_OVERRIDE
        {
            SbeTypeMetaData metadata = input.typeMetaData();
            std::cout << "Encodes data for free composite \"" << metadata.name() << "\"" << std::endl;

            // Prepare clear structure
            UserMessageHeaderLayout messageHeader;
            memset(&messageHeader, 0, sizeof(messageHeader));

            UInt32 blockLength, templateId, schemaId, version;
            input.getMessageHeaderData(&blockLength, &templateId, &schemaId, &version);

            messageHeader.blockLength = (UInt16)blockLength;
            messageHeader.templateId = (UInt16)templateId;
            messageHeader.schemaId = (char)schemaId;
            messageHeader.version = (char)version;

            out.put(&messageHeader, sizeof(messageHeader));
        }
    };

    UserMessageHeaderEncoder userMessageHeaderEncoder_;

};

void encodeAndDecode();

};

int main(int, const char * [])
{
    try {
        EngineSettings engineSettings;
        engineSettings.listenPort(-1)  // ListenPort is set to '-1' to disable the telecommunication level.
                      .licenseStore(LicenseStore);

        Engine::init(engineSettings);

        encodeAndDecode();

        waitUntilEnterKey();
    }
    catch(const Exception & ex) {
        processSampleException(ex.what());
        return 3;
    }
}

namespace {

void encodeAndDecode() {
    CustomCodersLibrary coderLib;

    const std::string sbeTemplates = readTextFile("CustomizedSbeTemplate.xml");
    Dictionary dict = Decoder::generateFixDictionary(sbeTemplates);

    Message msg("X", dict);
    msg.set(60, 12345);
    msg.set(5799, 2);
    Group group = msg.setGroup(268, 2);
    group[0].set(279, 0);
    group[0].set(269, 'J');
    group[0].set(1180, 100);
    group[0].set(8010, "Text for item 0");


    group[1].set(279, 0);
    group[1].set(269, 'J');
    group[1].set(1180, 100);
    msg.set(8001, "Variable length text");
    msg.set(8002, "Another variable length text");

    msg.updateBodyLengthAndCheckSum();

    std::cout << msg.toString() << std::endl;

    Encoder encoder(sbeTemplates, &coderLib);

    std::cout << "Message header type for encoding: " << encoder.schemaHeaderType() << std::endl;

    unsigned char buffer[1024];
    size_t rootBlockLength = 0;
    size_t encoded = encoder.encodeWithHeader(msg, 104, 3, buffer, sizeof(buffer), &rootBlockLength);

    Decoder decoder(sbeTemplates, dict, &coderLib);
    int templateId = 0;
    int version = 0;
    const Message& decodedMsg = decoder.decode(buffer, encoded, ONIXS_FIXENGINE_NULLPTR, &templateId, &version);

    std::cout << decodedMsg.toString() << std::endl;
}

};