SpreaditRouting.h

Go to the documentation of this file.

/*
 Copyright (2016) Michael Baer, Daniel Bonrath, All rights reserved.

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
 */

#ifndef __M2ETIS_PUBSUB_ROUTING_SPREADITROUTING_H__
#define __M2ETIS_PUBSUB_ROUTING_SPREADITROUTING_H__

#include <algorithm>
#include <string>
#include <vector>

#include "m2etis/message/MessageType.h"
#include "m2etis/message/info/routing/SpreadItRoutingInfo.h"

#include "m2etis/pubsub/PubSubSystemEnvironment.h"
#include "m2etis/pubsub/routing/BaseRouting.h"

#include "boost/date_time/posix_time/posix_time_types.hpp"

namespace m2etis {
namespace pubsub {
namespace routing {

        template<class NetworkType>
        class SpreaditRouting : public BaseRouting<NetworkType> {
        public:
                typedef message::SpreadItRoutingInfo<NetworkType> RoutingInfoType;
                typedef std::pair<uint64_t, typename NetworkType::Key> TimePair;
                typedef std::vector<typename NetworkType::Key> KeyList;
                typedef std::vector<TimePair> TimeList;

                // [EOC] The next variables are more or less generic to all
                // distribution (algo) policies and describe their behavoir!

                // whether or not to register for subscribe messages in forward
                static const bool register_forward_subscribe = false;

                // whether or not to register for subscribe messages in deliver
                static const bool register_deliver_subscribe = true;

                // whether or not to register for unsubscribe messages in forward
                static const bool register_forward_unsubscribe = false;

                // whether or not to register for unsubscribe messages in deliver
                static const bool register_deliver_unsubscribe = true;
                // [EOC]

                // whether or not to activate an automatic periodic resent subscribtion
                static const bool periodicSubscribtion = true;

                // the time_duration between resent periodic subscribtions
                const uint64_t periodic_;

                // the time_duration to decide if a subscriber is purged form the list
                const uint64_t purge_distance_;

                PubSubSystemEnvironment * _pssi;

                // inform listener, whether a subscriber has been purged, e.g. by timeout
                // currently only needed for filter strategies
                boost::function<void(const typename NetworkType::Key)> removed_subscriber_eventlistener_;

                uint64_t purgeID_;

                SpreaditRouting(unsigned int topic_name, PubSubSystemEnvironment * pssi, const typename NetworkType::Key & root) :
                        BaseRouting<NetworkType>(topic_name, pssi)
                        , periodic_(spreadit::RESUBSCRIPTION_INTERVAL), purge_distance_(spreadit::PURGE_DISTANCE), _pssi(pssi)
                        , removed_subscriber_eventlistener_()
                        , topic_name_(topic_name), self_(), subscribed_(false), subscriber_()
                        , purging_(true), _root(root) {
                        purgeID_ = pssi->scheduler_.runRepeated(purge_distance_, boost::bind(&SpreaditRouting::purgeList, this), 0);
                }

                virtual ~SpreaditRouting() {
                        purging_ = false;
                        _pssi->scheduler_.stop(purgeID_);
                }

                void setSelf(const typename NetworkType::Key & self) {
                        self_ = self;
                }

        void setUnsubscriptionListener(const boost::function<void(const typename NetworkType::Key)> & listener) {
            removed_subscriber_eventlistener_ = listener;
        }

                // set the self subscription status
                void selfSubscribed(const bool b) {
                        subscribed_ = b;
                }

                // get the self subscribtion status
                bool selfSubscribed() const {
                        return subscribed_;
                }

                void configureRoutingInfo(message::ActionType & msgType, typename message::RoutingInfo<NetworkType>::Ptr routingInfo, typename NetworkType::Key &) {
                        typename RoutingInfoType::Ptr rInfo = cast(routingInfo);
                        switch (msgType) {
                        case message::SUBSCRIBE: {
                                msgType = message::SUBSCRIBE;
                                break;
                        }
                        case message::UNSUBSCRIBE: {
                                msgType = message::UNSUBSCRIBE;
                                if (!parent_vector.empty()) {
                                        rInfo->node_adress = parent_vector.at(0).second;
                                }
                                break;
                        }
                        case message::PUBLISH: {
                                msgType = message::PUBLISH;
                                break;
                        }
                        case message::NOTIFY: {
                                msgType = message::NOTIFY;
                                break;
                        }
                        case message::CONTROL: {
                                msgType = message::CONTROL;
                                break;
                        }
                        default:
                                M2ETIS_THROW_FAILURE("Spreadit Routing", "configureRoutingInfo called with wrong action type", -1);
                        }
                        return;
                }

                KeyList getTargetNodes(const message::ActionType mtype, typename message::RoutingInfo<NetworkType>::Ptr routingInfo, typename NetworkType::Key & receiver) const {
                        typename RoutingInfoType::Ptr rInfo = cast(routingInfo);
                        /*
                         * All messages must be sent to root.
                         * Even if it's a publishmessage and root is subscribed, too.
                         * In deliver::publish, root will take care to deliver the message and spread it to the subscribers.
                         */
                        KeyList m;
                        if (mtype == message::SUBSCRIBE) {
                                // I've been redirected to a new parent
                                if (rInfo->action == message::RoutingInfo<NetworkType>::RoutingType::REDIRECT) {
                                        m.push_back(rInfo->node_adress);
                                } else if (!parent_vector.empty()) { // if I already have a parent, send subscribe to him - works like a heart beat
                                        m.push_back(parent_vector.begin()->second);
                                } else { // subscription process starts from the root
                                        if (self_ == _root) {
                                                return m;
                                        } else {
                                                m.push_back(_root);
                                        }
                                }
                        } else if (mtype == message::CONTROL) {
                                // receiver has been saved earlier
                                if (receiver == typename NetworkType::Key()) {
                                        struct T {
                                                 static typename NetworkType::Key get_from_pair(const SpreaditRouting::TimePair & p) {
                                                        return p.second;
                                                 }
                                        };

                                        m.resize(subscriber_.size());
                                        std::transform(subscriber_.begin(), subscriber_.end(), m.begin(), T::get_from_pair);
                                } else {
                                        m.push_back(receiver);
                                }
                        } else if (mtype == message::PUBLISH) {
                                // here we need to send the message always to root (even if we are the root)
                                // because the processing for publish messages is done in deliver
                                m.push_back(_root);
                        } else if (mtype == message::NOTIFY) {
                                // message must be published, that means from root to subscribers
                                // Simply return a list of subscribed nodes

                                // That's nicer than for_each and push_back, isn't it?
                                // std::transform is really good!
                                struct T {
                                         static typename NetworkType::Key get_from_pair(const SpreaditRouting::TimePair & p) {
                                                return p.second;
                                         }
                                };

                                m.resize(subscriber_.size());
                                std::transform(subscriber_.begin(), subscriber_.end(), m.begin(), T::get_from_pair);
                        } else if (mtype == message::UNSUBSCRIBE) {
                                // message must be published, that means from root to subscribers
                                // Simply return a list of subscribed nodes

                                // That's nicer than for_each and push_back, isn't it?
                                // std::transform is really good!
                                struct T {
                                         static typename NetworkType::Key get_from_pair(const TimePair & p) {
                                                return p.second;
                                         }
                                };

                                m.resize(subscriber_.size());
                                std::transform(subscriber_.begin(), subscriber_.end(), m.begin(), T::get_from_pair);

                                if (!parent_vector.empty()) {
                                        m.push_back(parent_vector.at(0).second);
                                }
                        } else {
                                M2ETIS_THROW_FAILURE("Spreadit Routing", "unknown action type", -1);
                        }

                        return m;
                }

                bool processSubscribePayload(typename message::RoutingInfo<NetworkType>::Ptr routingInfo, const typename NetworkType::Key & sender, typename NetworkType::Key & receiver, message::ActionType & msgType) override {
                        typename RoutingInfoType::Ptr rInfo = cast(routingInfo);

                        bool is_subscribe_successful = true;
                        bool found = false;
                        typename TimeList::iterator iter;

                        for (iter = subscriber_.begin(); iter != subscriber_.end(); ++iter) {
                                if (iter->second == sender) {
                                        found = true;
                                        break;
                                }
                        }
                        if (found) {
                                // update entry
                                iter->first = _pssi->clock_.getTime();
                                // send ack to my subscriber - heart beat
                                rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::ACK;
                        } else {
                                // check if I have capacity for more children
                                if (bandwidth_capacity > subscriber_.size()) {
                                        // insert entry
                                        subscriber_.push_back(std::make_pair(_pssi->clock_.getTime(), sender));
                                        M2ETIS_LOG_DEBUG(self_.toStr(), " added " << sender.toStr());
                                        rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::ACK;
                                        msgType = message::CONTROL;
                                } else { // redirect to another node
                                        rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::REDIRECT;
                                        rInfo->node_adress = typename NetworkType::Key(subscriber_.at(round_robin_pointer).second);
                                        round_robin_pointer = (round_robin_pointer + 1) % subscriber_.size();
                                        msgType = message::CONTROL;
                                        is_subscribe_successful = false;
                                }
                        }

                        receiver = sender;
                        return is_subscribe_successful;
                }

                void processUnsubscribePayload(typename message::RoutingInfo<NetworkType>::Ptr routingInfo, const typename NetworkType::Key & sender, typename NetworkType::Key & receiver, message::ActionType & msgType) {
                        typename RoutingInfoType::Ptr rInfo = cast(routingInfo);

                        struct T {
                                static bool test(const typename NetworkType::Key & send, const TimePair & paar) {
                                        return paar.second == send;
                                }
                        };

                        subscriber_.erase(std::remove_if(subscriber_.begin(), subscriber_.end(), boost::bind(T::test, sender, _1)), subscriber_.end());

                        // if it was my parent, he has sent a redirect address
                        if (!parent_vector.empty()) {
                                if (sender == parent_vector.at(0).second) {
                                        msgType = message::SUBSCRIBE;
                                        receiver = rInfo->node_adress;
                                        rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::REDIRECT;
                                }
                        }
                        rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::STOP;
                }

                void processPublishPayload(typename message::RoutingInfo<NetworkType>::Ptr routingInfo, const typename NetworkType::Key &, typename NetworkType::Key &, message::ActionType & msgType) {
                        msgType = message::NOTIFY;
                }

                void processNotifyPayload(typename message::RoutingInfo<NetworkType>::Ptr routingInfo, const typename NetworkType::Key &, typename NetworkType::Key &, message::ActionType &) {
                        if (subscriber_.empty()) {
                                routingInfo->action = message::RoutingInfo<NetworkType>::RoutingType::STOP;
                        }
                }

                void processControlPayload(typename message::RoutingInfo<NetworkType>::Ptr routingInfo, const typename NetworkType::Key & sender, typename NetworkType::Key & receiver, message::ActionType & msgType) {
                        typename RoutingInfoType::Ptr rInfo = cast(routingInfo);
                        // new parent acknowledges the subscription
                        if (rInfo->action == message::RoutingInfo<NetworkType>::RoutingType::ACK) {
                                if (parent_vector.empty()) {
                                        parent_vector.push_back(std::make_pair(_pssi->clock_.getTime(), sender));
                                } else {
                                        parent_vector.begin()->second = sender;
                                        parent_vector.begin()->first = _pssi->clock_.getTime();
                                }
                                rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::STOP;
                        } else if (rInfo->action == message::RoutingInfo<NetworkType>::RoutingType::REDIRECT) { // parent forwards my subscription request to another node
                                msgType = message::SUBSCRIBE;
                                receiver = rInfo->node_adress;
                                rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::REDIRECT;
                        } else { // nothing to do, stop the workflow
                                rInfo->action = message::RoutingInfo<NetworkType>::RoutingType::STOP;
                        }
                }

        private:
                // The topic name
                const unsigned int topic_name_;
                // The own key of this actual node
                typename NetworkType::Key self_;
                // Is this node subscribed on the topic?
                bool subscribed_;

                unsigned int bandwidth_capacity = spreadit::ALLOWED_CHILDS;
                unsigned int round_robin_pointer = 0;

                // List of all subscribers (TimeStamp and Key)
                TimeList subscriber_;
                // List of all parents (TimeStamp and Key)
                TimeList parent_vector;

                // Control variables for the purging thread
                volatile bool purging_;

                typename NetworkType::Key _root;

                typename RoutingInfoType::Ptr cast(typename message::RoutingInfo<NetworkType>::Ptr ptr) const {
                        typename RoutingInfoType::Ptr ret = boost::dynamic_pointer_cast<RoutingInfoType>(ptr);

                        if (!ret) {
                                M2ETIS_LOG_ERROR("Routing Strategy", "Downcast error of routingInfo");
                        }

                        return ret;
                }

                bool purgeList() {
                        // Helper-Struct to create a specialized method which can be used to generically erase from a vector
                        struct T {
                                static bool test(const uint64_t & p, const uint64_t & jetzt, const TimePair & paar) {
                                        if ((jetzt - paar.first) > p) {
                                                return true;
                                        }
                                        return false;
                                }
                        };

                        if (subscriber_.empty() && parent_vector.empty()) {
                                return purging_;
                        }

                        // lock the list
                        // get the acutal timestamp
                        // and remove every subscriber which hasn't resubscribed within the "purge_distance"
                        uint64_t jetzt = _pssi->clock_.getTime();

             // also: inform unsubscription listeners:
                        auto iter_first_erased_subscriber = std::remove_if(subscriber_.begin(), subscriber_.end(), boost::bind(T::test, purge_distance_, jetzt, _1));
                        // notify listeners about erased subscribers:
                        for (auto iter_subscriber = iter_first_erased_subscriber; iter_subscriber != subscriber_.end(); ++iter_subscriber) {
                                removed_subscriber_eventlistener_(iter_subscriber->second);
                        }

                        subscriber_.erase(iter_first_erased_subscriber, subscriber_.end());

                        return purging_;
                }

                SpreaditRouting & operator=(const SpreaditRouting &);
        };

} /* namespace routing */
} /* namespace pubsub */
} /* namespace m2etis */

#endif /* __M2ETIS_PUBSUB_ROUTING_SPREADITROUTING_H__ */