PubSubSystem.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_PUBSUBSYSTEM_H__
#define __M2ETIS_PUBSUB_PUBSUBSYSTEM_H__

#include "m2etis/util/ExceptionQueue.h"
#include "m2etis/util/SystemParameters.h"

#include "m2etis/message/M2Message.h"

#include "m2etis/pubsub/ChannelEventInterface.h"
#include "m2etis/pubsub/ChannelInterface.h"
#include "m2etis/pubsub/config/ChannelConfiguration.h"
#include "m2etis/pubsub/filter/FilterPredicate.h"

#include "boost/function.hpp"

// Forward Declarations

namespace m2etis {
namespace pubsub {

        template<class EventType> class BasicDeliverCallbackInterface;
        class PubSubSystemEnvironment;

        enum exceptionEvents {
                CONNECTION_FAILED = 0,
                CONNECTION_CLOSED = 1,
                NODEFAILURE = 2,
                UNDEFINED = 3,
                BINDFAILURE = 4
        };

        class M2ETIS_API PubSubSystem {
        public:
                PubSubSystem(const std::string & listenIP, const uint16_t listenPort, const std::string & connectIP, const uint16_t connectPort, const std::vector<std::string> & rootList);

                ~PubSubSystem();

                template<class EventType> BasicChannelInterface<EventType> & subscribe(const ChannelName channel, BasicDeliverCallbackInterface<EventType> & callback);

                // FIXME create predicate interface
                template<class EventType> BasicChannelInterface<EventType> & subscribe(const ChannelName channel, BasicDeliverCallbackInterface<EventType> & callback, boost::shared_ptr<filter::FilterExp<EventType> > predicate);

                template<class EventType> const PubSubSystem & unsubscribe(const ChannelName channel) const;

                template<class EventType> const PubSubSystem & unsubscribe(const ChannelName channel, const boost::shared_ptr<filter::FilterExp<EventType> > predicate) const;

                template<class EventType> const PubSubSystem & publish(const ChannelName channel, const typename message::M2Message<EventType>::Ptr publish_message) const;

                // FIXME: (Daniel) if we kick M2Message, we can kick also these to methods and publish Payload directly
                template<class EventType> typename message::M2Message<EventType>::Ptr createMessage(const ChannelName channel, const EventType & payload);
                template<class EventType> typename message::M2Message<EventType>::Ptr createMessage(const ChannelName channel);

                template<class EventType> std::string getSelf(const ChannelName channel) const;

                inline bool isInitialized() const { return initialized; }

                void registerExceptionCallback(exceptionEvents e, boost::function<void(const std::string &)> _ptr);

        private:
                PubSubSystem(const PubSubSystem &) = delete;
                PubSubSystem & operator=(const PubSubSystem &) = delete;

                bool exceptionLoop();

        public:
                PubSubSystemEnvironment * _pssi;

        private:
                ChannelConfiguration * channels_;
                bool initialized;

                std::vector<std::vector<boost::function<void(const std::string &)>>> _exceptionCallbacks;

                uint64_t exceptionID_;

                bool _running;

#ifdef WITH_SIM
        public:
#endif

                template<class EventType> BasicChannelInterface<EventType> * getChannelHandle(const ChannelName channel) const;
        };

        template<class EventType> std::string PubSubSystem::getSelf(ChannelName channel) const {
                return "";
        }

        template<class EventType> BasicChannelInterface<EventType> * PubSubSystem::getChannelHandle(ChannelName channel) const {
                if (!initialized) {
                        M2ETIS_THROW_API("PubSubSystem", "Invalid call, initialize PubSubSystem first.");
                }
                if (channel >= channels_->count) {
                        M2ETIS_THROW_API("PubSubSystem", std::string("Invalid channel, enum exceeds CHANNEL_COUNT: ") + std::to_string(channel));
                }
                BasicChannelInterface<EventType> * const ret = dynamic_cast<BasicChannelInterface<EventType> * const>(channels_->channels()[channel]);
                if (ret == NULL) {
                        M2ETIS_THROW_API("PubSubSystem", "Invalid channel type cast. Check your channel configuration.");
                }
                return ret;
        }

        template<class EventType> BasicChannelInterface<EventType> & PubSubSystem::subscribe(ChannelName channel, BasicDeliverCallbackInterface<EventType> & callback) {
                const boost::shared_ptr<filter::FilterExp<EventType> > pred = boost::make_shared<filter::TruePredicate<EventType> >();

                return subscribe(channel, callback, pred);
        }

        template<class EventType> BasicChannelInterface<EventType> & PubSubSystem::subscribe(const ChannelName channel, BasicDeliverCallbackInterface<EventType> & callback, const boost::shared_ptr<filter::FilterExp<EventType> > predicate) {
                BasicChannelInterface<EventType> * const ret = getChannelHandle<EventType>(channel);

                ret->subscribe(callback, predicate);
                return *ret;
        }

        template<class EventType> const PubSubSystem & PubSubSystem::unsubscribe(ChannelName channel) const {
                BasicChannelInterface<EventType> * const ret = getChannelHandle<EventType>(channel);
                ret->unsubscribe();
                return *this;
        }


        // for filter strategies: deregistering single filter:
        template<class EventType> const PubSubSystem & PubSubSystem::unsubscribe(const ChannelName channel, const boost::shared_ptr<filter::FilterExp<EventType> > predicate) const {
                BasicChannelInterface<EventType> * const ret = getChannelHandle<EventType>(channel);
                ret->unsubscribe(predicate);
                return *this;
        }

        template<class EventType> const PubSubSystem & PubSubSystem::publish(const ChannelName channel, const typename message::M2Message<EventType>::Ptr publish_message) const {
                BasicChannelInterface<EventType> * const ret = getChannelHandle<EventType>(channel);
                ret->publish(publish_message);
                return *this;
        }

        template<class EventType> typename message::M2Message<EventType>::Ptr PubSubSystem::createMessage(const ChannelName channel, const EventType & payload) {
                return getChannelHandle<EventType>(channel)->createMessage(payload);
        }

        template<class EventType> typename message::M2Message<EventType>::Ptr PubSubSystem::createMessage(const ChannelName channel) {
                return createMessage<EventType>(channel, EventType());
        }

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

#endif /* __M2ETIS_PUBSUB_PUBSUBSYSTEM_H__ */