17 #ifndef _SAG_CONNECTIVITY_PLUGINS_HPP_
18 #define _SAG_CONNECTIVITY_PLUGINS_HPP_
26 #ifndef __STDC_FORMAT_MACROS
27 #define __STDC_FORMAT_MACROS 1
32 namespace softwareag {
33 namespace connectivity {
56 typedef std::unique_ptr<PluginHost> ptr_t;
73 if (
SAG_ERROR_OK != sag_enable_reliability(chain, static_cast<int>(direction))) {
74 throw std::runtime_error(
"An error occurred while setting chain reliability");
81 if (
SAG_ERROR_OK != sag_is_host_shutting_down(chain, isShuttingDown)) {
82 throw std::runtime_error(
"An error occurred while checking if host is shutting down");
90 PluginHost(
void* chain =
nullptr) :chain(chain) {}
136 : chainId(chainId), pluginName(pluginName), config(config.copy()), connectivityManager(connectivityManager), chain(chain)
141 std::string pluginName;
144 void* connectivityManager;
221 sag_delete_user_status_item(connectivityManager, underlying);
222 underlying.item =
nullptr;
232 if (value == lastValue)
return;
235 sag_set_user_status_item(underlying, value.c_str());
248 std::unique_lock<std::mutex> ul(status_lock);
250 setStatus(convert_to_details::integerToString(value));
269 std::unique_lock<std::mutex> ul(status_lock);
270 intValue += incrementValue;
271 setStatus(convert_to_details::integerToString(intValue));
276 const std::string &
key() {
return mkey; }
282 StatusItem(
void* connectivityManager,
const std::string &
key,
const std::string &initialValue,
const int64_t intValue)
283 : intValue(intValue),
285 lastValue(initialValue),
286 connectivityManager(connectivityManager),
287 underlying(sag_create_user_status_item(connectivityManager, key.c_str(), initialValue.c_str()))
289 if (!underlying.item)
291 std::ostringstream oss;
292 oss <<
"Failed to create status item '" << key <<
"' (ensure the key is unique and that this plug-in has not been shutdown already)";
293 throw std::runtime_error(oss.str());
298 const std::string mkey;
299 std::string lastValue;
300 void* connectivityManager;
301 sag_status_item_t underlying;
302 std::mutex status_lock;
318 std::unique_ptr<StatusItem>
createStatusItem(
const std::string &key,
const std::string &initialValue)
320 return std::unique_ptr<StatusItem>(
new StatusItem(connectivityManager, key, initialValue, 0));
338 std::ostringstream oss;
340 return std::unique_ptr<StatusItem>(
new StatusItem(connectivityManager, key, oss.str(), initialValue));
355 sag_set_user_status_map(connectivityManager, reinterpret_cast<const sag_underlying_map_t&>(statusmap));
357 for (
auto it = statusmap.
cbegin(); it != statusmap.
cend(); it++)
360 if (it->second.empty())
361 mapKeysToCleanup.
erase(it->first);
363 mapKeysToCleanup.
insert(it->first.copy(), std::move(
data_t()));
374 if (!mapKeysToCleanup.
empty())
375 sag_set_user_status_map(connectivityManager, reinterpret_cast<const sag_underlying_map_t&>(mapKeysToCleanup));
381 explicit StatusReporter(
void *connectivityManager) : connectivityManager(connectivityManager), mapKeysToCleanup()
386 void* connectivityManager;
387 map_t mapKeysToCleanup;
390 StatusReporter() =
delete;
393 StatusReporter(
const StatusReporter& other) =
delete;
394 StatusReporter& operator=(
const StatusReporter&) =
delete;
414 logger.
info(
"The %s connectivity plug-in in chain '%s' is using the deprecated legacy (std::string&,std::string&,map_t&) constructor. We recommend changing to use the recommended (PluginConstructorParameters&) constructor instead.",
415 pluginName.c_str(), chainId.c_str());
505 if (statusReporter)
return *statusReporter;
506 throw std::runtime_error(
"Cannot call getStatusReporter when using the legacy constructor");
522 std::unique_ptr<StatusReporter> statusReporter;
535 typedef std::auto_ptr<HostSide>
ptr_t;
562 RemoteHostSide(sag_plugin_t other, sag_send_fn_t fn): other(other), fn(fn) {}
580 typedef std::auto_ptr<TransportSide>
ptr_t;
661 :
Plugin(pluginName, chainId, config)
752 :
Plugin(name, chainId, config)
838 for (
Message *it = start; it != end; ++it) {
840 if (it->getPayload().empty()) {
880 }
catch (
const std::exception &e) {
881 logger.
warn(
"Caught exception delivering message: %s", e.what());
883 logger.
warn(
"Caught unknown exception delivering message: %s");
947 for (
Message *it = start; it != end; ++it) {
950 if (it->getPayload().empty()) {
960 if (it != curr) it->swap(std::move(*curr));
974 for (
Message *it = start; it != end; ++it) {
978 if (it->getPayload().empty()) {
988 if (it != curr) it->swap(std::move(*curr));
1044 }
catch (
const std::exception &e) {
1045 logger.
warn(
"Caught exception transforming message: %s", e.what());
1047 logger.
warn(
"Caught unknown exception transforming message");
1056 #include <sag_internal/exception.hpp>
1057 #include <sag_internal/remote_plugins.hpp>
1058 #include <sag_internal/plugin_macros.hpp>
1069 #define SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_TRANSPORT_CLASS(Class)
1082 #define SAG_DECLARE_CONNECTIVITY_TRANSPORT(Class) _SAG_DECLARE_CONNECTIVITY_TRANSPORT_LEGACY(Class)
1093 #define SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class) _SAG_DECLARE_CONNECTIVITY_CODEC_CLASS(Class)
1105 #define SAG_DECLARE_CONNECTIVITY_CODEC(Class) _SAG_DECLARE_CONNECTIVITY_CODEC_LEGACY(Class)
1107 #endif // _SAG_CONNECTIVITY_PLUGINS_HPP_
A container for parameters passed to the constructor of a codec plug-in.
Definition: sag_connectivity_plugins.hpp:162
Logger LOGGER
Legacy logging for writing to the host log file [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:515
bool empty() const
Returns true if the map is empty (size() == 0)
Definition: sag_connectivity_cpp.hpp:289
const std::string pluginName
The name used for this plug-in in the configuration file.
Definition: sag_connectivity_plugins.hpp:487
Base class that simplifies implementation of codec plug-ins that deal only with individual messages n...
Definition: sag_connectivity_plugins.hpp:906
AbstractTransport(const PluginConstructorParameters::TransportConstructorParameters ¶ms)
Constructor.
Definition: sag_connectivity_plugins.hpp:763
Base class that simplifies implementation of transport plug-ins that deal only with individual messag...
Definition: sag_connectivity_plugins.hpp:797
Wrap the next plugin in order as a TransportSide object.
Definition: sag_connectivity_plugins.hpp:604
void increment(int64_t incrementValue=1)
Set an integer status value by incrementing the previous integer value that was set by this object...
Definition: sag_connectivity_plugins.hpp:267
HostSide::ptr_t hostSide
The next plug-in in the chain towards host.
Definition: sag_connectivity_plugins.hpp:692
An interface to the next component (plugin or host) towards the host.
Definition: sag_connectivity_plugins.hpp:529
virtual void sendBatchTowardsHost(Message *start, Message *end)=0
Abstract method that must be implemented to handle delivery of a batch of messages heading towards th...
map_t config
The configuration of this plug-in.
Definition: sag_connectivity_plugins.hpp:491
Direction
The enumeration indicating the direction of message flow - towards the transport or towards the host...
Definition: sag_connectivity_plugins.hpp:38
void warn(const char *format, ARGS...args) const
Log a message at WARN level.
Definition: sag_plugin_logging.hpp:153
Class for writing to the system logger.
Definition: sag_plugin_logging.hpp:72
const PluginHost::ptr_t host
Interface to support miscellaneous requests from this plug-in to the host system. ...
Definition: sag_connectivity_plugins.hpp:497
StatusReporter & getStatusReporter()
Allows reporting status information from this plug-in, such as connected/disconnected status and numb...
Definition: sag_connectivity_plugins.hpp:504
A map class which implements many of the functions on std::map.
Definition: map_t.hpp:36
Contains the C++ implementation of the underlying datatypes used by connectivity plugins and their ac...
void setStatus(const map_t &statusmap)
Set multiple related string status values at the same time (atomically).
Definition: sag_connectivity_plugins.hpp:353
virtual void sendBatchTowardsTransport(Message *start, Message *end)
Implements batch sending, calling deliverMessageTowardsTransport(Message&) for each message individua...
Definition: sag_connectivity_plugins.hpp:836
Base of the inheritance tree for Connectivity plugins.
Definition: sag_connectivity_plugins.hpp:405
TransportSide::ptr_t transportSide
The next plug-in in the chain towards transport.
Definition: sag_connectivity_plugins.hpp:698
virtual void sendBatchTowardsTransport(Message *start, Message *end)=0
Abstract method that must be implemented to handle delivery of a batch of messages heading towards th...
void setStatus(const std::string &value)
Set a string status value.
Definition: sag_connectivity_plugins.hpp:230
Definition: sag_connectivity_threading.h:178
std::auto_ptr< TransportSide > ptr_t
Pointers to TransportSides should always be this ptr_t type, which is a std::auto_ptr.
Definition: sag_connectivity_plugins.hpp:580
AbstractSimpleTransport(const PluginConstructorParameters::TransportConstructorParameters ¶ms)
Constructor.
Definition: sag_connectivity_plugins.hpp:827
Logger logger
Logging for writing to the host log file.
Definition: sag_connectivity_plugins.hpp:520
void info(const char *format, ARGS...args) const
Log a message at INFO level.
Definition: sag_plugin_logging.hpp:155
virtual bool transformNullPayloadTowardsTransport(Message &msg)
Transform a message with a null payload in a transport-wards direction.
Definition: sag_connectivity_plugins.hpp:1014
virtual void start()
Called when an entire chain has been created and the plugin is allowed to start up (after all plugins...
Definition: sag_connectivity_plugins.hpp:445
std::unique_ptr< StatusItem > createStatusItem(const std::string &key, const std::string &initialValue)
Creates a StatusItem instance that can be used to report status for a given key.
Definition: sag_connectivity_plugins.hpp:318
A class allowing a plug-in to report status values to the host.
Definition: sag_connectivity_plugins.hpp:187
size_t erase(const data_t &k)
Remove the item with the specified key.
virtual ~TransportSide()
Ensure virtual destruction.
Definition: sag_connectivity_plugins.hpp:578
Base class for transport plug-ins.
Definition: sag_connectivity_plugins.hpp:732
static const char * STATUS_ONLINE()
Returns a constant that should be used as the status value when a component is online, operational, connected, and ready to handles messages.
Definition: sag_connectivity_plugins.hpp:194
const_iterator cbegin() const
Forward const_iterator begin.
Definition: sag_connectivity_cpp.hpp:281
AbstractSimpleTransport(const std::string &name, const std::string &chainId, const map_t &config)
Legacy constructor [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:815
bool isShuttingDown()
Check if host is shutting down.
Definition: sag_connectivity_plugins.hpp:79
const_iterator cend() const
Forward const_iterator end.
Definition: sag_connectivity_cpp.hpp:283
virtual void handleException(Message &m)
Handle an exception thrown while delivering a message towards the transport.
Definition: sag_connectivity_plugins.hpp:876
Wrap the next plugin in order as a HostSide object.
Definition: sag_connectivity_plugins.hpp:559
const std::string name
The name used for this plug-in in the configuration file [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:484
static const char * STATUS_FAILED()
Returns a constant that should be used as the status value when a component is not currently operatio...
Definition: sag_connectivity_plugins.hpp:202
virtual ~HostSide()
Ensure virtual destruction.
Definition: sag_connectivity_plugins.hpp:533
AbstractSimpleCodec(const PluginConstructorParameters::CodecConstructorParameters ¶ms)
Constructor.
Definition: sag_connectivity_plugins.hpp:935
virtual void hostReady()
Called some time after start(), when the host is ready to start receiving input (sends will be queued...
Definition: sag_connectivity_plugins.hpp:454
std::pair< iterator, bool > insert(const std::pair< data_t &&, data_t && > &v)
Insert a new key/value pair into the map.
Base class for codec plug-ins.
Definition: sag_connectivity_plugins.hpp:640
No error.
Definition: sag_connectivity_c.h:62
static const char * STATUS_STARTING()
Returns a constant that should be used as the status value when a component is still starting...
Definition: sag_connectivity_plugins.hpp:198
std::unique_ptr< StatusItem > createStatusItem(const std::string &key, int64_t initialValue)
Creates a StatusItem instance that can be used to report status for a given key, using an integral in...
Definition: sag_connectivity_plugins.hpp:336
virtual bool transformMessageTowardsTransport(Message &msg)=0
Abstract method that must be implemented to handle transformation of an individual message in a trans...
AbstractSimpleCodec(const std::string &name, const std::string &chainId, const map_t &config)
Legacy constructor [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:923
void setStatus(int64_t value)
Set an integer status value.
Definition: sag_connectivity_plugins.hpp:246
virtual bool transformMessageTowardsHost(Message &msg)=0
Abstract method that must be implemented to handle transformation of an individual message...
A map class which implements many of the functions on std::map.
Definition: sag_connectivity_cpp.hpp:37
const std::string & key()
Get the unique key specified when this status item was created.
Definition: sag_connectivity_plugins.hpp:276
const std::string & getPluginName() const
Get the name used in the configuration file for this plug-in.
Definition: sag_connectivity_plugins.hpp:129
A class that can be used to efficiently update the value associated with a single status key...
Definition: sag_connectivity_plugins.hpp:214
A container for an payload and associated metadata.
Definition: sag_connectivity_cpp.hpp:28
virtual void shutdown()
Stop processing messages and terminate and join any background threads.
Definition: sag_connectivity_plugins.hpp:478
virtual void sendBatchTowardsHost(Message *start, Message *end)
Implements batch sending, calling transformMessageTowardsHost(Message &) for each message individuall...
Definition: sag_connectivity_plugins.hpp:944
An interface to the next component (plugin or host) towards the transport.
Definition: sag_connectivity_plugins.hpp:574
AbstractCodec(const PluginConstructorParameters::CodecConstructorParameters ¶ms)
Constructor.
Definition: sag_connectivity_plugins.hpp:672
virtual void deliverMessageTowardsTransport(Message &msg)=0
Abstract method that must be implemented to handle delivery of an individual message.
virtual void deliverNullPayloadTowardsTransport(Message &msg)
Deliver a message with a null payload.
Definition: sag_connectivity_plugins.hpp:853
void enableReliability(Direction direction)
Enable reliable messaging for the chain that this plug-in belongs to, in a particular direction i...
Definition: sag_connectivity_plugins.hpp:72
Contains the headers needed to implement your own Connectivity Plugins.
AbstractCodec(const std::string &pluginName, const std::string &chainId, const map_t &config)
Legacy constructor [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:660
A container for parameters passed to the constructor of a transport plug-in.
Definition: sag_connectivity_plugins.hpp:150
The direction of messages flowing towards the host (from the transport).
map_t copy() const
Return a deep copy of this map.
Definition: sag_connectivity_cpp.hpp:210
AbstractTransport(const std::string &name, const std::string &chainId, const map_t &config)
Legacy constructor [DEPRECATED].
Definition: sag_connectivity_plugins.hpp:751
const std::string & getChainId() const
Get the identifier used for the chain this plug-in is part of.
Definition: sag_connectivity_plugins.hpp:124
Interface to support miscellaneous requests from a particular plug-in to the host system...
Definition: sag_connectivity_plugins.hpp:53
A base interface for parameters passed to the constructor of transport or codec plug-ins.
Definition: sag_connectivity_plugins.hpp:106
virtual bool handleException(Message &m, bool towardsTransport)
Handle an exception thrown while delivering a message.
Definition: sag_connectivity_plugins.hpp:1040
utf8-encoded const char*
Definition: sag_connectivity_c.h:46
virtual void sendBatchTowardsHost(Message *start, Message *end)
Abstract method that must be implemented to handle delivery of a batch of messages heading towards th...
const std::string chainId
The identifier used for the chain this plug-in is part of.
Definition: sag_connectivity_plugins.hpp:489
The direction of messages flowing towards the transport (from the host).
std::string getStatus()
Return the value this status item was set to most recently by this class.
Definition: sag_connectivity_plugins.hpp:258
A variant type which can be one of the following:
Definition: sag_connectivity_cpp.hpp:43
const map_t & getConfig() const
Get the configuration for this plug-in.
Definition: sag_connectivity_plugins.hpp:119
virtual void sendBatchTowardsTransport(Message *start, Message *end)
Implements batch sending, calling transformMessageTowardsTransport(Message &) for each message indivi...
Definition: sag_connectivity_plugins.hpp:971
std::auto_ptr< HostSide > ptr_t
Pointers to HostSides should always be this ptr_t type, which is a std::auto_ptr. ...
Definition: sag_connectivity_plugins.hpp:535
virtual ~Plugin()
This destructor must be virtual.
Definition: sag_connectivity_plugins.hpp:433
HostSide::ptr_t hostSide
The next plug-in in the chain towards host.
Definition: sag_connectivity_plugins.hpp:779
virtual void sendBatchTowardsTransport(Message *start, Message *end)
Abstract method that must be implemented to handle delivery of a batch of messages heading towards th...
const std::string & getName() const
Return the name of this plugin instance.
Definition: sag_connectivity_plugins.hpp:480
virtual bool transformNullPayloadTowardsHost(Message &msg)
Transform a message with a null payload in a host-wards direction.
Definition: sag_connectivity_plugins.hpp:1006