mtcore::thread::Subscription< T > Struct Template Reference

Represents a pub/sub subscription which can receive messages of type T Under the hood, each publisher maintains a message broker which stores all messages Subscribers will hold an Arc to the broker and will retrieve messages from the broker Due to the broker being shared, it does require more locks for both reads and writes This means it has a pretty low throughput when there is lots of contention The primary advantage is it has a much more convenient API for more non-blocking operations or timed blocks The reason this happens is that publishers are given greater insight into each subscription, and can better avoid partial writes. More...

#include <pub_sub.hpp>

Public Types
using	Message = T

Public Member Functions
Result< Optional< T >, ChannelTryReceiveErrors >	try_receive ()
	Tries to receive a message without blocking.
Optional< T >	receive_block ()
	Blocks until a message is ready.
template<typename Clock, typename Duration>
Result< Optional< T >, ChannelReceiveBeforeErrors >	receive_before (const std::chrono::time_point< Clock, Duration > &timePoint)
	Blocks until a message is ready, or erros if a timeout occurs.
void	deinit (Allocator &alloc)
	Deinitializes and cancels Subscription Note: This will cancel all copies of this particular Subscription.
void	close ()
	Subscription ()=default
	Subscription (const Arc< impl::Broker< T > > &broker, const Handle &handle)

Detailed Description

template<typename T>
struct mtcore::thread::Subscription< T >

Represents a pub/sub subscription which can receive messages of type T Under the hood, each publisher maintains a message broker which stores all messages Subscribers will hold an Arc to the broker and will retrieve messages from the broker Due to the broker being shared, it does require more locks for both reads and writes This means it has a pretty low throughput when there is lots of contention The primary advantage is it has a much more convenient API for more non-blocking operations or timed blocks The reason this happens is that publishers are given greater insight into each subscription, and can better avoid partial writes.

Partial sends may still happen, but they only happen if an allocating send function is used, and an allocation fails to grow the size of one of the channels.

Meant for use with select()

Template Parameters

T	Message Type

Definition at line 103 of file pub_sub.hpp.

Member Typedef Documentation

Message

template<typename T>

using mtcore::thread::Subscription< T >::Message = T

Definition at line 104 of file pub_sub.hpp.

Constructor & Destructor Documentation

Subscription() [1/2]

template<typename T>

mtcore::thread::Subscription< T >::Subscription ( )

default

Subscription() [2/2]

template<typename T>

mtcore::thread::Subscription< T >::Subscription ( const Arc< impl::Broker< T > > & broker, const Handle & handle )

inline

Definition at line 224 of file pub_sub.hpp.

: broker(broker), inboxHandle(handle) {}

Member Function Documentation

close()

template<typename T>

void mtcore::thread::Subscription< T >::close ( )

inline

Definition at line 220 of file pub_sub.hpp.

{ deinit(); }

Here is the call graph for this function:

deinit()

template<typename T>

void mtcore::thread::Subscription< T >::deinit ( Allocator & alloc )

inline

Deinitializes and cancels Subscription Note: This will cancel all copies of this particular Subscription.

Definition at line 207 of file pub_sub.hpp.

                                  {
      if (!broker.valid()) {
        return;
      }
      {
        auto lock = std::unique_lock(broker->mux);
        broker->inboxes[inboxHandle].deinit(*broker->alloc);
        broker->inboxes.remove(inboxHandle);
        broker->senderSignal.notify_all();
      }
      broker.release(alloc);
    }

Here is the caller graph for this function:

receive_before()

template<typename T>

template<typename Clock, typename Duration>

Result< Optional< T >, ChannelReceiveBeforeErrors > mtcore::thread::Subscription< T >::receive_before ( const std::chrono::time_point< Clock, Duration > & timePoint )

inline

Blocks until a message is ready, or erros if a timeout occurs.

Returns

Message (if present), nullopt if pub/sub closed or subscription cancelled, or error on timeout

Definition at line 173 of file pub_sub.hpp.

                                                                          {
      if (!broker.valid()) {
        return success(Optional<T>{nullopt});
      }
 
      auto lock = std::unique_lock(broker->mux, timePoint);
      if (!lock.owns_lock()) {
        return error(ChannelReceiveBeforeErrors::TIMEOUT);
      }
 
      if (!broker->inboxes.is_handle_valid(inboxHandle)) {
        return success(Optional<T>{nullopt});
      }
 
      do {
        if (auto res = broker->inboxes[inboxHandle].pop(); res.has_value()) {
          broker->senderSignal.notify_all();
          return res;
        }
 
        if (broker->closed) {
          return success(Optional<T>{nullopt});
        }
 
        if (broker->readSignal.wait_until(lock, timePoint) == std::cv_status::timeout) {
          return error(ChannelReceiveBeforeErrors::TIMEOUT);
        }
      } while (true);
    }

Here is the call graph for this function:

receive_block()

template<typename T>

Optional< T > mtcore::thread::Subscription< T >::receive_block ( )

inline

Blocks until a message is ready.

Returns

Message (if present), nullopt if pub/sub closed or subscription cancelled

Definition at line 142 of file pub_sub.hpp.

                                {
      if (!broker.valid()) {
        return Optional<T>{nullopt};
      }
 
      auto lock = std::unique_lock(broker->mux);
 
      if (!broker->inboxes.is_handle_valid(inboxHandle)) {
        return Optional<T>{nullopt};
      }
 
      do {
        if (auto res = broker->inboxes[inboxHandle].pop(); res.has_value()) {
          broker->senderSignal.notify_all();
          return res;
        }
 
        if (broker->closed) {
          return Optional<T>{nullopt};
        }
        broker->readSignal.wait(lock);
      } while (true);
    }

try_receive()

template<typename T>

Result< Optional< T >, ChannelTryReceiveErrors > mtcore::thread::Subscription< T >::try_receive ( )

inline

Tries to receive a message without blocking.

Returns

Message (if present), nullopt if pub/sub closed or subscription cancelled, or error if had to block

Definition at line 111 of file pub_sub.hpp.

                                                               {
      if (!broker.valid()) {
        return success(Optional<T>{nullopt});
      }
 
      auto lock = std::unique_lock(broker->mux, std::try_to_lock_t{});
      if (!lock.owns_lock()) {
        return error(ChannelTryReceiveErrors::COULD_NOT_LOCK);
      }
 
      if (!broker->inboxes.is_handle_valid(inboxHandle)) {
        return success(Optional<T>{nullopt});
      }
 
      if (auto res = broker->inboxes[inboxHandle].pop(); res.has_value()) {
        broker->senderSignal.notify_all();
        return res;
      }
 
      if (broker->closed) {
        return success(Optional<T>{nullopt});
      }
 
      return error(ChannelTryReceiveErrors::MESSAGE_NOT_READY);
    }

Here is the call graph for this function:

---

The documentation for this struct was generated from the following file:

• pub_sub.hpp