note: This is an updated design after a previous review round.

I have written a class that handles listeners registering callbacks to receive messages. Registered callbacks can be replaced or removed. Code is multithreaded in that any thread can broadcast to the listeners, and any thread can add, replace or remove listeners.

For this second round, I have made a list of requirements for the class:

1. there may be zero, one or multiple listeners
2. listeners can be registered or deregistered from different threads
3. each listener should be called exactly once for each message
4. all listeners should receive all messages
5. messages may originate from multiple sources in different threads
6. each message should be forwarded to all listeners before the next message is processed (messages are not sent in parallel)
7. each listener should receive the messages in the same order, i.e., the order in which they are provided to the broadcaster (from the view of the broadcaster)
8. listeners can be removed or replaced
9. listeners are uniquely identified by a cookie that remains valid if the listener is not deregistered
10. the cookie remains valid if the listener is replaced
11. anyone with the cookie should be able to deregister or replace the listener
12. listeners should never be called again when deregistration completes
13. the old listener should never be called again when replace completes
14. listeners should be able to deregister or replace themselves when invoked, listeners do not need to be able to register more callbacks
15. listeners are noexcept, so code shouldn’t deal with exceptions thrown by listeners
16. trying to remove or replace the callback associated with an expired cookie is not an error, the operation should simply do nothing, and report to the caller that nothing was done.
17. adding, replacing and removing listeners happens much less frequently than broadcasting.

A further consideration that is not enforced in code:

• Users are expected to read documentation that states which functions are safe to call from inside a listener, and which must not. (Ideal would be a way to check if a function is called from a listener, so we can error in that case. I can’t figure out a way to do that. How to distinguish the case where the lock in, e.g., add_listener can’t be acquired because we’re calling from the callback (can’t call lock, would deadlock), vs where it can’t be acquired because callbacks are running concurrently (calling lock is correct in that case)?)

## Implementation

#include <vector>
#include <map>
#include <utility>
#include <optional>
#include <mutex>
#include <functional>
#include <type_traits>

// usage notes:
// 1. This class provides no lifetime management of the link, which means
//    the user is responsible to ensure that either
//    a. the registered callback outlives this broadcaster, or
//    b. they revoke their callbacks before destroying any resource it
//       points to.
// 2. None of the member functions of this class may be called from
//    inside an executing listener, except
//    replace_callback_from_inside_callback() and
//    remove_listener_from_inside_listener(). Calling them anyway will
// 3. replace_callback_from_inside_callback() and
//    remove_listener_from_inside_listener() may _only_ be called from
//    inside an executing listener.
template <class... Message>
{
public:
using listener   = std::function<void(Message...)>;

// should never be called from inside a running listener
template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
{
auto l = lock();
_targets.emplace_back(listener{ std::forward<F>(r_) });
_target_registry[++_lastID] = _targets.size() - 1;
return _lastID;
}

// should never be called from inside a running listener,
// in that case, use replace_listener_from_inside_listener()
template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
{
auto l = lock();
return replace_listener_impl(c_, listener{ std::forward<F>(r_) });
}

// should never be called from inside a running listener,
// in that case, use remove_listener_from_inside_listener()
{
auto l = lock();
return remove_listener_impl(c_);
}

void notify_all(const Message&... msg_) noexcept
{
auto l = lock();

// invoke all listeners with message
for (const auto& r: _targets)
r(msg_...);

// execute any updates to registered listeners
if (!_mutation_list.empty())
{
for (auto&& [cookie, r] : _mutation_list)
{
if (r.has_value())
else
}
_mutation_list.clear();
}
}

// !! should only be called from inside a running listener!
template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
{
_mutation_list.emplace_back(c_, listener{ std::forward<F>(r_) });
}

// !! should only be called from inside a running listener!
{
_mutation_list.emplace_back(c_, std::nullopt);
}

private:
{
return std::unique_lock<std::mutex>(_mut);
}
{
auto index = get_target_index(c_);
if (index)
{
// valid slot, update listener stored in it
auto it = _targets.begin() + *index;
*it = std::move(r_);
return true;
}
return false;
}
{
auto index = get_target_index(c_);
if (index)
{
// valid slot, remove listener stored in it
auto it = _targets.begin() + *index;
_targets.erase(it);

// fixup slot registry, items past deleted one just moved one idx back
{
if (idx > *index)
--idx;
}

// remove entry from slot registry
_target_registry.erase(c_);

return true;
}
return false;
}
{
auto map_it = _target_registry.find(c_);
if (map_it == _target_registry.end())
return {};

return map_it->second;
}

private:
std::mutex                    _mut;

std::vector<listener>         _targets;

};


## Example usage

#include <iostream>
#include <string>
#include <functional>

void freeTestFunction(std::string msg_) noexcept
{
std::cout << "from freeTestFunction: " << msg_ << std::endl;
}
struct test
{

void simpleCallback(std::string msg_) noexcept
{
std::cout << "from simpleCallback: " << msg_ << std::endl;
}
void oneShotCallback(std::string msg_) noexcept
{
std::cout << "from oneShotCallback: " << msg_ << std::endl;
}
void twoStepCallback_step1(std::string msg_) noexcept
{
std::cout << "from twoStepCallback_step1: " << msg_ << std::endl;

// replace callback (so don't request to delete through return argument!)
}
void twoStepCallback_step2(std::string msg_) noexcept
{
std::cout << "from twoStepCallback_step2: " << msg_ << std::endl;
}

void runExample()
{

}

};

int main(int argc, char **argv)
{
test t;
t.runExample();

return 0;
}

• "one or multiple listeners"? Does that mean that there can't be zero listeners, or did you forget to include that in the requirements? – Toby Speight Jan 8 at 16:52
• @TobySpeight Ah, right. yes there could also be zero, and that should work fine (it does, was in the test case already). Thanks, i've corrected the text. – Diederick C. Niehorster Jan 8 at 21:05

I have just one suggestion but you should evaluate in terms of performance. What I mean is that you should do a performance test with your current code and then with the new changes proposed to see if you gain.

You are using a lock() function inside the notify_all function.

void notify_all(const Message&... msg_) noexcept
{
auto l = lock();

// invoke all listeners with message
for (const auto& r: _targets)
r(msg_...);

// execute any updates to registered listeners
if (!_mutation_list.empty())
{
for (auto&& [cookie, r] : _mutation_list)
{
if (r.has_value())
else
}
_mutation_list.clear();
}
}


The suggested change is to move the scope of that lock the close to where you need to lock it, for example close to the replace_listener_impl and remove_listener_impl.

void notify_all(const Message&... msg_) noexcept
{
// invoke all listeners with message
for (const auto& r: _targets)
r(msg_...);

// execute any updates to registered listeners
if (!_mutation_list.empty())
{
auto l = lock(); // First option
for (auto&& [cookie, r] : _mutation_list)
{
if (r.has_value())
else
}
_mutation_list.clear();
}
}


Another alternative is to move the lock() calls that you have on the functions replace_listener and remove_listener to the _impl functions.

template <class F> requires (std::is_nothrow_invocable_r_v<void, F, Message...>)
{
// Remove auto l = lock();
return replace_listener_impl(c_, listener{ std::forward<F>(r_) });
}


Move it to replace_listener_impl

bool replace_listener_impl(cookieType c_, listener&& r_)
{
auto l = lock(); // Added
auto index = get_target_index(c_);
if (index)
{
// valid slot, update listener stored in it
auto it = _targets.begin() + *index;
*it = std::move(r_);
return true;
}
return false;
}


In any of the options that you will choose you should conduct a performance test to see what is more suitable for you, may be is slower or not, so better get the numbers so you have the evidence of what is better for you.

Hope it helps

• Thanks for the comment! I can't move the lock, or it wouldn't be safe anymore. That is, if i move the lock down in notify_all(), or to the _impl function, the vector of listeners could be changed while iterating over it, which would break stuff. Do i see that correctly? – Diederick C. Niehorster Jan 8 at 21:07