Шаблон наблюдателя + Шаблон посетителя для системы сообщений

Некоторые шаблоны метапрограммирования:

// a bundle of types:
template<class...>struct types{using type=types;};

// a type that does nothing but carry a type around
// without being that type:
template<class T>struct tag{using type=T;};

// a template that undoes the `tag` operation above:
template<class Tag>using type_t=typename Tag::type;

// a shorter way to say `std::integral_constant<size_t, x>`:
template<std::size_t i>struct index:std::integral_constant<std::size_t, i>{};

Получить индекс типа в types<...>:

// this code takes a type T, and a types<...> and returns
// the index of the type in there.
// index_of
namespace details {
template<class T, class Types>
struct index_of{};
}
template<class T, class Types>
using index_of_t=type_t<details::index_of<T,Types>>;
namespace details {
// if the first entry in the list of types is T,
// our value is 0
template<class T, class...Ts>struct index_of<T, types<T,Ts...>>:
tag< index<0> >
{};
// otherwise, it is 1 plus our value on the tail of the list:
template<class T, class T0, class...Ts>
struct index_of<T, types<T0, Ts...>>:
tag< index< index_of_t<T,types<Ts...>{}+1 > >
{};
}

Вот один «канал» вещатель (он отправляет один вид сообщения):

// a token is a shared pointer to anything
// below, it tends to be a shared pointer to a std::function
// all we care about is the lifetime, however:
using token = std::shared_ptr<void>;
template<class M>
struct broadcaster {
// f is the type of something that can eat our message:
using f = std::function< void(M) >;
// we keep a vector of weak pointers to people who can eat
// our message.  This lets them manage lifetime independently:
std::vector<std::weak_ptr<f>> listeners;

// reg is register.  You pass in a function to eat the message
// it returns a token.  So long as the token, or a copy of it,
// survives, broadcaster will continue to send stuff at the
// function you pass in:
token reg( f target ) {
// if thread safe, (write)lock here
auto sp = std::make_shared<f>(std::move(target));
listeners.push_back( sp );
return sp;
// unlock here
}
// removes dead listeners:
void trim() {
// if thread safe, (try write)lock here
// and/or have trim take a lock as an argument
listeners.erase(
std::remove_if( begin(listeners), end(listeners), [](auto&& p){
return p.expired();
} ),
listeners.end()
);
// unlock here
}
// Sends a message M m to every listener who is not dead:
void send( M m ) {
trim(); // remove dead listeners
// (read) lock here
auto tmp_copy = listeners; // copy the listeners, just in case
// unlock here

for (auto w:tmp_copy) {
auto p = w.lock();
if (p) (*p)(m);
}
}
};

Вот многоканальный subject который может поддерживать любое количество различных типов сообщений (определяется во время компиляции). Если вы не соответствуете типу сообщения, send и / или reg не удастся скомпилировать. Вы несете ответственность за решение, является ли сообщение const& или значение или что-то еще. Пытаясь reg Rvalue сообщение не будет работать. Предполагается, что M передается reg а также send в явном виде.

// fancy wrapper around a tuple of broadcasters:
template<class...Ts>
struct subject {
std::tuple<broadcaster<Ts>...> stations;
// helper function that gets a broadcaster compatible
// with a message type M:
template<class M>
broadcaster<M>& station() {
return std::get< index_of_t<M, types<Ts...>>{} >( stations );
}
// register a message of type M.  You should call with M explicit usually:
template<class M>
token reg( std::function<void(M)> listener ) {
return station<M>().reg(std::move(listener));
}
// send a message of type M.  You should explicitly pass M usually:
template<class M>
void send( M m ) {
station<M>().send(std::forward<M>(m));
}
};

живой пример.

Когда ты reg, он возвращает tokenака std::shared_ptr<void>, Пока этот токен (или копия) сохраняется, сообщения будут передаваться. Если это уходит, сообщения к reged callback закончатся. Как правило, это означает, что слушатели должны поддерживать std::vector<token>и рег лямбды, которые используют this волей-неволей.

В C ++ 14 / 1z вышеупомянутое становится немного лучше (мы можем покончить с types<...> а также index_of для одного).

Если вы добавите прослушиватель во время цикла вещания, он не будет отправлен. Если вы удалите слушателя во время цикла вещания, он не будет отправлен после точки, в которую вы его удалили.

Потокобезопасные комментарии настроены для блокировки чтения / записи на вещателе.

Память, выделенная для мертвых слушателей для данного вещателя, восстанавливается, когда trim или же send называется. Тем не менее std::function будет уничтожен очень давно, так что до следующего send, Я делаю это тогда, потому что мы собираемся перебирать список сообщений, так или иначе, может сначала убрать любой беспорядок.

Это решение не имеет RTTI или динамического приведения, а сообщения отправляются только слушателям, которые их понимают.

В C ++ 17 все становится проще. Удалите весь шаблон метапрограммирования, удалите subject (держать broadcaster) и просто сделайте это для обработки более одного канала:

template<class...Ms>
struct broadcasters : broadcaster<Ms>... {
using broadcaster<Ms>::reg...;
using broadcaster<Ms>::send...;

template<class M>
broadcaster<M>& station() { return *this; }
};

этот broadcasters в настоящее время почти полное улучшение subject выше.

Из-за улучшений в std::function поскольку C ++ 11, reg Функция обычно делает правильные вещи, если параметры сигнала не слишком похожи. Если у вас возникнут проблемы с reg или же sendВы вынуждены звонить .station<type>().reg(blah),

Но 99/100 раз вы можете просто сделать .reg( lambda ) а также .send( msg ) и разрешение перегрузки делает правильную вещь.

Живой пример.

А вот и вся система, дополненная модульной системой обеспечения безопасности нитей:

struct not_thread_safe {
struct not_lock {~not_lock(){}};
auto lock() const { return not_lock{}; }
};
struct mutex_thread_safe {
auto lock() const { return std::unique_lock<std::mutex>(m); }
private:
mutable std::mutex m;
};
struct rw_thread_safe {
auto lock() { return std::unique_lock<std::shared_timed_mutex>(m); }
auto lock() const { return std::shared_lock<std::shared_timed_mutex>(m); }
private:
mutable std::shared_timed_mutex m;
};
template<class D, class>
struct derived_ts {
auto lock() { return static_cast<D*>(this)->lock(); }
auto lock() const { return static_cast<D const*>(this)->lock(); }
};
using token = std::shared_ptr<void>;
template<class M, class TS=not_thread_safe>

struct broadcaster:
TS
{
using f = std::function< void(M) >;
mutable std::vector<std::weak_ptr<f>> listeners;
token reg( f target )
{
auto l = this->lock();
auto sp = std::make_shared<f>(std::move(target));
listeners.push_back( sp );
return sp;
}
// logically const, but not really:
void trim() const {
auto l = const_cast<broadcaster&>(*this).lock();
auto it = std::remove_if( listeners.begin(), listeners.end(), [](auto&& p){
return p.expired();
} );
listeners.erase( it, listeners.end() );
}
// logically const, but not really:
void send( M m ) const
{
trim(); // remove dead listeners
auto tmp_copy = [this]{
auto l = this->lock();
return listeners; // copy the listeners, just in case
}();

for (auto w:tmp_copy) {
auto p = w.lock();
if (p) (*p)(m);
}
}
};
template<class TS, class...Ms>
struct basic_broadcasters :
TS,
broadcaster<Ms, derived_ts<basic_broadcasters<TS, Ms...>, Ms> >...
{
using TS::lock;
using broadcaster<Ms, derived_ts<basic_broadcasters<TS, Ms...>, Ms> >::reg...;
using broadcaster<Ms, derived_ts<basic_broadcasters<TS, Ms...>, Ms> >::send...;

template<class M>
broadcaster<M, derived_ts<basic_broadcasters<TS, Ms...>, M>>& station() { return *this; }
template<class M>
broadcaster<M, derived_ts<basic_broadcasters<TS, Ms...>, M>> const& station() const { return *this; }
};
template<class...Ms>
using broadcasters = basic_broadcasters<rw_thread_safe, Ms...>;

Живой пример.

broadcasters<Messages...> теперь является широковещательным классом с блокировкой чтения и записи, который использует 1 общую разделяемую блокировку для синхронизации каждой очереди широковещания.

basic_broadcasters<not_thread_safe, Messages...> вместо этого создает один без блокировки (то есть не является потокобезопасным).