5
\$\begingroup\$

I've written a basic event messenger broadcasting service. This is meant to allow for broadcasting events between loosely coupled areas of code where you may not have or care about a direct reference to the specific broadcast source (as you might if using Signals/Slots for example), and the broadcaster does not know who might care about that event.

The shared concept for events must be a common broadcast argument, or string key (or both). There was a desire not to need to inherit from some kind of "EventData" object when deciding what message to pass. I have two methods of observing. Either connection via lambdas or having an observer class derive from MessengerObserver depending on your needs. There are data locality vs size of observer logic tradeoffs that I leave up to the developer by providing these interfaces.

I have a single header file dependency on typestring which is visible here: https://github.com/irrequietus/typestring due to limiting this code to C++17 (C++20 will resolve this dependency by introducing template type strings.)

I'm looking for any feedback/thoughts. Issues/errors, and general thoughts on style and usability of the API. Thank you! This will be published on my github after review: https://github.com/M2tM

messenger.hpp

#include <iostream>
#include <string>
#include <vector>
#include <map>
#include <mutex>
#include <algorithm>
#include <typeindex>
#include "typestring.hpp"

#define MV_KEY(x) typestring_is(x)

namespace MV {
    struct MessageHandle {
        virtual ~MessageHandle() {}
    };

    class MessageCallerNoArgCastable : public MessageHandle {
    public:
        typedef void type;

        virtual bool operator()() = 0;
    };
    template<typename F>
    class MessageCallerNoArg : public MessageCallerNoArgCastable {
    public:
        MessageCallerNoArg(F&& a_f) : f(a_f) {}
        bool operator()() override {
            if constexpr (std::is_pointer<F>::value){
                if constexpr (std::is_same<decltype((*f)()), bool>::value){
                    return (*f)();
                } else {
                    (*f)();
                    return true; //never self kill
                }
            }else{
                if constexpr (std::is_same<decltype(f()), bool>::value){
                    return f();
                } else {
                    f();
                    return true; //never self kill
                }
            }
        }
    private:
        F f;
    };

    template<typename T>
    class MessageCallerCastable : public MessageHandle {
    public:
        typedef T type;

        virtual bool operator()(const T&) = 0;
    };
    template<typename F, typename T>
    class MessageCaller : public MessageCallerCastable<T> {
    public:
        MessageCaller(F&& a_f) :f(a_f){}
        bool operator()(const T &a_value) override {
            if constexpr (std::is_pointer<F>::value){
                if constexpr (std::is_same<decltype((*f)(a_value)), bool>::value){
                    return (*f)(a_value);
                } else {
                    (*f)(a_value);
                    return true; //never self kill
                }
            }else{
                if constexpr (std::is_same<decltype(f(a_value)), bool>::value){
                    return f(a_value);
                } else {
                    f(a_value);
                    return true; //never self kill
                }
            }
        }
    private:
        F f;
    };

    template<typename F>
    std::shared_ptr<MessageHandle> make_message_caller(F&& a_f) {
        return std::static_pointer_cast<MessageHandle>(std::make_shared<MessageCallerNoArg<typename std::decay<F>::type>>(std::forward<F>(a_f)));
    }
    template<typename T, typename F>
    std::shared_ptr<MessageHandle> make_message_caller(F&& a_f) {
        if constexpr (std::is_same<T, void>::value){
            return make_message_caller(std::forward<F>(a_f));
        }else{
            return std::static_pointer_cast<MessageHandle>(std::make_shared<MessageCaller<typename std::decay<F>::type, T>>(std::forward<F>(a_f)));
        }
    }

    namespace detail {
        template<typename T>
        struct is_string : public std::disjunction<
            std::is_same<char *, typename std::decay<T>::type>,
            std::is_same<const char *, typename std::decay<T>::type>,
            std::is_same<std::string, typename std::decay<T>::type>> {
        };
    }

    class Messenger {
    private:
        struct EventKey {
            std::string key;
            std::type_index valueType;
            bool operator<(const EventKey &a_rhs) const{
                return std::tie(valueType, key) < std::tie(a_rhs.valueType, a_rhs.key);
            }
        };

        template <typename T>
        static EventKey makeKey(const std::string &a_key) {
            return {a_key, typeid(T)};
        }

    public:
        template <typename T>
        void broadcast(const std::string &a_key, const T &a_value) {
            if constexpr(detail::is_string<T>::value){ //force const char * -> std::string to enable m.broadcast("Key", "Value");
                broadcastCommon<MessageCallerCastable<std::string>>(a_key, [&](auto&& callable) -> bool{ return (*callable)(a_value); });
            }else{
                broadcastCommon<MessageCallerCastable<T>>(a_key, [&](auto&& callable) -> bool{ return (*callable)(a_value); });
            }
        }

        template <typename ValueType>
        void broadcast(const ValueType& a_value) {
            static_assert(!detail::is_string<ValueType>::value, "Disambiguate broadcast of a string by using either broadcastKey or broadcastValue to clarify intent.");
            broadcast(std::string(), a_value);
        }

        //Zero Argument broadcast of a specific event key.
        void broadcastKey(const std::string& a_key) {
            broadcastCommon<MessageCallerNoArgCastable>(a_key, [&](auto&& callable) -> bool{ return (*callable)(); });
        }

        //One Argument broadcast of a string value to any unkeyed observers.
        void broadcastValue(const std::string& a_value){
            broadcast(std::string(), a_value);
        }

        template <typename T, typename F>
        [[nodiscard]] std::shared_ptr<MessageHandle> observe(const std::string &a_key, F&& a_method) {
            std::lock_guard<std::recursive_mutex> guard(mutex);
            auto observer = make_message_caller<T>(std::forward<F>(a_method));
            auto key = makeKey<T>(a_key);
            if (activeKeys[key] > 0) {
                pendingObservers[key].push_back(observer);
            } else {
                observers[key].push_back(observer);
            }
            return observer;
        }
        template <typename F>
        [[nodiscard]] std::shared_ptr<MessageHandle> observe(const std::string &a_key, F&& a_method) {
            return observe<void, F>(a_key, std::forward<F>(a_method));
        }
        template <typename T, typename F>
        [[nodiscard]] std::shared_ptr<MessageHandle> observe(F&& a_method) {
            return observe<T, F>(std::string(), std::forward<F>(a_method));
        }
    private:
        template <typename T>
        struct ScopedKeyLock {
            ScopedKeyLock(const std::string &a_key, Messenger& a_owner) :
                key(makeKey<T>(a_key)),
                owner(a_owner),
                guard(a_owner.mutex),
                observerCollection(a_owner.observers[key]){

                ++owner.activeKeys[key];
            }
            ~ScopedKeyLock() {
                if (--owner.activeKeys[key] == 0) {
                    auto& pendingOserversRef = owner.pendingObservers[key];
                    for (auto&& pendingObserver : pendingOserversRef) {
                        if (!pendingObserver.expired()) {
                            observerCollection.push_back(pendingObserver);
                        }
                    }
                    pendingOserversRef.clear();
                }
            }

            EventKey key;
            Messenger& owner;
            std::lock_guard<std::recursive_mutex> guard;
            std::vector<std::weak_ptr<MessageHandle>>& observerCollection;
        };

        template <typename T>
        friend class ScopedKeyLock;

        template <typename T, typename F>
        void broadcastCommon(const std::string& a_key, F&& a_invokeCaller) {
            ScopedKeyLock<typename T::type> scopedLock(a_key, *this);
            scopedLock.observerCollection.erase(std::remove_if(scopedLock.observerCollection.begin(), scopedLock.observerCollection.end(), [&](auto weakObserver) {
                if (auto observer = weakObserver.lock()) {
                    return !a_invokeCaller(std::static_pointer_cast<T>(observer));
                } else {
                    return true;
                }
            }), scopedLock.observerCollection.end());
        }

        std::recursive_mutex mutex;
        std::map<EventKey, int> activeKeys;
        std::map<EventKey, std::vector<std::weak_ptr<MessageHandle>>> observers;
        std::map<EventKey, std::vector<std::weak_ptr<MessageHandle>>> pendingObservers;
    };

    struct KeyPairComparer {};

    template <typename KeyParam, typename T = void>
    struct KeyPair : public KeyPairComparer {
        static const char * key() noexcept {return KeyParam::data();}
        typedef T type;
    };

    template <typename DerivedType, typename ... ObservableTypes>
    class MessengerObserver {
    public:
        MessengerObserver(MV::Messenger &a_m){
            autoObserve<ObservableTypes...>(a_m);
        }
        virtual ~MessengerObserver(){}

    protected:
        //Allow manual connection
        void addHandle(const std::shared_ptr<MV::MessageHandle> &a_handle){
            handles.push_back(a_handle);
        }
    private:
        template <int = 0>
        void autoObserve(MV::Messenger &a_m) {
        }

        template <typename T, typename... Ts>
        void autoObserve(MV::Messenger &a_m) {
            if constexpr (std::is_base_of<KeyPairComparer, T>::value){
                if constexpr (std::is_same<typename T::type, void>::value){
                    handles.push_back(a_m.observe(T::key(),static_cast<DerivedType*>(this)));
                }else{
                    handles.push_back(a_m.observe<typename T::type>(T::key(),static_cast<DerivedType*>(this)));
                }
                autoObserve<Ts...>(a_m);
            }else{
                handles.push_back(a_m.observe<T>(static_cast<DerivedType*>(this)));
                autoObserve<Ts...>(a_m);
            }
        }
        std::vector<std::shared_ptr<MV::MessageHandle>> handles;
    };
}

main.cpp

#include "messenger.hpp"

struct MyEventData {
    int a;
    std::string b;
};

//This serves as an example of creating a type which manages its own connection and implicitly connects to a messenger through the derived MessengerObserver interface.
struct MyMultiObserver : public MV::MessengerObserver<MyMultiObserver, MV::KeyPair<MV_KEY("NoDataEvent")>, MV::KeyPair<MV_KEY("KeyedEvent"), int>, double, std::string> {
    MyMultiObserver(MV::Messenger &a_m):
        MessengerObserver(a_m){
        addHandle(a_m.observe<bool>(this)); //allow manual hookup/ownership too.
    }
    void operator()(int v){
        std::cout << "MultiObserver [KeyedEvent|int]: " << v << std::endl;
    }
    bool operator()(double v){
        std::cout << "MultiObserver Self Disconnect [double]: " << v << std::endl;
        return false; //only listen once.
    }
    void operator()(){
        std::cout << "MultiObserver [NoData]" << std::endl;
    }

    //hooked up manually
    void operator()(bool v){
        std::cout << "MultiObserver [bool]: " << v << std::endl;
    }

    void operator()(const std::string &a_message){
        std::cout << "MultiObserver [string]: " << a_message << std::endl;
    }
};

void broadcastTest(MV::Messenger& m);

int main() {
    MV::Messenger m;

    std::shared_ptr<MV::MessageHandle> disconnectableHandle;
    std::cout << "\nBegin Test: Manual Disconnect\n";
    {
        disconnectableHandle = m.observe<int>([](int value) {
            std::cout << "Manual Disconnect [int]: " << value << std::endl;
        });
        m.broadcast(2);
        disconnectableHandle.reset();
        m.broadcast(2); // no output.
    }

    std::cout << "\nBegin Test: Lambda\n";
    {
        std::vector<std::shared_ptr<MV::MessageHandle>> handles;
        handles.push_back(m.observe<int>([&](int value) {
            std::cout << "Lambda Self Disconnect [int]: " << value << std::endl;
            //example of adding an observer from within a callback.
            handles.push_back(m.observe<int>([](int value) {
                std::cout << "Lambda Nested [int]: " << value << std::endl;
            }));
            return false; //return false to remove this, the return value is optional.
        }));
        handles.push_back(m.observe<double>([](double value) {
            std::cout << "Lambda [double]: " << value << std::endl;
        }));
        handles.push_back(m.observe<MyEventData>([](const MyEventData& value) {
            std::cout << "Lambda [MyEventData]: " << value.a << ", " << value.b << std::endl;
        }));
        handles.push_back(m.observe("NoDataEvent", []() {
            std::cout << "Lambda [NoData]" << std::endl;
        }));
        handles.push_back(m.observe<int>("KeyedEvent", [](int value) {
            std::cout << "Lambda [KeyedEvent|int]: " << value << std::endl;
        }));
        broadcastTest(m);
    }

    std::cout << "\nBegin Test: MyMultiObserver\n";
    {
        MyMultiObserver listener(m);
        broadcastTest(m);
    }

    broadcastTest(m); // no output expected
}

void broadcastTest(MV::Messenger& m){
    m.broadcast(1);
    m.broadcast(2);
    m.broadcast(3.5);
    m.broadcast(4.5);

    m.broadcast(MyEventData{ 5, std::string("Cool Event") });
    //Example of a keyed event.
    m.broadcast("KeyedEvent", 6);
    m.broadcastKey("NoDataEvent");
    m.broadcastValue("My_Message");
    m.broadcast(false);
}

Console Output:

Begin Test: Manual Disconnect
Manual Disconnect [int]: 2

Begin Test: Lambda
Lambda Self Disconnect [int]: 1
Lambda Nested [int]: 2
Lambda [double]: 3.5
Lambda [double]: 4.5
Lambda [MyEventData]: 5, Cool Event
Lambda [KeyedEvent|int]: 6
Lambda [NoData]

Begin Test: MyMultiObserver
MultiObserver Self Disconnect [double]: 3.5
MultiObserver [KeyedEvent|int]: 6
MultiObserver [NoData]
MultiObserver [string]: My_Message
MultiObserver [bool]: 0
\$\endgroup\$
3
  • 1
    \$\begingroup\$ Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. Feel free to post a new question linking back to this one. \$\endgroup\$ – Mast Feb 19 '20 at 6:56
  • \$\begingroup\$ Thank you, I think I'll just be posting the updated version to my github at this point. I apologize for breaking that rule. \$\endgroup\$ – M2tM Feb 19 '20 at 20:07
  • \$\begingroup\$ Oh, don't worry. It's confusing for people new to the site and we don't advertise it broadly enough. Luckily we keep track of these things, no problem. \$\endgroup\$ – Mast Feb 19 '20 at 20:13
3
+50
\$\begingroup\$

The code fails to build for me, but that could be platform and compiler compatibility. I attempted to build this on Windows 10 using CLion and Visual Studio 2019. Here are the errors and warnings I received:

main.cpp(9): error C2672: 'irqus::typeek': no matching overloaded function found
main.cpp(9): error C2893: Failed to specialize function template 'unknown-type irqus::typeek(irqus::typestring)'
typestring.hpp(110): note: see declaration of 'irqus::typeek'
main.cpp(9): note: With the following template arguments:
main.cpp(9): note: 'C={75, 101, 121, 101, 100, 69, 118, 101, 110, 116, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}'
main.cpp(9): error C3203: 'KeyPair': unspecialized class template can't be used as a template argument for template parameter 'ObservableTypes', expected a real type

Note: line 9 in main.cpp is where the macro MV_KEY(x) is being used and the expansion of that macro is failing.

Maintainability

The most major issues in this code maintainability and expand-ability. When software engineer/developer designs a solution for a product they need to keep in mind that the product may be around for a long time and bug fixes and feature requests will be very common, especially in the early stages. The code must be easy to read, write, debug and expand. The code should be modular so that parts that are working as expected will not need to be changed when other portions of the code need to be updated or corrected.

The structure of this program where all the classes and multiple namespaces are declared in a single header file is a problem. The file messenger.hpp should be broken up into multiple files, each class within messenger.hpp should have it's own header file. As an aide in building the project, each class should also have it's own .cpp file as well. Separating the declarations from the executable code reduces build time and eases shipping bug fixes to the users.

It is possible that messenger.hpp can be a single file that includes all the other messenger header file to ease implementation.

Using Macros in C++

Macros in C++ are generally frowned upon, a major reason is because macro's can't be debugged and there is no type checking in macros, There are a few good uses of macros, primarily as Include Guards that prevent an included file from being included a multiple times (can cause compilation errors). There is also a discussion of this on stackoverflow. You may want to read why macros are considered evil as well.

One alternative to macros are templates. In the C programming language templates are not available and macros are used for generic functions.

Including a whole library which is macros (typestring.hpp) is somewhat questionable in C++.

Object Oriented Programming

It's not really clear why there is a struct rather than a class that instantiates the MyMultiObserver objects.

Include Files

The file main.cpp should directly include <iostream>, and it should be removed from messenger.hpp since IO is only performed in main.cpp.

Complexity

In addition to the complexity caused by declaring multiple classes in messenger.hpp the function main() is too complex (does too much) and could be broken up into 3 function.

\$\endgroup\$
7
  • \$\begingroup\$ Thank you for your feedback! I've made meaningful changes based on a few of your points. The most notable of which was failure to compile in visual studio. I've gone ahead and removed the original dependency on irrequietus typestring since it seemed to fail, replacing it with a working solution. I've also worked around a few "if constexpr" bugs and reported it to Microsoft: developercommunity.visualstudio.com/content/problem/920784/… This should now work just fine in Visual Studio 2019 with C++17 compilation selected \$\endgroup\$ – M2tM Feb 18 '20 at 20:00
  • \$\begingroup\$ For your Maintainability headline: I do not feel like splitting 283 lines of relatively tightly coupled classes is going to significantly improve the maintainability of this project, if the library were larger I would do so. Thank you for the feedback anyway, modularization is important in code mantainability. This is a header-only library and as such there are no .cpp files required for the class. \$\endgroup\$ – M2tM Feb 18 '20 at 20:03
  • \$\begingroup\$ For your Macros in C++ Heading: If you can point me at a template-only solution in C++17 which supports a unique type based on a string as MV_KEY("Contents") creates, then I will wholly embrace it. Unfortunately I do not believe it is possible, and there are still cases where Macros are necessary in C++ for simpler syntax of certain kinds of otherwise very verbose and obfuscated hand-written code. It is overly simplistic to say Macros are evil, though I totally agree where they can be avoided they should be. Again, if an alternative exists for this case, please share it. \$\endgroup\$ – M2tM Feb 18 '20 at 20:14
  • \$\begingroup\$ For your feedback on Object Oriented Programming: struct and class are identical in C++ except for their default access level (public vs private). MyMultiObserver was specified as a struct originally simply to avoid having to have a single accessor: "public:" at the top, but I've changed it. \$\endgroup\$ – M2tM Feb 18 '20 at 20:16
  • \$\begingroup\$ For your feedback on Include Files: I've resolved this, thanks for pointing that out! \$\endgroup\$ – M2tM Feb 18 '20 at 20:16
4
\$\begingroup\$

Sorry, I probably should have used a "comment" for this, but it is way bigger than their limited size.

In all fairness I did not read your code in full; I only scanned it. The thing that struck me as I was reading, is that this is incredibly hard to do well. Assuming a robust enterprise environment, you run into all kinds of issues that take lots of thought and lots of code to deal with.

  • What happens if a thread/process/machine dies before processing a message?
  • What happens if you need to go multi-process or multi-server because of load?
  • What happens if what you want to send to a process or system that isn't up yet?

These and 100s more issues just seem to show themselves over time.

I'd use RabbitMQ instead (though I understand Microsoft also has a messaging offering). RabbitMQ is very robust. If you need bulletproofing, you can set up servers on multiple machines such that if one goes down, no messages are lost. And there all kinds of fanout options and keeping of messages even if the receiver is down. Even if you are in a single process on a single machine, RabbitMQ offers you options and protections you are going to want and haven't thought of yet.

Your first thought might be, but I don't want to make a multi-process call because it is too expensive. Since I didn't read your code, perhaps I am completely off base here. However, I claim that if this is for an enterprise-level system you are going to want more eventually.

The other thing you could do is switch to C# and use delegates which are built into the language.

\$\endgroup\$
5
  • \$\begingroup\$ This might also be a helpful read: stackoverflow.com/questions/9568150/what-is-a-c-delegate \$\endgroup\$ – Frank Merrow Feb 15 '20 at 4:46
  • \$\begingroup\$ After thinking about my rant above, I've decided to offer a more low key answer. This is a common design pattern. I Googled around and found a couple of delegate libraries already out there for C++. Why reinvent the wheel? NDH (Not Developed Here) is a professional disease. Everybody has it to one extent or another, but whenever possible I suggest fighting the urge unless there are obvious faults with other implementations. \$\endgroup\$ – Frank Merrow Feb 15 '20 at 5:39
  • \$\begingroup\$ Recommending a particular vendor is beyond the scope of a review, there are quite a number of vendors for messaging, many of them are free below a certain message level. ably.io \$\endgroup\$ – pacmaninbw Feb 15 '20 at 15:52
  • \$\begingroup\$ I appreciate being linked another library but I fail to see how this passes for a code review. You make a lot of assumptions presuming I have the option of shifting languages for the project I’m working on as well. Finally, I obviously know what a delegate is if you read my code I have such structures. :/ \$\endgroup\$ – M2tM Feb 17 '20 at 20:12
  • \$\begingroup\$ RabbitMQ grossly exceeds the scope of the problems I'm trying to solve. I’m working with iOS and Android and this is intended to work within a single application. \$\endgroup\$ – M2tM Feb 17 '20 at 20:17

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.