You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celix.apache.org by er...@apache.org on 2018/12/01 08:27:00 UTC
[1/3] celix git commit: Nanomsg
Repository: celix
Updated Branches:
refs/heads/nanomsg 7c141424d -> 707b8e54a
Nanomsg
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/15f268d1
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/15f268d1
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/15f268d1
Branch: refs/heads/nanomsg
Commit: 15f268d131370e55dc2dd4c5b6c247baa522fc58
Parents: 7c14142
Author: Erjan Altena <er...@gmail.com>
Authored: Wed Nov 28 21:07:00 2018 +0100
Committer: Erjan Altena <er...@gmail.com>
Committed: Wed Nov 28 21:37:36 2018 +0100
----------------------------------------------------------------------
.../src/pubsub_nanomsg_admin.cc | 12 +-
.../src/pubsub_nanomsg_common.cc | 15 +-
.../src/pubsub_nanomsg_common.h | 2 +-
.../src/pubsub_nanomsg_topic_receiver.cc | 5 +-
.../src/pubsub_nanomsg_topic_receiver.h | 2 +-
.../src/pubsub_nanomsg_topic_sender.cc | 211 +++++++------------
.../src/pubsub_nanomsg_topic_sender.h | 54 +++--
7 files changed, 138 insertions(+), 163 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
index 42ed632..9fdf2a3 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
@@ -341,7 +341,7 @@ celix_status_t pubsub_nanomsg_admin::setupTopicSender(const char *scope, const c
const char *serType = serEntry->serType;
newEndpoint = pubsubEndpoint_create(fwUUID, scope, topic, PUBSUB_PUBLISHER_ENDPOINT_TYPE, psaType, serType,
nullptr);
- celix_properties_set(newEndpoint, PUBSUB_NANOMSG_URL_KEY, sender->getUrl());
+ celix_properties_set(newEndpoint, PUBSUB_NANOMSG_URL_KEY, sender->getUrl().c_str());
//if available also set container name
const char *cn = celix_bundleContext_getProperty(ctx, "CELIX_CONTAINER_NAME", nullptr);
if (cn != nullptr) {
@@ -568,12 +568,12 @@ celix_status_t pubsub_nanomsg_admin::executeCommand(char *commandLine __attribut
long serSvcId = sender->getSerializerSvcId();
auto kvs = serializers.map.find(serSvcId);
const char* serType = ( kvs == serializers.map.end() ? "!Error" : kvs->second.serType);
- const char *scope = sender->getScope();
- const char *topic = sender->getTopic();
- const char *url = sender->getUrl();
- fprintf(out, "|- Topic Sender %s/%s\n", scope, topic);
+ const auto scope = sender->getScope();
+ const auto topic = sender->getTopic();
+ const auto url = sender->getUrl();
+ fprintf(out, "|- Topic Sender %s/%s\n", scope.c_str(), topic.c_str());
fprintf(out, " |- serializer type = %s\n", serType);
- fprintf(out, " |- url = %s\n", url);
+ fprintf(out, " |- url = %s\n", url.c_str());
}
}
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.cc
index 3ecd19c..333cb2d 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.cc
@@ -41,16 +41,15 @@ bool psa_nanomsg_checkVersion(version_pt msgVersion, const pubsub_nanmosg_msg_he
return check;
}
-void psa_nanomsg_setScopeAndTopicFilter(const std::string &scope, const std::string &topic, char *filter) {
- for (int i = 0; i < 5; ++i) { // 5 ??
- filter[i] = '\0';
- }
+std::string psa_nanomsg_setScopeAndTopicFilter(const std::string &scope, const std::string &topic) {
+ std::string result("");
if (scope.size() >= 2) { //3 ??
- filter[0] = scope[0];
- filter[1] = scope[1];
+ result += scope[0];
+ result += scope[1];
}
if (topic.size() >= 2) { //3 ??
- filter[2] = topic[0];
- filter[3] = topic[1];
+ result += topic[0];
+ result += topic[1];
}
+ return result;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.h
index 276169f..2c03d4c 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_common.h
@@ -49,7 +49,7 @@ typedef struct pubsub_nanomsg_msg_header pubsub_nanmosg_msg_header_t;
int psa_nanoMsg_localMsgTypeIdForMsgType(void *handle, const char *msgType, unsigned int *msgTypeId);
-void psa_nanomsg_setScopeAndTopicFilter(const std::string &scope, const std::string &topic, char *filter);
+std::string psa_nanomsg_setScopeAndTopicFilter(const std::string &scope, const std::string &topic);
bool psa_nanomsg_checkVersion(version_pt msgVersion, const pubsub_nanmosg_msg_header_t *hdr);
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
index 9f77a4c..30c2af7 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
@@ -72,7 +72,7 @@ pubsub::nanomsg::topic_receiver::topic_receiver(celix_bundle_context_t *_ctx,
ctx = _ctx;
logHelper = _logHelper;
serializer = _serializer;
- psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic, m_scopeAndTopicFilter);
+ m_scopeAndTopicFilter = psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic);
m_nanoMsgSocket = nn_socket(AF_SP, NN_BUS);
if (m_nanoMsgSocket < 0) {
@@ -86,8 +86,7 @@ pubsub::nanomsg::topic_receiver::topic_receiver(celix_bundle_context_t *_ctx,
std::bad_alloc{};
}
- char subscriberFilter[5]; // 5 ??
- psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic, subscriberFilter);
+ auto subscriberFilter = psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic);
int size = snprintf(NULL, 0, "(%s=%s)", PUBSUB_SUBSCRIBER_TOPIC, m_topic.c_str());
char buf[size + 1];
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
index 2519e4a..02c462e 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
@@ -100,7 +100,7 @@ namespace pubsub {
pubsub_serializer_service_t *serializer{nullptr};
const std::string m_scope{};
const std::string m_topic{};
- char m_scopeAndTopicFilter[5];
+ std::string m_scopeAndTopicFilter{};
int m_nanoMsgSocket{0};
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
index 1c75e71..9a253b2 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
@@ -18,6 +18,7 @@
*/
#include <memory.h>
+#include <iostream>
#include <stdlib.h>
#include <utils.h>
@@ -29,7 +30,6 @@
#include <pubsub_constants.h>
-#include <pubsub/publisher.h>
#include <pubsub_common.h>
#include "pubsub_nanomsg_topic_sender.h"
#include "pubsub_psa_nanomsg_constants.h"
@@ -48,23 +48,9 @@
logHelper_log(logHelper, OSGI_LOGSERVICE_ERROR, __VA_ARGS__)
-typedef struct psa_nanomsg_bounded_service_entry {
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *parent;
- pubsub_publisher_t service;
- long bndId;
- hash_map_t *msgTypes;
- int getCount;
-} psa_nanomsg_bounded_service_entry_t;
-
-//static void* psa_nanomsg_getPublisherService(void *handle, const celix_bundle_t *requestingBundle,
-// const celix_properties_t *svcProperties);
-//static void psa_nanomsg_ungetPublisherService(void *handle, const celix_bundle_t *requestingBundle,
-// const celix_properties_t *svcProperties);
static unsigned int rand_range(unsigned int min, unsigned int max);
static void delay_first_send_for_late_joiners(pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender);
-static int psa_nanomsg_topicPublicationSend(void *handle, unsigned int msgTypeId, const void *msg);
-
pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_bundle_context_t *_ctx,
log_helper_t *_logHelper,
const char *_scope,
@@ -77,53 +63,44 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_
ctx{_ctx},
logHelper{_logHelper},
serializerSvcId {_serializerSvcId},
- serializer{_ser}{
+ serializer{_ser},
+ scope{_scope},
+ topic{_topic}{
- psa_nanomsg_setScopeAndTopicFilter(_scope, _topic, scopeAndTopicFilter);
+ scopeAndTopicFilter = psa_nanomsg_setScopeAndTopicFilter(_scope, _topic);
//setting up nanomsg socket for nanomsg TopicSender
- {
-
- int socket = nn_socket(AF_SP, NN_BUS);
- if (socket == -1) {
- perror("Error for nanomsg_socket");
- }
+ int socket = nn_socket(AF_SP, NN_BUS);
+ if (socket == -1) {
+ perror("Error for nanomsg_socket");
+ }
- int rv = -1, retry=0;
- while(rv == -1 && retry < NANOMSG_BIND_MAX_RETRY ) {
- /* Randomized part due to same bundle publishing on different topics */
- unsigned int port = rand_range(_basePort,_maxPort);
- size_t len = (size_t)snprintf(NULL, 0, "tcp://%s:%u", _bindIp, port) + 1;
- char *_url = static_cast<char*>(calloc(len, sizeof(char*)));
- snprintf(_url, len, "tcp://%s:%u", _bindIp, port);
-
- len = (size_t)snprintf(NULL, 0, "tcp://0.0.0.0:%u", port) + 1;
- char *bindUrl = static_cast<char*>(calloc(len, sizeof(char)));
- snprintf(bindUrl, len, "tcp://0.0.0.0:%u", port);
- rv = nn_bind (socket, bindUrl);
- if (rv == -1) {
- perror("Error for nn_bind");
- free(_url);
- } else {
- this->url = _url;
- nanomsg.socket = socket;
- }
- retry++;
- free(bindUrl);
+ int rv = -1, retry=0;
+ while(rv == -1 && retry < NANOMSG_BIND_MAX_RETRY ) {
+ /* Randomized part due to same bundle publishing on different topics */
+ unsigned int port = rand_range(_basePort,_maxPort);
+ size_t len = (size_t)snprintf(nullptr, 0, "tcp://%s:%u", _bindIp, port) + 1;
+ char *_url = static_cast<char*>(calloc(len, sizeof(char*)));
+ snprintf(_url, len, "tcp://%s:%u", _bindIp, port);
+
+ len = (size_t)snprintf(nullptr, 0, "tcp://0.0.0.0:%u", port) + 1;
+ char *bindUrl = static_cast<char*>(calloc(len, sizeof(char)));
+ snprintf(bindUrl, len, "tcp://0.0.0.0:%u", port);
+ rv = nn_bind (socket, bindUrl);
+ if (rv == -1) {
+ perror("Error for nn_bind");
+ free(_url);
+ } else {
+ this->url = _url;
+ nanomsg.socket = socket;
}
+ retry++;
+ free(bindUrl);
}
- if (url != NULL) {
- scope = strndup(_scope, 1024 * 1024);
- topic = strndup(_topic, 1024 * 1024);
-
- celixThreadMutex_create(&boundedServices.mutex, NULL);
- celixThreadMutex_create(&nanomsg.mutex, NULL);
- boundedServices.map = hashMap_create(NULL, NULL, NULL, NULL);
- }
+ if (!url.empty()) {
- //register publisher services using a service factory
- if (url != NULL) {
+ //register publisher services using a service factory
publisher.factory.handle = this;
publisher.factory.getService = [](void *handle, const celix_bundle_t *requestingBundle, const celix_properties_t *svcProperties) {
return static_cast<pubsub::nanomsg::pubsub_nanomsg_topic_sender*>(handle)->getPublisherService(
@@ -137,8 +114,8 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_
};
celix_properties_t *props = celix_properties_create();
- celix_properties_set(props, PUBSUB_PUBLISHER_TOPIC, topic);
- celix_properties_set(props, PUBSUB_PUBLISHER_SCOPE, scope);
+ celix_properties_set(props, PUBSUB_PUBLISHER_TOPIC, topic.c_str());
+ celix_properties_set(props, PUBSUB_PUBLISHER_SCOPE, scope.c_str());
celix_service_registration_options_t opts = CELIX_EMPTY_SERVICE_REGISTRATION_OPTIONS;
opts.factory = &publisher.factory;
@@ -156,39 +133,27 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::~pubsub_nanomsg_topic_sender() {
nn_close(nanomsg.socket);
- celixThreadMutex_lock(&boundedServices.mutex);
- hash_map_iterator_t iter = hashMapIterator_construct(boundedServices.map);
- while (hashMapIterator_hasNext(&iter)) {
- psa_nanomsg_bounded_service_entry_t *entry = static_cast<psa_nanomsg_bounded_service_entry_t*>(hashMapIterator_nextValue(&iter));
- if (entry != NULL) {
- serializer->destroySerializerMap(serializer->handle, entry->msgTypes);
- free(entry);
- }
+ std::lock_guard<std::mutex> lock(boundedServices.mutex);
+ for (auto &it: boundedServices.map) {
+ serializer->destroySerializerMap(serializer->handle, it.second.msgTypes);
}
- hashMap_destroy(boundedServices.map, false, false);
- celixThreadMutex_unlock(&boundedServices.mutex);
-
- celixThreadMutex_destroy(&boundedServices.mutex);
- celixThreadMutex_destroy(&nanomsg.mutex);
+ boundedServices.map.clear();
- free(scope);
- free(topic);
- free(url);
}
long pubsub::nanomsg::pubsub_nanomsg_topic_sender::getSerializerSvcId() const {
return serializerSvcId;
}
-const char* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getScope() const {
+const std::string &pubsub::nanomsg::pubsub_nanomsg_topic_sender::getScope() const {
return scope;
}
-const char* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getTopic() const {
+const std::string &pubsub::nanomsg::pubsub_nanomsg_topic_sender::getTopic() const {
return topic;
}
-const char* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getUrl() const {
+const std::string &pubsub::nanomsg::pubsub_nanomsg_topic_sender::getUrl() const {
return url;
}
@@ -196,64 +161,57 @@ const char* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getUrl() const {
void* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getPublisherService(const celix_bundle_t *requestingBundle,
const celix_properties_t *svcProperties __attribute__((unused))) {
long bndId = celix_bundle_getId(requestingBundle);
-
- celixThreadMutex_lock(&boundedServices.mutex);
- psa_nanomsg_bounded_service_entry_t *entry = static_cast<psa_nanomsg_bounded_service_entry_t*>(hashMap_get(boundedServices.map, (void*)bndId));
- if (entry != NULL) {
- entry->getCount += 1;
+ void *service{nullptr};
+ std::lock_guard<std::mutex> lock(boundedServices.mutex);
+ auto existingEntry = boundedServices.map.find(bndId);
+ if (existingEntry != boundedServices.map.end()) {
+ existingEntry->second.getCount += 1;
+ service = &existingEntry->second.service;
} else {
- entry = static_cast<psa_nanomsg_bounded_service_entry_t*>(calloc(1, sizeof(*entry)));
- entry->getCount = 1;
- entry->parent = this;
- entry->bndId = bndId;
+ auto entry = boundedServices.map.emplace(std::piecewise_construct,
+ std::forward_as_tuple(bndId),
+ std::forward_as_tuple(this, bndId, logHelper));
+ int rc = serializer->createSerializerMap(serializer->handle, (celix_bundle_t*)requestingBundle, &entry.first->second.msgTypes);
- int rc = serializer->createSerializerMap(serializer->handle, (celix_bundle_t*)requestingBundle, &entry->msgTypes);
if (rc == 0) {
- entry->service.handle = entry;
- entry->service.localMsgTypeIdForMsgType = psa_nanoMsg_localMsgTypeIdForMsgType;
- entry->service.send = psa_nanomsg_topicPublicationSend;
- entry->service.sendMultipart = NULL; //not supported TODO remove
- hashMap_put(boundedServices.map, (void*)bndId, entry);
+ entry.first->second.service.handle = &entry.first->second;
+ entry.first->second.service.localMsgTypeIdForMsgType = psa_nanoMsg_localMsgTypeIdForMsgType;
+ entry.first->second.service.send = [](void *handle, unsigned int msgTypeId, const void *msg) {
+ return static_cast<pubsub::nanomsg::bounded_service_entry*>(handle)->topicPublicationSend(msgTypeId, msg);
+ };
+ entry.first->second.service.sendMultipart = NULL; //not supported TODO remove
+ service = &entry.first->second.service;
} else {
+ boundedServices.map.erase(bndId);
L_ERROR("Error creating serializer map for NanoMsg TopicSender %s/%s", scope, topic);
}
-
-
-
}
- celixThreadMutex_unlock(&boundedServices.mutex);
- return &entry->service;
+ return service;
}
void pubsub::nanomsg::pubsub_nanomsg_topic_sender::ungetPublisherService(const celix_bundle_t *requestingBundle,
- const celix_properties_t *svcProperties __attribute__((unused))) {
+ const celix_properties_t */*svcProperties*/) {
long bndId = celix_bundle_getId(requestingBundle);
- celixThreadMutex_lock(&boundedServices.mutex);
- psa_nanomsg_bounded_service_entry_t *entry = static_cast<psa_nanomsg_bounded_service_entry_t*>(hashMap_get(boundedServices.map, (void*)bndId));
- if (entry != NULL) {
- entry->getCount -= 1;
- }
- if (entry != NULL && entry->getCount == 0) {
- //free entry
- hashMap_remove(boundedServices.map, (void*)bndId);
- int rc = serializer->destroySerializerMap(serializer->handle, entry->msgTypes);
- if (rc != 0) {
- L_ERROR("Error destroying publisher service, serializer not available / cannot get msg serializer map\n");
+ std::lock_guard<std::mutex> lock(boundedServices.mutex);
+ auto entry = boundedServices.map.find(bndId);
+ if (entry != boundedServices.map.end()) {
+ entry->second.getCount -= 1;
+ if (entry->second.getCount == 0) {
+ int rc = serializer->destroySerializerMap(serializer->handle, entry->second.msgTypes);
+ if (rc != 0) {
+ L_ERROR("Error destroying publisher service, serializer not available / cannot get msg serializer map\n");
+ }
+ boundedServices.map.erase(bndId);
}
-
- free(entry);
}
- celixThreadMutex_unlock(&boundedServices.mutex);
}
-static int psa_nanomsg_topicPublicationSend(void *handle, unsigned int msgTypeId, const void *inMsg) {
- int status = CELIX_SUCCESS;
- psa_nanomsg_bounded_service_entry_t *bound = static_cast<psa_nanomsg_bounded_service_entry_t*>(handle);
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = bound->parent;
-
- pubsub_msg_serializer_t* msgSer = static_cast<pubsub_msg_serializer_t*>(hashMap_get(bound->msgTypes, (void*)(uintptr_t)msgTypeId));
+int pubsub::nanomsg::bounded_service_entry::topicPublicationSend(unsigned int msgTypeId, const void *inMsg) {
+ int status;
+ pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = parent;
+ pubsub_msg_serializer_t* msgSer = static_cast<pubsub_msg_serializer_t*>(hashMap_get(msgTypes, (void*)(uintptr_t)msgTypeId));
if (msgSer != NULL) {
delay_first_send_for_late_joiners(sender);
@@ -285,37 +243,32 @@ static int psa_nanomsg_topicPublicationSend(void *handle, unsigned int msgTypeId
msg.msg_iov[1].iov_len = serializedOutputLen;
msg.msg_control = nullptr;
msg.msg_controllen = 0;
- //zmsg_t *msg = zmsg_new();
- //TODO revert to use zmq_msg_init_data (or something like that) for zero copy for the payload
- //TODO remove socket mutex .. not needed (initialized during creation)
- //zmsg_addstr(msg, sender->scopeAndTopicFilter);
- //zmsg_addmem(msg, &msg_hdr, sizeof(msg_hdr));
- //zmsg_addmem(msg, serializedOutput, );
errno = 0;
int rc = nn_sendmsg(sender->nanomsg.socket, &msg, 0 );
free(serializedOutput);
if (rc < 0) {
- //TODO L_WARN("[PSA_ZMQ_TS] Error sending zmsg, rc is %i. %s", rc, strerror(errno));
+ L_WARN("[PSA_ZMQ_TS] Error sending zmsg, rc is %i. %s", rc, strerror(errno));
} else {
- //TODO L_INFO("[PSA_ZMQ_TS] Send message with size %d\n", rc);
- //TODO L_INFO("[PSA_ZMQ_TS] Send message ID %d, major %d, minor %d\n", msg_hdr.type, (int)msg_hdr.major, (int)msg_hdr.minor);
+ L_INFO("[PSA_ZMQ_TS] Send message with size %d\n", rc);
+ L_INFO("[PSA_ZMQ_TS] Send message ID %d, major %d, minor %d\n", msg_hdr.type, (int)msg_hdr.major, (int)msg_hdr.minor);
}
} else {
- //TODO L_WARN("[PSA_ZMQ_TS] Error serialize message of type %s for scope/topic %s/%s", msgSer->msgName, sender->scope, sender->topic);
+ L_WARN("[PSA_ZMQ_TS] Error serialize message of type %s for scope/topic %s/%s", msgSer->msgName, sender->scope, sender->topic);
}
} else {
status = CELIX_SERVICE_EXCEPTION;
- //TODO L_WARN("[PSA_ZMQ_TS] Error cannot serialize message with msg type id %i for scope/topic %s/%s", msgTypeId, sender->scope, sender->topic);
+ L_WARN("[PSA_ZMQ_TS] Error cannot serialize message with msg type id %i for scope/topic %s/%s", msgTypeId, sender->scope, sender->topic);
}
return status;
}
-static void delay_first_send_for_late_joiners(pubsub::nanomsg::pubsub_nanomsg_topic_sender */*sender*/) {
+static void delay_first_send_for_late_joiners(pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender) {
static bool firstSend = true;
if(firstSend){
- //TODO L_INFO("PSA_UDP_MC_TP: Delaying first send for late joiners...\n");
+ auto logHelper = sender->logHelper;
+ L_INFO("PSA_UDP_MC_TP: Delaying first send for late joiners...\n");
sleep(FIRST_SEND_DELAY_IN_SECONDS);
firstSend = false;
}
http://git-wip-us.apache.org/repos/asf/celix/blob/15f268d1/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
index 90ab6ce..1883a41 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
@@ -18,13 +18,37 @@
*/
#ifndef CELIX_PUBSUB_ZMQ_TOPIC_SENDER_H
#define CELIX_PUBSUB_ZMQ_TOPIC_SENDER_H
-
+#include <mutex>
+#include <map>
#include "celix_bundle_context.h"
#include <log_helper.h>
#include <pubsub_serializer.h>
+#include <pubsub/publisher.h>
namespace pubsub {
namespace nanomsg {
+ class pubsub_nanomsg_topic_sender;
+
+ class bounded_service_entry {
+ public:
+ bounded_service_entry(
+ pubsub::nanomsg::pubsub_nanomsg_topic_sender *_parent,
+ long _bndId,
+ log_helper_t* _logHelper) : parent{_parent}, bndId{_bndId}, logHelper{_logHelper} {
+
+ }
+
+ int topicPublicationSend(unsigned int msgTypeId, const void *inMsg);
+
+ pubsub::nanomsg::pubsub_nanomsg_topic_sender *parent{};
+ pubsub_publisher_t service{};
+ long bndId{};
+ hash_map_t *msgTypes{};
+ int getCount{1};
+ log_helper_t *logHelper{};
+ } ;
+
+
class pubsub_nanomsg_topic_sender {
public:
pubsub_nanomsg_topic_sender(celix_bundle_context_t *_ctx, log_helper_t *_logHelper, const char *_scope,
@@ -38,31 +62,30 @@ namespace pubsub {
const pubsub_nanomsg_topic_sender &operator=(const pubsub_nanomsg_topic_sender &) = delete;
long getSerializerSvcId() const ;
- const char *getScope() const ;
- const char *getTopic() const ;
- const char *getUrl() const;
+ const std::string &getScope() const ;
+ const std::string &getTopic() const ;
+ const std::string &getUrl() const;
void* getPublisherService(const celix_bundle_t *requestingBundle,
- const celix_properties_t *svcProperties __attribute__((unused)));
+ const celix_properties_t *svcProperties __attribute__((unused)));
void ungetPublisherService(const celix_bundle_t *requestingBundle,
- const celix_properties_t *svcProperties __attribute__((unused)));
+ const celix_properties_t *svcProperties __attribute__((unused)));
int topicPublicationSend(unsigned int msgTypeId, const void *inMsg);
void delay_first_send_for_late_joiners() ;
-
- //private:
+ //private:
celix_bundle_context_t *ctx;
log_helper_t *logHelper;
long serializerSvcId;
pubsub_serializer_service_t *serializer;
- char *scope{};
- char *topic{};
- char scopeAndTopicFilter[5];
- char *url{};
+ std::string scope{};
+ std::string topic{};
+ std::string scopeAndTopicFilter{};
+ std::string url{};
struct {
- celix_thread_mutex_t mutex;
+ std::mutex mutex;
int socket;
} nanomsg{};
@@ -72,8 +95,9 @@ namespace pubsub {
} publisher{};
struct {
- celix_thread_mutex_t mutex{};
- hash_map_t *map{}; //key = bndId, value = psa_nanomsg_bounded_service_entry_t
+ std::mutex mutex{};
+ std::map<long, bounded_service_entry> map{};
+ //hash_map_t *map{}; //key = bndId, value = psa_nanomsg_bounded_service_entry_t
} boundedServices{};
};
}
[2/3] celix git commit: NanoMsg
Posted by er...@apache.org.
NanoMsg
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/b6a03372
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/b6a03372
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/b6a03372
Branch: refs/heads/nanomsg
Commit: b6a0337233c48c48bf48a86274682ba8329f91f3
Parents: 15f268d
Author: Erjan Altena <er...@gmail.com>
Authored: Thu Nov 29 20:39:40 2018 +0100
Committer: Erjan Altena <er...@gmail.com>
Committed: Thu Nov 29 20:39:40 2018 +0100
----------------------------------------------------------------------
.../src/pubsub_nanomsg_admin.cc | 7 +-
.../src/pubsub_nanomsg_admin.h | 2 +-
.../src/pubsub_nanomsg_topic_receiver.cc | 34 +++---
.../src/pubsub_nanomsg_topic_receiver.h | 4 +-
.../src/pubsub_nanomsg_topic_sender.cc | 119 ++++++++++++-------
.../src/pubsub_nanomsg_topic_sender.h | 14 ++-
6 files changed, 106 insertions(+), 74 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
index 9fdf2a3..cc17ebb 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
@@ -250,13 +250,11 @@ void pubsub_nanomsg_admin::removeSerializerSvc(void */*svc*/, const celix_proper
auto &entry = kvsm->second;
{
std::lock_guard<std::mutex> senderLock(topicSenders.mutex);
- for (auto kv: topicSenders.map) {
+ for (auto &kv: topicSenders.map) {
auto *sender = kv.second;
if (sender != nullptr && entry.svcId == sender->getSerializerSvcId()) {
- char *key = kv.first;
topicSenders.map.erase(kv.first);
delete (sender);
- free(key);
}
}
}
@@ -377,9 +375,7 @@ celix_status_t pubsub_nanomsg_admin::teardownTopicSender(const char *scope, cons
std::lock_guard<std::mutex> topicSenderLock(topicSenders.mutex);
auto kv = topicSenders.map.find(key);
if (kv != topicSenders.map.end()) {
- char *mapKey = kv->first;
pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = kv->second;
- free(mapKey);
//TODO disconnect endpoints to sender. note is this needed for a nanomsg topic sender?
delete (sender);
} else {
@@ -557,7 +553,6 @@ celix_status_t pubsub_nanomsg_admin::removeEndpoint(const celix_properties_t *en
celix_status_t pubsub_nanomsg_admin::executeCommand(char *commandLine __attribute__((unused)), FILE *out,
FILE *errStream __attribute__((unused))) {
celix_status_t status = CELIX_SUCCESS;
-
fprintf(out, "\n");
fprintf(out, "Topic Senders:\n");
{
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
index 7c2e9a0..8ec35e5 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
@@ -119,7 +119,7 @@ private:
pubsub_serializer_service_t *svc;
} psa_nanomsg_serializer_entry_t;
ProtectedMap<long, psa_nanomsg_serializer_entry_t> serializers{};
- ProtectedMap<char*, pubsub::nanomsg::pubsub_nanomsg_topic_sender*> topicSenders{};
+ ProtectedMap<std::string, pubsub::nanomsg::pubsub_nanomsg_topic_sender*> topicSenders{};
ProtectedMap<std::string, pubsub::nanomsg::topic_receiver*> topicReceivers{};
ProtectedMap<const char*, celix_properties_t *> discoveredEndpoints{};
};
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
index 30c2af7..83d1caf 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.cc
@@ -72,7 +72,6 @@ pubsub::nanomsg::topic_receiver::topic_receiver(celix_bundle_context_t *_ctx,
ctx = _ctx;
logHelper = _logHelper;
serializer = _serializer;
- m_scopeAndTopicFilter = psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic);
m_nanoMsgSocket = nn_socket(AF_SP, NN_BUS);
if (m_nanoMsgSocket < 0) {
@@ -88,20 +87,7 @@ pubsub::nanomsg::topic_receiver::topic_receiver(celix_bundle_context_t *_ctx,
auto subscriberFilter = psa_nanomsg_setScopeAndTopicFilter(m_scope, m_topic);
- int size = snprintf(NULL, 0, "(%s=%s)", PUBSUB_SUBSCRIBER_TOPIC, m_topic.c_str());
- char buf[size + 1];
- snprintf(buf, (size_t) size + 1, "(%s=%s)", PUBSUB_SUBSCRIBER_TOPIC, m_topic.c_str());
- celix_service_tracking_options_t opts{};
- opts.filter.ignoreServiceLanguage = true;
- opts.filter.serviceName = PUBSUB_SUBSCRIBER_SERVICE_NAME;
- opts.filter.filter = buf;
- opts.callbackHandle = this;
- opts.addWithOwner = [](void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *svcOwner) {
- static_cast<topic_receiver*>(handle)->addSubscriber(svc, props, svcOwner);
- };
- opts.removeWithOwner = [](void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *svcOwner) {
- static_cast<topic_receiver*>(handle)->removeSubscriber(svc, props, svcOwner);
- };
+ auto opts = createOptions();
subscriberTrackerId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
recvThread.running = true;
@@ -110,6 +96,24 @@ pubsub::nanomsg::topic_receiver::topic_receiver(celix_bundle_context_t *_ctx,
}
}
+celix_service_tracking_options_t pubsub::nanomsg::topic_receiver::createOptions() {
+ std::stringstream filter_str;
+
+ filter_str << "(" << PUBSUB_SUBSCRIBER_TOPIC << "=" << m_topic << ")";
+ celix_service_tracking_options_t opts{};
+ opts.filter.ignoreServiceLanguage = true;
+ opts.filter.serviceName = PUBSUB_SUBSCRIBER_SERVICE_NAME;
+ opts.filter.filter = strdup(filter_str.str().c_str()); // TODO : memory leak ??
+ opts.callbackHandle = this;
+ opts.addWithOwner = [](void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *svcOwner) {
+ static_cast<pubsub::nanomsg::topic_receiver*>(handle)->addSubscriber(svc, props, svcOwner);
+ };
+ opts.removeWithOwner = [](void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *svcOwner) {
+ static_cast<pubsub::nanomsg::topic_receiver*>(handle)->removeSubscriber(svc, props, svcOwner);
+ };
+ return opts;
+}
+
pubsub::nanomsg::topic_receiver::~topic_receiver() {
{
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
index 02c462e..6b0950a 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_receiver.h
@@ -30,7 +30,7 @@
struct psa_nanomsg_subscriber_entry {
psa_nanomsg_subscriber_entry(pubsub_subscriber_t *_svc, int _usageCount) :
- svc{_svc}, usageCount{_usageCount} {
+ svc{_svc}, usageCount{_usageCount} {
}
pubsub_subscriber_t *svc{};
int usageCount;
@@ -92,6 +92,7 @@ namespace pubsub {
void processMsgForSubscriberEntry(psa_nanomsg_subscriber_entry* entry, const pubsub_nanmosg_msg_header_t *hdr, const char* payload, size_t payloadSize);
void addSubscriber(void *svc, const celix_properties_t *props, const celix_bundle_t *bnd);
void removeSubscriber(void */*svc*/, const celix_properties_t */*props*/, const celix_bundle_t *bnd);
+ celix_service_tracking_options_t createOptions();
private:
celix_bundle_context_t *ctx{nullptr};
@@ -100,7 +101,6 @@ namespace pubsub {
pubsub_serializer_service_t *serializer{nullptr};
const std::string m_scope{};
const std::string m_topic{};
- std::string m_scopeAndTopicFilter{};
int m_nanoMsgSocket{0};
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
index 9a253b2..7521d78 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.cc
@@ -19,7 +19,7 @@
#include <memory.h>
#include <iostream>
-
+#include <sstream>
#include <stdlib.h>
#include <utils.h>
#include <arpa/inet.h>
@@ -38,18 +38,46 @@
#define FIRST_SEND_DELAY_IN_SECONDS 2
#define NANOMSG_BIND_MAX_RETRY 10
-#define L_DEBUG(...) \
- logHelper_log(logHelper, OSGI_LOGSERVICE_DEBUG, __VA_ARGS__)
-#define L_INFO(...) \
- logHelper_log(logHelper, OSGI_LOGSERVICE_INFO, __VA_ARGS__)
-#define L_WARN(...) \
- logHelper_log(logHelper, OSGI_LOGSERVICE_WARNING, __VA_ARGS__)
-#define L_ERROR(...) \
- logHelper_log(logHelper, OSGI_LOGSERVICE_ERROR, __VA_ARGS__)
+template <typename T>
+std::stringstream LOG_STREAM(T first) {
+ std::stringstream ss;
+ ss << first;
+ return ss;
+}
+
+template <typename T, typename... Args>
+std::stringstream LOG_STREAM(T first, Args... args) {
+ std::stringstream ss;
+ ss << first << LOG_STREAM(args...).str();
+ return ss;
+}
+
+template <typename... Args>
+void L_DEBUG(log_helper_t *logHelper, Args... args) {
+ std::stringstream ss = LOG_STREAM(args...);
+ logHelper_log(logHelper, OSGI_LOGSERVICE_DEBUG, ss.str().c_str());
+}
+
+template <typename... Args>
+void L_INFO(log_helper_t *logHelper, Args... args) {
+ auto ss = LOG_STREAM(args...);
+ logHelper_log(logHelper, OSGI_LOGSERVICE_INFO, ss.str().c_str());
+}
+
+template <typename... Args>
+void L_WARN(log_helper_t *logHelper, Args... args) {
+ auto ss = LOG_STREAM(args...);
+ logHelper_log(logHelper, OSGI_LOGSERVICE_WARNING, ss.str().c_str());
+}
+template <typename... Args>
+void L_ERROR(log_helper_t *logHelper, Args... args) {
+ auto ss = LOG_STREAM(args...);
+ logHelper_log((log_helper_pt)logHelper, OSGI_LOGSERVICE_ERROR, ss.str().c_str());
+}
static unsigned int rand_range(unsigned int min, unsigned int max);
-static void delay_first_send_for_late_joiners(pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender);
+static void delay_first_send_for_late_joiners(log_helper_t* logHelper);
pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_bundle_context_t *_ctx,
log_helper_t *_logHelper,
@@ -70,8 +98,8 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_
scopeAndTopicFilter = psa_nanomsg_setScopeAndTopicFilter(_scope, _topic);
//setting up nanomsg socket for nanomsg TopicSender
- int socket = nn_socket(AF_SP, NN_BUS);
- if (socket == -1) {
+ int nnSock = nn_socket(AF_SP, NN_BUS);
+ if (nnSock == -1) {
perror("Error for nanomsg_socket");
}
@@ -79,23 +107,20 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::pubsub_nanomsg_topic_sender(celix_
while(rv == -1 && retry < NANOMSG_BIND_MAX_RETRY ) {
/* Randomized part due to same bundle publishing on different topics */
unsigned int port = rand_range(_basePort,_maxPort);
- size_t len = (size_t)snprintf(nullptr, 0, "tcp://%s:%u", _bindIp, port) + 1;
- char *_url = static_cast<char*>(calloc(len, sizeof(char*)));
- snprintf(_url, len, "tcp://%s:%u", _bindIp, port);
-
- len = (size_t)snprintf(nullptr, 0, "tcp://0.0.0.0:%u", port) + 1;
- char *bindUrl = static_cast<char*>(calloc(len, sizeof(char)));
- snprintf(bindUrl, len, "tcp://0.0.0.0:%u", port);
- rv = nn_bind (socket, bindUrl);
+ std::stringstream _url;
+ _url << "tcp://" << _bindIp << ":" << port;
+
+ std::stringstream bindUrl;
+ bindUrl << "tcp://0.0.0.0:" << port;
+
+ rv = nn_bind (nnSock, bindUrl.str().c_str());
if (rv == -1) {
perror("Error for nn_bind");
- free(_url);
} else {
- this->url = _url;
- nanomsg.socket = socket;
+ this->url = _url.str();
+ nanomsg.socket = nnSock;
}
retry++;
- free(bindUrl);
}
if (!url.empty()) {
@@ -132,7 +157,6 @@ pubsub::nanomsg::pubsub_nanomsg_topic_sender::~pubsub_nanomsg_topic_sender() {
celix_bundleContext_unregisterService(ctx, publisher.svcId);
nn_close(nanomsg.socket);
-
std::lock_guard<std::mutex> lock(boundedServices.mutex);
for (auto &it: boundedServices.map) {
serializer->destroySerializerMap(serializer->handle, it.second.msgTypes);
@@ -170,7 +194,7 @@ void* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getPublisherService(const ce
} else {
auto entry = boundedServices.map.emplace(std::piecewise_construct,
std::forward_as_tuple(bndId),
- std::forward_as_tuple(this, bndId, logHelper));
+ std::forward_as_tuple(scope, topic, bndId, nanomsg.socket, logHelper));
int rc = serializer->createSerializerMap(serializer->handle, (celix_bundle_t*)requestingBundle, &entry.first->second.msgTypes);
if (rc == 0) {
@@ -179,11 +203,14 @@ void* pubsub::nanomsg::pubsub_nanomsg_topic_sender::getPublisherService(const ce
entry.first->second.service.send = [](void *handle, unsigned int msgTypeId, const void *msg) {
return static_cast<pubsub::nanomsg::bounded_service_entry*>(handle)->topicPublicationSend(msgTypeId, msg);
};
- entry.first->second.service.sendMultipart = NULL; //not supported TODO remove
+ entry.first->second.service.sendMultipart = nullptr; //not supported TODO remove
service = &entry.first->second.service;
} else {
boundedServices.map.erase(bndId);
- L_ERROR("Error creating serializer map for NanoMsg TopicSender %s/%s", scope, topic);
+ auto x = LOG_STREAM(12, "hallo");
+ logHelper_log(logHelper, OSGI_LOGSERVICE_DEBUG, x.str().c_str());
+ log_helper_pt lh = logHelper;
+ L_ERROR(lh, "Error creating serializer map for NanoMsg TopicSender ", scope, topic);
}
}
@@ -201,7 +228,7 @@ void pubsub::nanomsg::pubsub_nanomsg_topic_sender::ungetPublisherService(const c
if (entry->second.getCount == 0) {
int rc = serializer->destroySerializerMap(serializer->handle, entry->second.msgTypes);
if (rc != 0) {
- L_ERROR("Error destroying publisher service, serializer not available / cannot get msg serializer map\n");
+ L_ERROR(logHelper, "Error destroying publisher service, serializer not available / cannot get msg serializer map\n");
}
boundedServices.map.erase(bndId);
}
@@ -210,31 +237,30 @@ void pubsub::nanomsg::pubsub_nanomsg_topic_sender::ungetPublisherService(const c
int pubsub::nanomsg::bounded_service_entry::topicPublicationSend(unsigned int msgTypeId, const void *inMsg) {
int status;
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = parent;
- pubsub_msg_serializer_t* msgSer = static_cast<pubsub_msg_serializer_t*>(hashMap_get(msgTypes, (void*)(uintptr_t)msgTypeId));
+ auto msgSer = static_cast<pubsub_msg_serializer_t*>(hashMap_get(msgTypes, (void*)(uintptr_t)msgTypeId));
- if (msgSer != NULL) {
- delay_first_send_for_late_joiners(sender);
+ if (msgSer != nullptr) {
+ delay_first_send_for_late_joiners(logHelper);
int major = 0, minor = 0;
pubsub_nanmosg_msg_header_t msg_hdr;// = calloc(1, sizeof(*msg_hdr));
msg_hdr.type = msgTypeId;
- if (msgSer->msgVersion != NULL) {
+ if (msgSer->msgVersion != nullptr) {
version_getMajor(msgSer->msgVersion, &major);
version_getMinor(msgSer->msgVersion, &minor);
msg_hdr.major = (unsigned char) major;
msg_hdr.minor = (unsigned char) minor;
}
- void *serializedOutput = NULL;
+ void *serializedOutput = nullptr;
size_t serializedOutputLen = 0;
status = msgSer->serialize(msgSer, inMsg, &serializedOutput, &serializedOutputLen);
if (status == CELIX_SUCCESS) {
nn_iovec data[2];
- nn_msghdr msg;
+ nn_msghdr msg{};
msg.msg_iov = data;
msg.msg_iovlen = 2;
msg.msg_iov[0].iov_base = static_cast<void*>(&msg_hdr);
@@ -244,31 +270,34 @@ int pubsub::nanomsg::bounded_service_entry::topicPublicationSend(unsigned int ms
msg.msg_control = nullptr;
msg.msg_controllen = 0;
errno = 0;
- int rc = nn_sendmsg(sender->nanomsg.socket, &msg, 0 );
+ int rc = nn_sendmsg(nanoMsgSocket, &msg, 0 );
free(serializedOutput);
if (rc < 0) {
- L_WARN("[PSA_ZMQ_TS] Error sending zmsg, rc is %i. %s", rc, strerror(errno));
+ L_WARN(logHelper, "[PSA_ZMQ_TS] Error sending zmsg, rc: ", rc, ", error: ", strerror(errno));
} else {
- L_INFO("[PSA_ZMQ_TS] Send message with size %d\n", rc);
- L_INFO("[PSA_ZMQ_TS] Send message ID %d, major %d, minor %d\n", msg_hdr.type, (int)msg_hdr.major, (int)msg_hdr.minor);
+ L_INFO(logHelper, "[PSA_ZMQ_TS] Send message with size ", rc, "\n");
+ L_INFO(logHelper, "[PSA_ZMQ_TS] Send message ID ", msg_hdr.type,
+ " major: ", (int)msg_hdr.major,
+ " minor: ", (int)msg_hdr.minor,"\n");
}
} else {
- L_WARN("[PSA_ZMQ_TS] Error serialize message of type %s for scope/topic %s/%s", msgSer->msgName, sender->scope, sender->topic);
+ L_WARN(logHelper, "[PSA_ZMQ_TS] Error serialize message of type ", msgSer->msgName,
+ " for scope/topic ", scope.c_str(), "/", topic.c_str(),"\n");
}
} else {
status = CELIX_SERVICE_EXCEPTION;
- L_WARN("[PSA_ZMQ_TS] Error cannot serialize message with msg type id %i for scope/topic %s/%s", msgTypeId, sender->scope, sender->topic);
+ L_WARN(logHelper, "[PSA_ZMQ_TS] Error cannot serialize message with msg type id ", msgTypeId,
+ " for scope/topic ", scope.c_str(), "/", topic.c_str(),"\n");
}
return status;
}
-static void delay_first_send_for_late_joiners(pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender) {
+static void delay_first_send_for_late_joiners(log_helper_t* logHelper) {
static bool firstSend = true;
if(firstSend){
- auto logHelper = sender->logHelper;
- L_INFO("PSA_UDP_MC_TP: Delaying first send for late joiners...\n");
+ L_INFO(logHelper, "PSA_UDP_MC_TP: Delaying first send for late joiners...\n");
sleep(FIRST_SEND_DELAY_IN_SECONDS);
firstSend = false;
}
http://git-wip-us.apache.org/repos/asf/celix/blob/b6a03372/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
index 1883a41..6a3254e 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_topic_sender.h
@@ -27,24 +27,28 @@
namespace pubsub {
namespace nanomsg {
- class pubsub_nanomsg_topic_sender;
class bounded_service_entry {
public:
bounded_service_entry(
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *_parent,
+ std::string &_scope,
+ std::string &_topic,
long _bndId,
- log_helper_t* _logHelper) : parent{_parent}, bndId{_bndId}, logHelper{_logHelper} {
+ int _nanoMsgSocket,
+ log_helper_t* _logHelper) : scope{_scope}, topic{_topic}, bndId{_bndId}, nanoMsgSocket{_nanoMsgSocket}, logHelper{_logHelper} {
}
-
+ bounded_service_entry(const bounded_service_entry&) = delete;
+ bounded_service_entry &operator=(const bounded_service_entry&) = delete;
int topicPublicationSend(unsigned int msgTypeId, const void *inMsg);
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *parent{};
pubsub_publisher_t service{};
+ std::string scope;
+ std::string topic;
long bndId{};
hash_map_t *msgTypes{};
int getCount{1};
+ int nanoMsgSocket{};
log_helper_t *logHelper{};
} ;
[3/3] celix git commit: Nanomsg
Posted by er...@apache.org.
Nanomsg
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/707b8e54
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/707b8e54
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/707b8e54
Branch: refs/heads/nanomsg
Commit: 707b8e54aeda61c5eba4b8a3fa65059ec0f5821a
Parents: b6a0337
Author: Erjan Altena <er...@gmail.com>
Authored: Sat Dec 1 09:26:44 2018 +0100
Committer: Erjan Altena <er...@gmail.com>
Committed: Sat Dec 1 09:26:44 2018 +0100
----------------------------------------------------------------------
.../src/pubsub_nanomsg_admin.cc | 66 ++++++++------------
.../src/pubsub_nanomsg_admin.h | 4 +-
2 files changed, 29 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/707b8e54/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
index cc17ebb..ba47723 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.cc
@@ -28,6 +28,7 @@
#include <netdb.h>
#include <ifaddrs.h>
#include <pubsub_endpoint.h>
+#include <algorithm>
#include "pubsub_utils.h"
#include "pubsub_nanomsg_admin.h"
@@ -96,11 +97,11 @@ pubsub_nanomsg_admin::pubsub_nanomsg_admin(celix_bundle_context_t *_ctx, log_hel
pubsub_nanomsg_admin::~pubsub_nanomsg_admin() {
//note assuming al psa register services and service tracker are removed.
{
- std::lock_guard<std::mutex> lock(topicSenders.mutex);
- for (auto kv : topicSenders.map) {
- auto *sender = kv.second;
- delete (sender);
- }
+// std::lock_guard<std::mutex> lock(topicSenders.mutex);
+// for (auto &kv : topicSenders.map) {
+// auto &sender = kv.second;
+// delete (sender);
+// }
}
{
@@ -224,11 +225,6 @@ void pubsub_nanomsg_admin::addSerializerSvc(void *svc, const celix_properties_t
serializers.map.emplace(std::piecewise_construct,
std::forward_as_tuple(svcId),
std::forward_as_tuple(serType, svcId, static_cast<pubsub_serializer_service_t*>(svc)));
-// auto entry = static_cast<psa_nanomsg_serializer_entry_t*>(calloc(1, sizeof(psa_nanomsg_serializer_entry_t)));
-// entry->serType = serType;
-// entry->svcId = svcId;
-// entry->svc = static_cast<pubsub_serializer_service_t*>(svc);
-// serializers.map[svcId] = entry;
}
}
}
@@ -250,11 +246,12 @@ void pubsub_nanomsg_admin::removeSerializerSvc(void */*svc*/, const celix_proper
auto &entry = kvsm->second;
{
std::lock_guard<std::mutex> senderLock(topicSenders.mutex);
- for (auto &kv: topicSenders.map) {
- auto *sender = kv.second;
- if (sender != nullptr && entry.svcId == sender->getSerializerSvcId()) {
- topicSenders.map.erase(kv.first);
- delete (sender);
+ for(auto it = topicSenders.map.begin(); it != topicSenders.map.end(); /*nothing*/) {
+ auto &sender = it->second;
+ if (entry.svcId == sender.getSerializerSvcId()) {
+ it = topicSenders.map.erase(it);
+ } else {
+ ++it;
}
}
}
@@ -320,32 +317,30 @@ celix_status_t pubsub_nanomsg_admin::setupTopicSender(const char *scope, const c
celix_properties_t *newEndpoint = nullptr;
char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = nullptr;
std::lock_guard<std::mutex> serializerLock(serializers.mutex);
std::lock_guard<std::mutex> topicSenderLock(topicSenders.mutex);
- sender = topicSenders.map.find(key)->second;
- if (sender == nullptr) {
+ auto sender = topicSenders.map.find(key);
+ if (sender == topicSenders.map.end()) {
psa_nanomsg_serializer_entry_t *serEntry = nullptr;
auto kv = serializers.map.find(serializerSvcId);
if (kv != serializers.map.end()) {
serEntry = &kv->second;
}
if (serEntry != nullptr) {
- sender = new pubsub::nanomsg::pubsub_nanomsg_topic_sender(ctx, log, scope, topic, serializerSvcId, serEntry->svc, ipAddress,
- basePort, maxPort);
- }
- if (sender != nullptr) {
+ auto e = topicSenders.map.emplace(std::piecewise_construct,
+ std::forward_as_tuple(key),
+ std::forward_as_tuple(ctx, log, scope, topic, serializerSvcId, serEntry->svc, ipAddress,
+ basePort, maxPort));
const char *psaType = PUBSUB_NANOMSG_ADMIN_TYPE;
const char *serType = serEntry->serType;
newEndpoint = pubsubEndpoint_create(fwUUID, scope, topic, PUBSUB_PUBLISHER_ENDPOINT_TYPE, psaType, serType,
nullptr);
- celix_properties_set(newEndpoint, PUBSUB_NANOMSG_URL_KEY, sender->getUrl().c_str());
+ celix_properties_set(newEndpoint, PUBSUB_NANOMSG_URL_KEY, e.first->second.getUrl().c_str());
//if available also set container name
const char *cn = celix_bundleContext_getProperty(ctx, "CELIX_CONTAINER_NAME", nullptr);
if (cn != nullptr) {
celix_properties_set(newEndpoint, "container_name", cn);
}
- topicSenders.map[key] = sender;
} else {
L_ERROR("[PSA NANOMSG] Error creating a TopicSender");
free(key);
@@ -354,9 +349,6 @@ celix_status_t pubsub_nanomsg_admin::setupTopicSender(const char *scope, const c
free(key);
L_ERROR("[PSA_NANOMSG] Cannot setup already existing TopicSender for scope/topic %s/%s!", scope, topic);
}
- if (sender != nullptr && newEndpoint != nullptr) {
- //TODO connect endpoints to sender, NOTE is this needed for a nanomsg topic sender?
- }
if (newEndpoint != nullptr && outPublisherEndpoint != nullptr) {
*outPublisherEndpoint = newEndpoint;
@@ -373,12 +365,8 @@ celix_status_t pubsub_nanomsg_admin::teardownTopicSender(const char *scope, cons
char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
std::lock_guard<std::mutex> topicSenderLock(topicSenders.mutex);
- auto kv = topicSenders.map.find(key);
- if (kv != topicSenders.map.end()) {
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = kv->second;
- //TODO disconnect endpoints to sender. note is this needed for a nanomsg topic sender?
- delete (sender);
- } else {
+ ;
+ if (topicSenders.map.erase(key) == 0) {
L_ERROR("[PSA NANOMSG] Cannot teardown TopicSender with scope/topic %s/%s. Does not exists", scope, topic);
}
free(key);
@@ -558,14 +546,14 @@ celix_status_t pubsub_nanomsg_admin::executeCommand(char *commandLine __attribut
{
std::lock_guard<std::mutex> serializerLock(serializers.mutex);
std::lock_guard<std::mutex> topicSenderLock(topicSenders.mutex);
- for (auto kvts: topicSenders.map) {
- pubsub::nanomsg::pubsub_nanomsg_topic_sender *sender = kvts.second;
- long serSvcId = sender->getSerializerSvcId();
+ for (auto &senderEntry: topicSenders.map) {
+ auto &sender = senderEntry.second;
+ long serSvcId = sender.getSerializerSvcId();
auto kvs = serializers.map.find(serSvcId);
const char* serType = ( kvs == serializers.map.end() ? "!Error" : kvs->second.serType);
- const auto scope = sender->getScope();
- const auto topic = sender->getTopic();
- const auto url = sender->getUrl();
+ const auto scope = sender.getScope();
+ const auto topic = sender.getTopic();
+ const auto url = sender.getUrl();
fprintf(out, "|- Topic Sender %s/%s\n", scope.c_str(), topic.c_str());
fprintf(out, " |- serializer type = %s\n", serType);
fprintf(out, " |- url = %s\n", url.c_str());
http://git-wip-us.apache.org/repos/asf/celix/blob/707b8e54/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
index 8ec35e5..8ed5ddd 100644
--- a/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
+++ b/bundles/pubsub/pubsub_admin_nanomsg/src/pubsub_nanomsg_admin.h
@@ -119,9 +119,9 @@ private:
pubsub_serializer_service_t *svc;
} psa_nanomsg_serializer_entry_t;
ProtectedMap<long, psa_nanomsg_serializer_entry_t> serializers{};
- ProtectedMap<std::string, pubsub::nanomsg::pubsub_nanomsg_topic_sender*> topicSenders{};
+ ProtectedMap<std::string, pubsub::nanomsg::pubsub_nanomsg_topic_sender> topicSenders{};
ProtectedMap<std::string, pubsub::nanomsg::topic_receiver*> topicReceivers{};
- ProtectedMap<const char*, celix_properties_t *> discoveredEndpoints{};
+ ProtectedMap<const std::string, celix_properties_t *> discoveredEndpoints{};
};
#ifdef __cplusplus