1 #include "hmbdc/Copyright.hpp" 3 #include "hmbdc/app/utils/NetContextUtil.hpp" 4 #include "hmbdc/app/Config.hpp" 5 #include "hmbdc/app/udpcast/Messages.hpp" 6 #include "hmbdc/app/udpcast/Sender.hpp" 7 #include "hmbdc/app/udpcast/SendTransportEngine.hpp" 8 #include "hmbdc/app/udpcast/RecvTransportEngine.hpp" 9 #include "hmbdc/app/udpcast/DefaultUserConfig.hpp" 10 #include "hmbdc/comm/Topic.hpp" 11 #include "hmbdc/pattern/GuardedSingleton.hpp" 13 #include <unordered_map> 23 namespace hmbdc {
namespace app {
namespace udpcast {
45 template <
typename Buffer,
typename MsgArbitrator = RecvTransport::NoOpArb>
61 ,
size_t maxMessageSize) {
62 Config dft(DefaultUserConfig,
"tx");
66 std::lock_guard<std::mutex> tlock(sendTransportsLock_);
67 auto id = cfg.
getExt<std::string>(
"hmbdcName");
68 if (sendTransports_.find(
id) != sendTransports_.end()) {
69 HMBDC_THROW(std::runtime_error
70 ,
"seemingly recreating the same engine, " 71 "same hmbdcNamed engin already created with hmbdcName=" <<
id);
74 sendTransports_.emplace(
id, SendTransport::ptr(res));
83 ,
size_t maxMessageSize
84 , std::tuple<> args) {
106 template <
typename Buffer,
typename MsgArbitrator = RecvTransport::NoOpArb>
110 Config dft(DefaultUserConfig,
"rx");
114 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
117 , std::forward<MsgArbitrator>(arb));
118 recvTransports_.emplace_back(res);
127 template <
typename Buffer,
typename ArgsTuple>
131 , ArgsTuple&& args) {
146 std::lock_guard<std::mutex> lock(sendersLock_);
147 auto sender = senders_.find(t);
148 if ( sender != senders_.end()) {
149 return sender->second.get();
151 std::lock_guard<std::mutex> slock(sendTransportsLock_);
152 for (
auto& s : sendTransports_) {
155 auto newSender =
new Sender(st, t);
156 senders_[t].reset(newSender);
173 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
174 for (
auto r : recvTransports_) {
185 std::lock_guard<std::mutex> tlock(recvTransportsLock_);
186 for (
auto r : recvTransports_) {
214 template <
typename CcContext>
216 , Config::Base cfgIn = Config::Base{}
217 ,
char const* sendSec = nullptr
218 ,
char const* recvSec = nullptr
219 ,
size_t maxMessageSize = 0
220 , uint8_t runningUsingThreadIndex = 0
221 ,
char const* listenToTopic =
"_")
222 : runningCtx(0u, 2ul) {
223 checkEpollTaskInitialization();
224 createMinimumNetContext(
225 *
this, runningCtx, ctx, cfgIn, sendSec, recvSec, maxMessageSize, runningUsingThreadIndex, listenToTopic);
233 checkEpollTaskInitialization();
239 std::unordered_map<std::string, SendTransport::ptr> sendTransports_;
240 std::mutex sendTransportsLock_;
242 std::vector<RecvTransport::ptr> recvTransports_;
243 std::mutex recvTransportsLock_;
245 std::map<Topic, Sender::ptr> senders_;
246 std::mutex sendersLock_;
T getExt(const path_type ¶m, bool throwIfMissing=true) const
get a value from the config
Definition: Config.hpp:226
void setDefaultUserConfig(Config const &c)
depracated
Definition: Config.hpp:165
class to hold an hmbdc configuration
Definition: Config.hpp:46
a singleton that holding udpcast resources
Definition: NetContext.hpp:37
topic as in the publish / subscribe communication paradigm
Definition: Topic.hpp:14
NetContext()
this is for users that want finer control of engine creation - from a blank NetContext.
Definition: NetContext.hpp:232
a take all arbitrator (no arbitration at all)
Definition: RecvTransportEngine.hpp:33
void listenTo(Topic const &t)
This process is interested in a Topic.
Definition: NetContext.hpp:172
SendTransportEngine * createSendTransportEngine(Config const &cfgIn, size_t maxMessageSize)
construct a send transport engine (and remember it within the class)
Definition: NetContext.hpp:60
Definition: NetContextUtil.hpp:10
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:107
fascade class for sending network messages
Definition: Sender.hpp:15
void stopListenTo(Topic const &t)
undo the subscription
Definition: NetContext.hpp:184
SendTransportEngine * createSendTransportEngineTuply(Config const &cfg, size_t maxMessageSize, std::tuple<> args)
same as above but provide a unified interface - not preferred
Definition: NetContext.hpp:82
Definition: SendTransportEngine.hpp:145
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:215
Definition: RecvTransportEngine.hpp:270
A Context is like a media object that facilitates the communications for the Clients that it is holdi...
Definition: Context.hpp:477
auto createRecvTransportEngineTuply(Config const &cfg, Buffer &buffer, ArgsTuple &&args)
same as above but to provide a unified interface - not preferred
Definition: NetContext.hpp:128
Sender * getSender(Topic const &t=Topic("_"))
get (or create for the first time) a Sender - whose function is to send messages on its associated To...
Definition: NetContext.hpp:145