DataStorm.h

Go to the documentation of this file.

// **********************************************************************
//
// Copyright (c) 2018-present ZeroC, Inc. All rights reserved.
//
// **********************************************************************

#pragma once

#include <DataStorm/Config.h>

#include <Ice/Ice.h>

#include <DataStorm/Types.h>
#include <DataStorm/Sample.h>
#include <DataStorm/Node.h>
#include <DataStorm/InternalI.h>
#include <DataStorm/InternalT.h>
#include <DataStorm/CtrlCHandler.h>

#include <regex>

namespace DataStorm
{

template<typename Key, typename Value, typename UpdateTag=std::string> class Sample
{
public:

    using KeyType = Key;

    using ValueType = Value;

    using UpdateTagType = UpdateTag;

    SampleEvent getEvent() const noexcept;

    const Key& getKey() const noexcept;

    const Value& getValue() const noexcept;

    UpdateTag getUpdateTag() const noexcept;

    std::chrono::time_point<std::chrono::system_clock> getTimeStamp() const noexcept;

    std::string getOrigin() const noexcept;

    std::string getSession() const noexcept;

    Sample(const std::shared_ptr<DataStormI::Sample>&) noexcept;

private:

    std::shared_ptr<DataStormI::SampleT<Key, Value, UpdateTag>> _impl;
};

std::ostream&
operator<<(std::ostream& os, SampleEvent sampleType)
{
    switch(sampleType)
    {
    case SampleEvent::Add:
        os << "Add";
        break;
    case SampleEvent::Update:
        os << "Update";
        break;
    case SampleEvent::Remove:
        os << "Remove";
        break;
    case SampleEvent::PartialUpdate:
        os << "PartialUpdate";
        break;
    default:
        os << static_cast<int>(sampleType);
        break;
    }
    return os;
}

std::ostream&
operator<<(std::ostream& os, const std::vector<SampleEvent>& types)
{
    os << "[";
    for(auto p = types.begin(); p != types.end(); ++p)
    {
        if(p != types.begin())
        {
            os << ',';
        }
        os << *p;
    }
    os << "]";
    return os;
}

template<typename K, typename V, typename U>
std::ostream&
operator<<(std::ostream& os, const Sample<K, V, U>& sample)
{
    os << sample.getValue();
    return os;
}

template<typename Key, typename Value, typename UpdateTag>
class Reader
{
public:

    using KeyType = Key;

    using ValueType = Value;

    Reader(Reader&& reader) noexcept;

    ~Reader();

    Reader& operator=(Reader&& reader) noexcept;

    bool hasWriters() const noexcept;

    void waitForWriters(unsigned int count = 1) const;

    void waitForNoWriters() const;

    std::vector<std::string> getConnectedWriters() const noexcept;

    std::vector<Key> getConnectedKeys() const noexcept;

    std::vector<Sample<Key, Value, UpdateTag>> getAllUnread() noexcept;

    void waitForUnread(unsigned int count = 1) const;

    bool hasUnread() const noexcept;

    Sample<Key, Value, UpdateTag> getNextUnread();

    void onConnectedKeys(std::function<void(std::vector<Key>)> init,
                         std::function<void(CallbackReason, Key)> update) noexcept;

    void onConnectedWriters(std::function<void(std::vector<std::string>)> init,
                            std::function<void(CallbackReason, std::string)> update) noexcept;

    void onSamples(std::function<void(std::vector<Sample<Key, Value, UpdateTag>>)> init,
                   std::function<void(Sample<Key, Value, UpdateTag>)> queue) noexcept;

protected:

    Reader(const std::shared_ptr<DataStormI::DataReader>& impl) noexcept : _impl(impl)
    {
    }

    std::shared_ptr<DataStormI::DataReader> _impl;
};

template<typename Key, typename Value, typename UpdateTag>
class Writer
{
public:

    using KeyType = Key;

    using ValueType = Value;

    Writer(Writer&& writer) noexcept;

    Writer& operator=(Writer&& writer) noexcept;

    ~Writer();

    bool hasReaders() const noexcept;

    void waitForReaders(unsigned int count = 1) const;

    void waitForNoReaders() const;

    std::vector<std::string> getConnectedReaders() const noexcept;

    std::vector<Key> getConnectedKeys() const noexcept;

    Sample<Key, Value, UpdateTag> getLast();

    std::vector<Sample<Key, Value, UpdateTag>> getAll() noexcept;

    void onConnectedKeys(std::function<void(std::vector<Key>)> init,
                         std::function<void(CallbackReason, Key)> update) noexcept;

    void onConnectedReaders(std::function<void(std::vector<std::string>)> init,
                            std::function<void(CallbackReason, std::string)> update) noexcept;

protected:

    Writer(const std::shared_ptr<DataStormI::DataWriter>& impl) noexcept : _impl(impl)
    {
    }

    std::shared_ptr<DataStormI::DataWriter> _impl;
};

template<typename Key, typename Value, typename UpdateTag=std::string>
class Topic
{
public:

    using KeyType = Key;

    using ValueType = Value;

    using UpdateTagType = UpdateTag;

    using WriterType = Writer<Key, Value, UpdateTag>;

    using ReaderType = Reader<Key, Value, UpdateTag>;

    using SampleType = Sample<Key, Value, UpdateTag>;

    Topic(const Node& node, const std::string& name) noexcept;

    Topic(Topic&& topic) noexcept;

    ~Topic();

    Topic& operator=(Topic&& topic) noexcept;

    bool hasWriters() const noexcept;

    void waitForWriters(unsigned int count = 1) const;

    void waitForNoWriters() const;

    void setWriterDefaultConfig(const WriterConfig& config) noexcept;

    bool hasReaders() const noexcept;

    void waitForReaders(unsigned int count = 1) const;

    void waitForNoReaders() const;

    void setReaderDefaultConfig(const ReaderConfig& config) noexcept;

    template<typename UpdateValue>
    void setUpdater(const UpdateTag& tag, std::function<void (Value&, UpdateValue)> updater) noexcept;

    template<typename Criteria>
    void setKeyFilter(const std::string& name,
                      std::function<std::function<bool (const Key&)> (const Criteria&)> factory) noexcept;

    template<typename Criteria>
    void setSampleFilter(const std::string& name,
                         std::function<std::function<bool (const SampleType&)> (const Criteria&)> factory) noexcept;

private:

    std::shared_ptr<DataStormI::TopicReader> getReader() const noexcept;
    std::shared_ptr<DataStormI::TopicWriter> getWriter() const noexcept;
    std::shared_ptr<Ice::Communicator> getCommunicator() const noexcept;

    template<typename, typename, typename> friend class SingleKeyWriter;
    template<typename, typename, typename> friend class MultiKeyWriter;
    template<typename, typename, typename> friend class SingleKeyReader;
    template<typename, typename, typename> friend class MultiKeyReader;
    template<typename, typename, typename> friend class FilteredKeyReader;

    const std::string _name;
    const std::shared_ptr<DataStormI::TopicFactory> _topicFactory;
    const std::shared_ptr<DataStormI::KeyFactoryT<Key>> _keyFactory;
    const std::shared_ptr<DataStormI::TagFactoryT<UpdateTag>> _tagFactory;

    const std::shared_ptr<DataStormI::FilterManagerT<DataStormI::KeyT<Key>>> _keyFilterFactories;
    const std::shared_ptr<DataStormI::FilterManagerT<DataStormI::SampleT<Key, Value, UpdateTag>>> _sampleFilterFactories;

    mutable std::mutex _mutex;
    mutable std::shared_ptr<DataStormI::TopicReader> _reader;
    mutable std::shared_ptr<DataStormI::TopicWriter> _writer;
    mutable std::map<std::shared_ptr<DataStormI::Tag>, DataStormI::Topic::Updater> _updaters;
};

template<typename T> struct Filter
{
    template<typename TT>
    Filter(const std::string& name, TT&& criteria) noexcept : name(name), criteria(std::forward<TT>(criteria))
    {
    }

    std::string name;

    T criteria;
};

template<typename Key, typename Value, typename UpdateTag=std::string>
class SingleKeyReader : public Reader<Key, Value, UpdateTag>
{
public:

    SingleKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                    const Key& key,
                    const std::string& name = std::string(),
                    const ReaderConfig& config = ReaderConfig()) noexcept;

    template<typename SampleFilterCriteria>
    SingleKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                    const Key& key,
                    const Filter<SampleFilterCriteria>& sampleFilter,
                    const std::string& name = std::string(),
                    const ReaderConfig& config = ReaderConfig()) noexcept;

    SingleKeyReader(SingleKeyReader&& reader) noexcept;

    SingleKeyReader& operator=(SingleKeyReader&& reader) noexcept;
};

template<typename Key, typename Value, typename UpdateTag=std::string>
class MultiKeyReader : public Reader<Key, Value, UpdateTag>
{
public:

    MultiKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                   const std::vector<Key>& keys,
                   const std::string& name = std::string(),
                   const ReaderConfig& config = ReaderConfig()) noexcept;

    template<typename SampleFilterCriteria>
    MultiKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                   const std::vector<Key>& keys,
                   const Filter<SampleFilterCriteria>& sampleFilter,
                   const std::string& name = std::string(),
                   const ReaderConfig& config = ReaderConfig()) noexcept;

    MultiKeyReader(MultiKeyReader&& reader) noexcept;

    MultiKeyReader& operator=(MultiKeyReader&& reader) noexcept;
};

template<typename K, typename V, typename UT>
SingleKeyReader<K, V, UT>
makeSingleKeyReader(const Topic<K, V, UT>& topic,
                    const typename Topic<K, V, UT>::KeyType& key,
                    const std::string& name = std::string(),
                    const ReaderConfig& config = ReaderConfig()) noexcept
{
    return SingleKeyReader<K, V, UT>(topic, key, name, config);
}

template<typename SFC, typename K, typename V, typename UT>
SingleKeyReader<K, V, UT>
makeSingleKeyReader(const Topic<K, V, UT>& topic,
                    const typename Topic<K, V, UT>::KeyType& key,
                    const Filter<SFC>& sampleFilter,
                    const std::string& name = std::string(),
                    const ReaderConfig& config = ReaderConfig()) noexcept
{
    return SingleKeyReader<K, V, UT>(topic, key, sampleFilter, name, config);
}

template<typename K, typename V, typename UT> MultiKeyReader<K, V, UT>
makeMultiKeyReader(const Topic<K, V, UT>& topic,
                   const std::vector<typename Topic<K, V, UT>::KeyType>& keys,
                   const std::string& name = std::string(),
                   const ReaderConfig& config = ReaderConfig()) noexcept
{
    return MultiKeyReader<K, V, UT>(topic, keys, name, config);
}

template<typename SFC, typename K, typename V, typename UT> MultiKeyReader<K, V, UT>
makeMultiKeyReader(const Topic<K, V, UT>& topic,
                   const std::vector<typename Topic<K, V, UT>::KeyType>& keys,
                   const Filter<SFC>& sampleFilter,
                   const std::string& name = std::string(),
                   const ReaderConfig& config = ReaderConfig()) noexcept
{
    return MultiKeyReader<K, V, UT>(topic, keys, sampleFilter, name, config);
}

template<typename K, typename V, typename UT> MultiKeyReader<K, V, UT>
makeAnyKeyReader(const Topic<K, V, UT>& topic,
                 const std::string& name = std::string(),
                 const ReaderConfig& config = ReaderConfig()) noexcept
{
    return MultiKeyReader<K, V, UT>(topic, {}, name, config);
}

template<typename SFC, typename K, typename V, typename UT> MultiKeyReader<K, V, UT>
makeAnyKeyReader(const Topic<K, V, UT>& topic,
                 const Filter<SFC>& sampleFilter,
                 const std::string& name = std::string(),
                 const ReaderConfig& config = ReaderConfig()) noexcept
{
    return MultiKeyReader<K, V, UT>(topic, {}, sampleFilter, name, config);
}

template<typename Key, typename Value, typename UpdateTag=std::string>
class FilteredKeyReader : public Reader<Key, Value, UpdateTag>
{
public:

    template<typename KeyFilterCriteria>
    FilteredKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                      const Filter<KeyFilterCriteria>& keyFilter,
                      const std::string& name = std::string(),
                      const ReaderConfig& config = ReaderConfig());

    template<typename KeyFilterCriteria, typename SampleFilterCriteria>
    FilteredKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                      const Filter<KeyFilterCriteria>& keyFilter,
                      const Filter<SampleFilterCriteria>& sampleFilter,
                      const std::string& name = std::string(),
                      const ReaderConfig& config = ReaderConfig());

    FilteredKeyReader(FilteredKeyReader&& reader) noexcept;

    FilteredKeyReader& operator=(FilteredKeyReader&& reader) noexcept;
};

template<typename KFC, typename K, typename V, typename UT> FilteredKeyReader<K, V, UT>
makeFilteredKeyReader(const Topic<K, V, UT>& topic,
                      const Filter<KFC>& filter,
                      const std::string& name = std::string(),
                      const ReaderConfig& config = ReaderConfig())
{
    return FilteredKeyReader<K, V, UT>(topic, filter, name, config);
}

template<typename KFC, typename SFC, typename K, typename V, typename UT> FilteredKeyReader<K, V, UT>
makeFilteredKeyReader(const Topic<K, V, UT>& topic,
                      const Filter<KFC>& keyFilter,
                      const Filter<SFC>& sampleFilter,
                      const std::string& name = std::string(),
                      const ReaderConfig& config = ReaderConfig())
{
    return FilteredKeyReader<K, V, UT>(topic, keyFilter, sampleFilter, name, config);
}

template<typename Key, typename Value, typename UpdateTag=std::string>
class SingleKeyWriter : public Writer<Key, Value, UpdateTag>
{
public:

    SingleKeyWriter(const Topic<Key, Value, UpdateTag>& topic,
                    const Key& key,
                    const std::string& name = std::string(),
                    const WriterConfig& config = WriterConfig()) noexcept;

    SingleKeyWriter(SingleKeyWriter&& writer) noexcept;

    SingleKeyWriter& operator=(SingleKeyWriter&& writer) noexcept;

    void add(const Value& value) noexcept;

    void update(const Value& value) noexcept;

    template<typename UpdateValue> std::function<void(const UpdateValue&)> partialUpdate(const UpdateTag& tag) noexcept;

    void remove() noexcept;

private:

    const std::shared_ptr<DataStormI::TagFactoryT<UpdateTag>> _tagFactory;
};

template<typename Key, typename Value, typename UpdateTag=std::string>
class MultiKeyWriter : public Writer<Key, Value, UpdateTag>
{
public:

    MultiKeyWriter(const Topic<Key, Value, UpdateTag>& topic,
                   const std::vector<Key>& keys,
                   const std::string& name = std::string(),
                   const WriterConfig& config = WriterConfig()) noexcept;

    MultiKeyWriter(MultiKeyWriter&& writer) noexcept;

    MultiKeyWriter& operator=(MultiKeyWriter&& writer) noexcept;

    void add(const Key& key, const Value& value) noexcept;

    void update(const Key& key, const Value& value) noexcept;

    template<typename UpdateValue> std::function<void(const Key&, const UpdateValue&)>
    partialUpdate(const UpdateTag& tag) noexcept;

    void remove(const Key& key) noexcept;

private:

    const std::shared_ptr<DataStormI::KeyFactoryT<Key>> _keyFactory;
    const std::shared_ptr<DataStormI::TagFactoryT<UpdateTag>> _tagFactory;
};

template<typename K, typename V, typename UT> SingleKeyWriter<K, V, UT>
makeSingleKeyWriter(const Topic<K, V, UT>& topic,
                    const typename Topic<K, V, UT>::KeyType& key,
                    const std::string& name = std::string(),
                    const WriterConfig& config = WriterConfig()) noexcept
{
    return SingleKeyWriter<K, V, UT>(topic, key, name, config);
}

template<typename K, typename V, typename UT> MultiKeyWriter<K, V, UT>
makeMultiKeyWriter(const Topic<K, V, UT>& topic,
                   const std::vector<typename Topic<K, V, UT>::KeyType>& keys,
                   const std::string& name = std::string(),
                   const WriterConfig& config = WriterConfig()) noexcept
{
    return MultiKeyWriter<K, V, UT>(topic, keys, name, config);
}

template<typename K, typename V, typename UT> MultiKeyWriter<K, V, UT>
makeAnyKeyWriter(const Topic<K, V, UT>& topic,
                 const std::string& name = std::string(),
                 const WriterConfig& config = WriterConfig()) noexcept
{
    return MultiKeyWriter<K, V, UT>(topic, {}, name, config);
}

}

//
// Public template based API implementation
//

namespace DataStorm
{

//
// Sample template implementation
//
template<typename Key, typename Value, typename UpdateTag> SampleEvent
Sample<Key, Value, UpdateTag>::getEvent() const noexcept
{
    return _impl->event;
}

template<typename Key, typename Value, typename UpdateTag> const Key&
Sample<Key, Value, UpdateTag>::getKey() const noexcept
{
    return _impl->getKey();
}

template<typename Key, typename Value, typename UpdateTag> const Value&
Sample<Key, Value, UpdateTag>::getValue() const noexcept
{
    return _impl->getValue();
}

template<typename Key, typename Value, typename UpdateTag> UpdateTag
Sample<Key, Value, UpdateTag>::getUpdateTag() const noexcept
{
    return _impl->getTag();
}

template<typename Key, typename Value, typename UpdateTag> std::chrono::time_point<std::chrono::system_clock>
Sample<Key, Value, UpdateTag>::getTimeStamp() const noexcept
{
    return _impl->timestamp;
}

template<typename Key, typename Value, typename UpdateTag> std::string
Sample<Key, Value, UpdateTag>::getOrigin() const noexcept
{
    return _impl->origin;
}

template<typename Key, typename Value, typename UpdateTag> std::string
Sample<Key, Value, UpdateTag>::getSession() const noexcept
{
    return _impl->session;
}

template<typename Key, typename Value, typename UpdateTag> Sample<Key, Value, UpdateTag>::Sample(
    const std::shared_ptr<DataStormI::Sample>& impl) noexcept :
    _impl(std::static_pointer_cast<DataStormI::SampleT<Key, Value, UpdateTag>>(impl))
{
}

//
// Reader template implementation
//
template<typename Key, typename Value, typename UpdateTag>
Reader<Key, Value, UpdateTag>::Reader(Reader<Key, Value, UpdateTag>&& reader) noexcept : _impl(std::move(reader._impl))
{
}

template<typename Key, typename Value, typename UpdateTag>
Reader<Key, Value, UpdateTag>::~Reader()
{
    if(_impl)
    {
        _impl->destroy();
    }
}

template<typename Key, typename Value, typename UpdateTag> Reader<Key, Value, UpdateTag>&
Reader<Key, Value, UpdateTag>::operator=(Reader&& reader) noexcept
{
    _impl = std::move(reader._impl);
    return *this;
}

template<typename Key, typename Value, typename UpdateTag> bool
Reader<Key, Value, UpdateTag>::hasWriters() const noexcept
{
    return _impl->hasWriters();
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::waitForWriters(unsigned int count) const
{
    _impl->waitForWriters(count);
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::waitForNoWriters() const
{
    _impl->waitForWriters(-1);
}

template<typename Key, typename Value, typename UpdateTag> std::vector<std::string>
Reader<Key, Value, UpdateTag>::getConnectedWriters() const noexcept
{
    return _impl->getConnectedElements();
}

template<typename Key, typename Value, typename UpdateTag> std::vector<Key>
Reader<Key, Value, UpdateTag>::getConnectedKeys() const noexcept
{
    std::vector<Key> keys;
    auto connectedKeys = _impl->getConnectedKeys();
    keys.reserve(connectedKeys.size());
    for(const auto& k : connectedKeys)
    {
        keys.push_back(std::static_pointer_cast<DataStormI::KeyT<Key>>(k)->get());
    }
    return keys;
}

template<typename Key, typename Value, typename UpdateTag> std::vector<Sample<Key, Value, UpdateTag>>
Reader<Key, Value, UpdateTag>::getAllUnread() noexcept
{
    auto unread = _impl->getAllUnread();
    std::vector<Sample<Key, Value, UpdateTag>> samples;
    samples.reserve(unread.size());
    for(auto sample : unread)
    {
        samples.emplace_back(sample);
    }
    return samples;
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::waitForUnread(unsigned int count) const
{
    _impl->waitForUnread(count);
}

template<typename Key, typename Value, typename UpdateTag> bool
Reader<Key, Value, UpdateTag>::hasUnread() const noexcept
{
    return _impl->hasUnread();
}

template<typename Key, typename Value, typename UpdateTag> Sample<Key, Value, UpdateTag>
Reader<Key, Value, UpdateTag>::getNextUnread()
{
    return Sample<Key, Value, UpdateTag>(_impl->getNextUnread());
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::onConnectedKeys(std::function<void(std::vector<Key>)> init,
                                               std::function<void(CallbackReason, Key)> update) noexcept
{
    _impl->onConnectedKeys(init ? [init](std::vector<std::shared_ptr<DataStormI::Key>> connectedKeys)
    {
        std::vector<Key> keys;
        keys.reserve(connectedKeys.size());
        for(const auto& k : connectedKeys)
        {
            keys.push_back(std::static_pointer_cast<DataStormI::KeyT<Key>>(k)->get());
        }
        init(move(keys));
    } : std::function<void(std::vector<std::shared_ptr<DataStormI::Key>>)>(),
    update ? [update](CallbackReason action, std::shared_ptr<DataStormI::Key> key)
    {
        update(action, std::static_pointer_cast<DataStormI::KeyT<Key>>(key)->get());
    } : std::function<void(CallbackReason, std::shared_ptr<DataStormI::Key>)>());
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::onConnectedWriters(std::function<void(std::vector<std::string>)> init,
                                                  std::function<void(CallbackReason, std::string)> update) noexcept
{
    _impl->onConnectedElements(init, update);
}

template<typename Key, typename Value, typename UpdateTag> void
Reader<Key, Value, UpdateTag>::onSamples(std::function<void(std::vector<Sample<Key, Value, UpdateTag>>)> init,
                                         std::function<void(Sample<Key, Value, UpdateTag>)> update) noexcept
{
    auto communicator = _impl->getCommunicator();
    _impl->onSamples(init ? [communicator, init](const std::vector<std::shared_ptr<DataStormI::Sample>>& samplesI)
    {
        std::vector<Sample<Key, Value, UpdateTag>> samples;
        samples.reserve(samplesI.size());
        for(const auto& s : samplesI)
        {
            samples.emplace_back(s);
        }
        init(move(samples));
    } : std::function<void(const std::vector<std::shared_ptr<DataStormI::Sample>>&)>(),
    update ? [communicator, update](const std::shared_ptr<DataStormI::Sample>& sampleI)
    {
        update(sampleI);
    } : std::function<void(const std::shared_ptr<DataStormI::Sample>&)>());
}

template<typename Key, typename Value, typename UpdateTag>
SingleKeyReader<Key, Value, UpdateTag>::SingleKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                        const Key& key,
                                                        const std::string& name,
                                                        const ReaderConfig& config) noexcept :
    Reader<Key, Value, UpdateTag>(topic.getReader()->create({ topic._keyFactory->create(key) }, name, config))
{
}

template<typename Key, typename Value, typename UpdateTag> template<typename SFC>
SingleKeyReader<Key, Value, UpdateTag>::SingleKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                        const Key& key,
                                                        const Filter<SFC>& sampleFilter,
                                                        const std::string& name,
                                                        const ReaderConfig& config) noexcept :
    Reader<Key, Value, UpdateTag>(topic.getReader()->create({ topic._keyFactory->create(key) },
                                                            name,
                                                            config,
                                                            sampleFilter.name,
                                                            DataStormI::EncoderT<SFC>::encode(topic.getCommunicator(),
                                                                                              sampleFilter.criteria)))
{
}

template<typename Key, typename Value, typename UpdateTag>
SingleKeyReader<Key, Value, UpdateTag>::SingleKeyReader(SingleKeyReader<Key, Value, UpdateTag>&& reader) noexcept :
    Reader<Key, Value, UpdateTag>(std::move(reader))
{
}

template<typename Key, typename Value, typename UpdateTag> SingleKeyReader<Key, Value, UpdateTag>&
SingleKeyReader<Key, Value, UpdateTag>::operator=(SingleKeyReader&& reader) noexcept
{
    Reader<Key, Value, UpdateTag>::operator=(std::move(reader));
    return *this;
}

template<typename Key, typename Value, typename UpdateTag>
MultiKeyReader<Key, Value, UpdateTag>::MultiKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                      const std::vector<Key>& keys,
                                                      const std::string& name,
                                                      const ReaderConfig& config) noexcept :
    Reader<Key, Value, UpdateTag>(topic.getReader()->create(topic._keyFactory->create(keys),
                                                            name,
                                                            config))
{
}

template<typename Key, typename Value, typename UpdateTag> template<typename SFC>
MultiKeyReader<Key, Value, UpdateTag>::MultiKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                      const std::vector<Key>& keys,
                                                      const Filter<SFC>& sampleFilter,
                                                      const std::string& name,
                                                      const ReaderConfig& config) noexcept :
    Reader<Key, Value, UpdateTag>(topic.getReader()->create(topic._keyFactory->create(keys),
                                                            name,
                                                            config,
                                                            sampleFilter.name,
                                                            Encoder<SFC>::encode(topic.getCommunicator(),
                                                                                 sampleFilter.criteria)))
{
}

template<typename Key, typename Value, typename UpdateTag>
MultiKeyReader<Key, Value, UpdateTag>::MultiKeyReader(MultiKeyReader<Key, Value, UpdateTag>&& reader) noexcept :
    Reader<Key, Value, UpdateTag>(std::move(reader))
{
}

template<typename Key, typename Value, typename UpdateTag>
MultiKeyReader<Key, Value, UpdateTag>&
MultiKeyReader<Key, Value, UpdateTag>::operator=(MultiKeyReader&& reader) noexcept
{
    Reader<Key, Value, UpdateTag>::operator=(std::move(reader));
    return *this;
}

template<typename Key, typename Value, typename UpdateTag> template<typename KFC>
FilteredKeyReader<Key, Value, UpdateTag>::FilteredKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                            const Filter<KFC>& filter,
                                                            const std::string& name,
                                                            const ReaderConfig& config) :
    Reader<Key, Value, UpdateTag>(topic.getReader()->createFiltered(topic._keyFilterFactories->create(filter.name,
                                                                                                      filter.criteria),
                                                                    name,
                                                                    config))
{
}

template<typename Key, typename Value, typename UpdateTag> template<typename KFC, typename SFC>
FilteredKeyReader<Key, Value, UpdateTag>::FilteredKeyReader(const Topic<Key, Value, UpdateTag>& topic,
                                                            const Filter<KFC>& keyFilter,
                                                            const Filter<SFC>& sampleFilter,
                                                            const std::string& name,
                                                            const ReaderConfig& config) :
    Reader<Key, Value, UpdateTag>(topic.getReader()->createFiltered(topic._keyFilterFactories->create(keyFilter.name,
                                                                                                      keyFilter.criteria),
                                                                    name,
                                                                    config,
                                                                    sampleFilter.name,
                                                                    Encoder<SFC>::encode(topic.getCommunicator(),
                                                                                         sampleFilter.criteria)))
{
}

template<typename Key, typename Value, typename UpdateTag>
FilteredKeyReader<Key, Value, UpdateTag>::FilteredKeyReader(FilteredKeyReader<Key, Value, UpdateTag>&& reader) noexcept :
    Reader<Key, Value, UpdateTag>(std::move(reader))
{
}

template<typename Key, typename Value, typename UpdateTag> FilteredKeyReader<Key, Value, UpdateTag>&
FilteredKeyReader<Key, Value, UpdateTag>::operator=(FilteredKeyReader&& reader) noexcept
{
    Reader<Key, Value, UpdateTag>::operator=(std::move(reader));
    return *this;
}

//
// Writer template implementation
//
template<typename Key, typename Value, typename UpdateTag>
Writer<Key, Value, UpdateTag>::Writer(Writer&& writer) noexcept : _impl(std::move(writer._impl))
{
}

template<typename Key, typename Value, typename UpdateTag> Writer<Key, Value, UpdateTag>&
Writer<Key, Value, UpdateTag>::operator=(Writer&& writer) noexcept
{
    _impl = std::move(writer._impl);
    return *this;
}

template<typename Key, typename Value, typename UpdateTag>
Writer<Key, Value, UpdateTag>::~Writer()
{
    if(_impl)
    {
        _impl->destroy();
    }
}

template<typename Key, typename Value, typename UpdateTag> bool
Writer<Key, Value, UpdateTag>::hasReaders() const noexcept
{
    return _impl->hasReaders();
}

template<typename Key, typename Value, typename UpdateTag> void
Writer<Key, Value, UpdateTag>::waitForReaders(unsigned int count) const
{
    return _impl->waitForReaders(count);
}

template<typename Key, typename Value, typename UpdateTag> void
Writer<Key, Value, UpdateTag>::waitForNoReaders() const
{
    return _impl->waitForReaders(-1);
}

template<typename Key, typename Value, typename UpdateTag> std::vector<std::string>
Writer<Key, Value, UpdateTag>::getConnectedReaders() const noexcept
{
    return _impl->getConnectedElements();
}

template<typename Key, typename Value, typename UpdateTag> std::vector<Key>
Writer<Key, Value, UpdateTag>::getConnectedKeys() const noexcept
{
    std::vector<Key> keys;
    auto connectedKeys = _impl->getConnectedKeys();
    keys.reserve(connectedKeys.size());
    for(const auto& k : connectedKeys)
    {
        keys.push_back(std::static_pointer_cast<DataStormI::KeyT<Key>>(k)->get());
    }
    return keys;
}

template<typename Key, typename Value, typename UpdateTag> Sample<Key, Value, UpdateTag>
Writer<Key, Value, UpdateTag>::getLast()
{
    auto sample = _impl->getLast();
    if(!sample)
    {
        throw std::logic_error("no sample");
    }
    return Sample<Key, Value, UpdateTag>(sample);
}

template<typename Key, typename Value, typename UpdateTag> std::vector<Sample<Key, Value, UpdateTag>>
Writer<Key, Value, UpdateTag>::getAll() noexcept
{
    auto all = _impl->getAll();
    std::vector<Sample<Key, Value, UpdateTag>> samples;
    samples.reserve(all.size());
    for(auto sample : all)
    {
        samples.emplace_back(sample);
    }
    return samples;
}

template<typename Key, typename Value, typename UpdateTag> void
Writer<Key, Value, UpdateTag>::onConnectedKeys(std::function<void(std::vector<Key>)> init,
                                               std::function<void(CallbackReason, Key)> update) noexcept
{
    _impl->onConnectedKeys(init ? [init](std::vector<std::shared_ptr<DataStormI::Key>> connectedKeys)
    {
        std::vector<Key> keys;
        keys.reserve(connectedKeys.size());
        for(const auto& k : connectedKeys)
        {
            keys.push_back(std::static_pointer_cast<DataStormI::KeyT<Key>>(k)->get());
        }
        init(move(keys));
    } : std::function<void(std::vector<std::shared_ptr<DataStormI::Key>>)>(),
    update ? [update](CallbackReason action, std::shared_ptr<DataStormI::Key> key)
    {
        update(action, std::static_pointer_cast<DataStormI::KeyT<Key>>(key)->get());
    } : std::function<void(CallbackReason, std::shared_ptr<DataStormI::Key>)>());
}

template<typename Key, typename Value, typename UpdateTag> void
Writer<Key, Value, UpdateTag>::onConnectedReaders(std::function<void(std::vector<std::string>)> init,
                                                  std::function<void(CallbackReason, std::string)> update) noexcept
{
    _impl->onConnectedElements(init, update);
}

template<typename Key, typename Value, typename UpdateTag>
SingleKeyWriter<Key, Value, UpdateTag>::SingleKeyWriter(const Topic<Key, Value, UpdateTag>& topic,
                                                        const Key& key,
                                                        const std::string& name,
                                                        const WriterConfig& config) noexcept :
    Writer<Key, Value, UpdateTag>(topic.getWriter()->create({ topic._keyFactory->create(key) },
                                                            name,
                                                            config)),
    _tagFactory(topic._tagFactory)
{
}

template<typename Key, typename Value, typename UpdateTag>
SingleKeyWriter<Key, Value, UpdateTag>::SingleKeyWriter(SingleKeyWriter<Key, Value, UpdateTag>&& writer) noexcept :
    Writer<Key, Value, UpdateTag>(std::move(writer)),
    _tagFactory(std::move(writer._tagFactory))
{
}

template<typename Key, typename Value, typename UpdateTag> SingleKeyWriter<Key, Value, UpdateTag>&
SingleKeyWriter<Key, Value, UpdateTag>::operator=(SingleKeyWriter&& writer) noexcept
{
    Writer<Key, Value, UpdateTag>::operator=(std::move(writer));
    return *this;
}

template<typename Key, typename Value, typename UpdateTag> void
SingleKeyWriter<Key, Value, UpdateTag>::add(const Value& value) noexcept
{
   Writer<Key, Value, UpdateTag>::_impl->publish(nullptr,
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Add, value));
}

template<typename Key, typename Value, typename UpdateTag> void
SingleKeyWriter<Key, Value, UpdateTag>::update(const Value& value) noexcept
{
    Writer<Key, Value, UpdateTag>::_impl->publish(nullptr,
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Update, value));
}

template<typename Key, typename Value, typename UpdateTag>
template<typename UpdateValue> std::function<void(const UpdateValue&)>
SingleKeyWriter<Key, Value, UpdateTag>::partialUpdate(const UpdateTag& tag) noexcept
{
    auto impl = Writer<Key, Value, UpdateTag>::_impl;
    auto updateTag = _tagFactory->create(tag);
    return [impl, updateTag](const UpdateValue& value) {
        auto encoded = Encoder<UpdateValue>::encode(impl->getCommunicator(), value);
        impl->publish(nullptr, std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(encoded, updateTag));
    };
}

template<typename Key, typename Value, typename UpdateTag> void
SingleKeyWriter<Key, Value, UpdateTag>::remove() noexcept
{
    Writer<Key, Value, UpdateTag>::_impl->publish(nullptr,
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Remove));
}

template<typename Key, typename Value, typename UpdateTag>
MultiKeyWriter<Key, Value, UpdateTag>::MultiKeyWriter(const Topic<Key, Value, UpdateTag>& topic,
                                                      const std::vector<Key>& keys,
                                                      const std::string& name,
                                                      const WriterConfig& config) noexcept :
    Writer<Key, Value, UpdateTag>(topic.getWriter()->create(topic._keyFactory->create(keys),
                                                            name,
                                                            config)),
    _keyFactory(topic._keyFactory),
    _tagFactory(topic._tagFactory)
{
}

template<typename Key, typename Value, typename UpdateTag>
MultiKeyWriter<Key, Value, UpdateTag>::MultiKeyWriter(MultiKeyWriter<Key, Value, UpdateTag>&& writer) noexcept :
    Writer<Key, Value, UpdateTag>(std::move(writer)),
    _keyFactory(std::move(writer._keyFactory)),
    _tagFactory(std::move(writer._tagFactory))
{
}

template<typename Key, typename Value, typename UpdateTag> MultiKeyWriter<Key, Value, UpdateTag>&
MultiKeyWriter<Key, Value, UpdateTag>::operator=(MultiKeyWriter&& writer) noexcept
{
    Writer<Key, Value, UpdateTag>::operator=(std::move(writer));
    return *this;
}

template<typename Key, typename Value, typename UpdateTag> void
MultiKeyWriter<Key, Value, UpdateTag>::add(const Key& key, const Value& value) noexcept
{
    Writer<Key, Value, UpdateTag>::_impl->publish(_keyFactory->create(key),
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Add, value));
}

template<typename Key, typename Value, typename UpdateTag> void
MultiKeyWriter<Key, Value, UpdateTag>::update(const Key& key, const Value& value) noexcept
{
    Writer<Key, Value, UpdateTag>::_impl->publish(_keyFactory->create(key),
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Update, value));
}

template<typename Key, typename Value, typename UpdateTag>
template<typename UpdateValue> std::function<void(const Key&, const UpdateValue&)>
MultiKeyWriter<Key, Value, UpdateTag>::partialUpdate(const UpdateTag& tag) noexcept
{
    auto impl = Writer<Key, Value, UpdateTag>::_impl;
    auto updateTag = _tagFactory->create(tag);
    auto keyFactory = _keyFactory;
    return [impl, updateTag, keyFactory](const Key& key, const UpdateValue& value) {
        auto encoded = Encoder<UpdateValue>::encode(impl->getCommunicator(), value);
        impl->publish(keyFactory->create(key),
                      std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(encoded, updateTag));
    };
}

template<typename Key, typename Value, typename UpdateTag> void
MultiKeyWriter<Key, Value, UpdateTag>::remove(const Key& key) noexcept
{
    Writer<Key, Value, UpdateTag>::_impl->publish(_keyFactory->create(key),
        std::make_shared<DataStormI::SampleT<Key, Value, UpdateTag>>(SampleEvent::Remove));
}

#if !defined(__clang__) && defined(__GNUC__) && ((__GNUC__* 100) + __GNUC_MINOR__) < 490

#include <regex.h>

class RegExp
{
public:

    RegExp(const std::string& criteria)
    {
        if(regcomp(&_expr, criteria.c_str(), REG_EXTENDED) != 0)
        {
            throw std::invalid_argument(criteria);
        }
    }

    ~RegExp()
    {
        regfree(&_expr);
    }

    bool match(const std::string& value) const
    {
        return regexec(&_expr, value.c_str(), 0, 0, 0) == 0;
    }

private:

    regex_t _expr;
};

#endif

template<typename Value> std::function<std::function<bool (const Value&)> (const std::string&)>
makeRegexFilter() noexcept
{
    return [](const std::string& criteria)
    {
#if !defined(__clang__) && defined(__GNUC__) && ((__GNUC__* 100) + __GNUC_MINOR__) < 490
        auto expr = std::make_shared<RegExp>(criteria);
        return [expr](const Value& value)
        {
            std::ostringstream os;
            os << value;
            return expr->match(os.str());
        };
#else
        std::regex expr(criteria);
        return [expr](const Value& value)
        {
            std::ostringstream os;
            os << value;
            return std::regex_match(os.str(), expr);
        };
#endif
    };
}

template<typename Key, typename Value, typename UpdateTag>
std::function<std::function<bool (const Sample<Key, Value, UpdateTag>&)> (const std::vector<SampleEvent>&)>
makeSampleEventFilter(const Topic<Key, Value, UpdateTag>& topic) noexcept
{
    return [](const std::vector<SampleEvent>& criteria)
    {
        return [criteria](const Sample<Key, Value, UpdateTag>& sample)
        {
            return std::find(criteria.begin(), criteria.end(), sample.getEvent()) != criteria.end();
        };
    };
}

template<typename T, typename V, typename Enabler=void>
struct RegexFilter
{
    template<typename F> static void
    add(F factory)
    {
    }
};

template<typename T, typename V>
struct RegexFilter<T, V, typename std::enable_if<DataStormI::is_streamable<V>::value>::type>
{
    template<typename F> static void
    add(F factory)
    {
        factory->set("_regex", makeRegexFilter<T>()); // Only set the _regex filter if the value is streamable
    }
};

//
// Topic template implementation
//
template<typename Key, typename Value, typename UpdateTag>
Topic<Key, Value, UpdateTag>::Topic(const Node& node, const std::string& name) noexcept :
    _name(name),
    _topicFactory(node._factory),
    _keyFactory(DataStormI::KeyFactoryT<Key>::createFactory()),
    _tagFactory(DataStormI::TagFactoryT<UpdateTag>::createFactory()),
    _keyFilterFactories(DataStormI::FilterManagerT<DataStormI::KeyT<Key>>::create()),
    _sampleFilterFactories(DataStormI::FilterManagerT<DataStormI::SampleT<Key, Value, UpdateTag>>::create())
{
    RegexFilter<Key, Key>::add(_keyFilterFactories);
    RegexFilter<Sample<Key, Value, UpdateTag>, Value>::add(_sampleFilterFactories);
    _sampleFilterFactories->set("_event", makeSampleEventFilter(*this));
}

template<typename Key, typename Value, typename UpdateTag>
Topic<Key, Value, UpdateTag>::Topic(Topic<Key, Value, UpdateTag>&& topic) noexcept :
    _name(std::move(topic._name)),
    _topicFactory(std::move(topic._topicFactory)),
    _keyFactory(std::move(topic._keyFactory)),
    _tagFactory(std::move(topic._tagFactory)),
    _keyFilterFactories(std::move(topic._keyFilterFactories)),
    _sampleFilterFactories(std::move(topic._sampleFilterFactories)),
    _reader(std::move(topic._reader)),
    _writer(std::move(topic._writer)),
    _updaters(std::move(topic._updaters))
{
}

template<typename Key, typename Value, typename UpdateTag>
Topic<Key, Value, UpdateTag>::~Topic()
{
    std::lock_guard<std::mutex> lock(_mutex);
    if(_reader)
    {
        _reader->destroy();
    }
    if(_writer)
    {
        _writer->destroy();
    }
}

template<typename Key, typename Value, typename UpdateTag> Topic<Key, Value, UpdateTag>&
Topic<Key, Value, UpdateTag>::operator=(Topic<Key, Value, UpdateTag>&& topic) noexcept
{
    _name = std::move(topic._name);
    _topicFactory = std::move(topic._topicFactory);
    _keyFactory = std::move(topic._keyFactory);
    _tagFactory = std::move(topic._tagFactory);
    _keyFilterFactories = std::move(topic._keyFilterFactories);
    _sampleFilterFactories = std::move(topic._sampleFilterFactories);
    _reader = std::move(topic._reader);
    _writer = std::move(topic._writer);
    _updaters = std::move(topic._updaters);
    return *this;
}

template<typename Key, typename Value, typename UpdateTag> bool
Topic<Key, Value, UpdateTag>::hasWriters() const noexcept
{
    return getReader()->hasWriters();
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::waitForWriters(unsigned int count) const
{
    getReader()->waitForWriters(count);
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::waitForNoWriters() const
{
    getReader()->waitForWriters(-1);
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::setReaderDefaultConfig(const ReaderConfig& config) noexcept
{
    getReader()->setDefaultConfig(config);
}

template<typename Key, typename Value, typename UpdateTag> bool
Topic<Key, Value, UpdateTag>::hasReaders() const noexcept
{
    return getWriter()->hasReaders();
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::waitForReaders(unsigned int count) const
{
    getWriter()->waitForReaders(count);
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::waitForNoReaders() const
{
    getWriter()->waitForReaders(-1);
}

template<typename Key, typename Value, typename UpdateTag> void
Topic<Key, Value, UpdateTag>::setWriterDefaultConfig(const WriterConfig& config) noexcept
{
    getWriter()->setDefaultConfig(config);
}

template<typename Key, typename Value, typename UpdateTag> template<typename UpdateValue> void
Topic<Key, Value, UpdateTag>::setUpdater(const UpdateTag& tag, std::function<void (Value&, UpdateValue)> updater) noexcept
{
    std::lock_guard<std::mutex> lock(_mutex);
    auto tagI = _tagFactory->create(std::move(tag));
    auto updaterImpl = updater ? [updater](const std::shared_ptr<DataStormI::Sample>& previous,
                                           const std::shared_ptr<DataStormI::Sample>& next,
                                           const std::shared_ptr<Ice::Communicator>& communicator)
    {
        Value value;
        if(previous)
        {
            value = Cloner<Value>::clone(
                std::static_pointer_cast<DataStormI::SampleT<Key, Value, UpdateTag>>(previous)->getValue());
        }
        updater(value, Decoder<UpdateValue>::decode(communicator, next->getEncodedValue()));
        std::static_pointer_cast<DataStormI::SampleT<Key, Value, UpdateTag>>(next)->setValue(std::move(value));
    } : std::function<void(const std::shared_ptr<DataStormI::Sample>&,
                           const std::shared_ptr<DataStormI::Sample>&,
                           const std::shared_ptr<Ice::Communicator>&)>();

    if(_reader && !_writer)
    {
        _reader->setUpdater(tagI, updaterImpl);
    }
    else if(_writer && !_reader)
    {
        _writer->setUpdater(tagI, updaterImpl);
    }
    else if(_reader && _writer)
    {
        _reader->setUpdater(tagI, updaterImpl);
        _writer->setUpdater(tagI, updaterImpl);
    }
    else
    {
        _updaters[tagI] = updaterImpl;
    }
}

template<typename Key, typename Value, typename UpdateTag> template<typename Criteria> void
Topic<Key, Value, UpdateTag>::setKeyFilter(const std::string& name,
    std::function<std::function<bool (const Key&)>(const Criteria&)> factory) noexcept
{
    std::lock_guard<std::mutex> lock(_mutex);
    _keyFilterFactories->set(name, factory);
}

template<typename Key, typename Value, typename UpdateTag> template<typename Criteria> void
Topic<Key, Value, UpdateTag>::setSampleFilter(const std::string& name,
    std::function<std::function<bool (const Sample<Key, Value, UpdateTag>&)>(const Criteria&)> factory) noexcept
{
    std::lock_guard<std::mutex> lock(_mutex);
    _sampleFilterFactories->set(name, factory);
}

template<typename Key, typename Value, typename UpdateTag> std::shared_ptr<DataStormI::TopicReader>
Topic<Key, Value, UpdateTag>::getReader() const noexcept
{
    std::lock_guard<std::mutex> lock(_mutex);
    if(!_reader)
    {
        auto sampleFactory = std::make_shared<DataStormI::SampleFactoryT<Key, Value, UpdateTag>>();
        _reader = _topicFactory->createTopicReader(_name, _keyFactory, _tagFactory, sampleFactory, _keyFilterFactories,
                                                   _sampleFilterFactories);
        _reader->setUpdaters(_writer ? _writer->getUpdaters() : _updaters);
        _updaters.clear();
    }
    return _reader;
}

template<typename Key, typename Value, typename UpdateTag> std::shared_ptr<DataStormI::TopicWriter>
Topic<Key, Value, UpdateTag>::getWriter() const noexcept
{
    std::lock_guard<std::mutex> lock(_mutex);
    if(!_writer)
    {
        _writer = _topicFactory->createTopicWriter(_name, _keyFactory, _tagFactory, nullptr, _keyFilterFactories,
                                                   _sampleFilterFactories);
        _writer->setUpdaters(_reader ? _reader->getUpdaters() : _updaters);
        _updaters.clear();
    }
    return _writer;
}

template<typename Key, typename Value, typename UpdateTag> std::shared_ptr<Ice::Communicator>
Topic<Key, Value, UpdateTag>::getCommunicator() const noexcept
{
    return _topicFactory->getCommunicator();
}

}