1 #include "hmbdc/Copyright.hpp"
3 #include "hmbdc/app/utils/NetContextUtil.hpp"
4 #include "hmbdc/app/Config.hpp"
5 #include "hmbdc/app/tcpcast/Messages.hpp"
6 #include "hmbdc/app/tcpcast/Sender.hpp"
7 #include "hmbdc/app/tcpcast/SendTransportEngine.hpp"
8 #include "hmbdc/app/tcpcast/RecvTransportEngine.hpp"
9 #include "hmbdc/app/tcpcast/DefaultUserConfig.hpp"
11 #include "hmbdc/comm/Topic.hpp"
12 #include "hmbdc/pattern/GuardedSingleton.hpp"
14 #include <unordered_map>
26 namespace hmbdc {
namespace app {
namespace tcpcast {
52 template <
typename Buffer,
typename MsgArbitrator = RecvTransport::NoOpArb>
68 ,
size_t maxMessageSize) {
69 Config dft(DefaultUserConfig,
"tx");
73 std::lock_guard<std::mutex> tlock(sendTransportEnginesLock_);
74 auto id = cfg.
getExt<std::string>(
"hmbdcName");
75 if (sendTransports_.find(
id) != sendTransports_.end()) {
76 HMBDC_THROW(std::runtime_error
77 ,
"seemingly recreating the same engine, "
78 "same hmbdcNamed engin already created with hmbdcName=" <<
id);
81 sendTransports_.emplace(
id, SendTransport::ptr(res));
90 ,
size_t maxMessageSize
91 , std::tuple<size_t> args) {
109 template <
typename Buffer,
typename MsgArbitrator = RecvTransport::NoOpArb>
113 Config dft(DefaultUserConfig,
"rx");
117 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
119 cfg, buffer, std::forward<MsgArbitrator>(arb));
120 recvTransports_.emplace_back(res);
129 template <
typename Buffer,
typename ArgsTuple>
133 , ArgsTuple&& args) {
148 std::lock_guard<std::mutex> lock(sendersLock_);
149 auto sender = senders_.find(t);
150 if ( sender != senders_.end()) {
151 return sender->second.get();
153 std::lock_guard<std::mutex> slock(sendTransportEnginesLock_);
154 for (
auto& s : sendTransports_) {
157 auto newSender =
new Sender(st, t);
158 senders_[t].reset(newSender);
175 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
176 for (
auto r : recvTransports_) {
187 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
188 for (
auto r : recvTransports_) {
216 template <
typename CcContext>
218 , Config::Base cfgIn = Config::Base{}
219 ,
char const* sendSec = nullptr
220 ,
char const* recvSec = nullptr
221 ,
size_t maxMessageSize = 0
222 , uint8_t runningUsingThreadIndex = 0
223 ,
char const* listenToTopic =
"_")
224 : runningCtx(0u, 2ul) {
225 checkEpollTaskInitialization();
226 createMinimumNetContext(
227 *
this, runningCtx, ctx, cfgIn, sendSec, recvSec, maxMessageSize, runningUsingThreadIndex, listenToTopic);
235 checkEpollTaskInitialization();
242 std::unordered_map<std::string, SendTransport::ptr> sendTransports_;
243 std::mutex sendTransportEnginesLock_;
245 std::vector<RecvTransport::ptr> recvTransports_;
246 std::mutex recvTransportsLock_;
248 std::map<comm::Topic, Sender::ptr> senders_;
249 std::mutex sendersLock_;
void setDefaultUserConfig(Config const &c)
depracated
Definition: Config.hpp:165
class to hold an hmbdc configuration
Definition: Config.hpp:46
NetContext()
this is for users that want finer control of engine creation - from a blank NetContext.
Definition: NetContext.hpp:234
fascade class for sending network messages
Definition: Sender.hpp:12
topic as in the publish / subscribe communication paradigm
Definition: Topic.hpp:14
NetContext(CcContext &ctx, Config::Base cfgIn=Config::Base{}, char const *sendSec=nullptr, char const *recvSec=nullptr, size_t maxMessageSize=0, uint8_t runningUsingThreadIndex=0, char const *listenToTopic="_")
this ctor is to create some commonly used basic engine setup to start with. This is private and only ...
Definition: NetContext.hpp:217
Definition: NetContextUtil.hpp:10
SendTransportEngine * createSendTransportEngine(Config const &cfgIn, size_t maxMessageSize)
construct a send transport enngine (and remember it)
Definition: NetContext.hpp:66
base for the Singleton that works with SingletonGuardian
Definition: GuardedSingleton.hpp:53
RAII representing the lifespan of the underlying Singleton which also ganrantees the singularity of u...
Definition: GuardedSingleton.hpp:20
auto createRecvTransportEngine(Config const &cfgIn, Buffer &buffer, MsgArbitrator &&arb=RecvTransport::NoOpArb())
construct a recv transport and remember it
Definition: NetContext.hpp:110
T getExt(const path_type ¶m, bool throwIfMissing=true) const
get a value from the config
Definition: Config.hpp:225
void listenTo(comm::Topic const &t)
This process is interested in a Topic.
Definition: NetContext.hpp:174
Definition: SendTransportEngine.hpp:139
SendTransportEngine * createSendTransportEngineTuply(Config const &cfg, size_t maxMessageSize, std::tuple< size_t > args)
same as above but provide an unified interface - not preferred
Definition: NetContext.hpp:89
a singleton that holding tcpcast resources
Definition: NetContext.hpp:44
void stopListenTo(comm::Topic const &t)
undo the subscription
Definition: NetContext.hpp:186
A Context is like a media object that facilitates the communications for the Clients that it is holdi...
Definition: Context.hpp:477
a take all arbitrator (no arbitration at all)
Definition: RecvTransportEngine.hpp:38
auto createRecvTransportEngineTuply(Config const &cfg, Buffer &buffer, ArgsTuple &&args)
same as above but to provide a unified interface - not preferred
Definition: NetContext.hpp:130
Sender * getSender(comm::Topic const &t)
get (or create for the first time) a Sender - whose function is to send messages on its associated To...
Definition: NetContext.hpp:147
impl class
Definition: RecvTransportEngine.hpp:61