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
