You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celix.apache.org by pn...@apache.org on 2018/10/12 09:03:31 UTC
[14/34] celix git commit: CELIX-454: Refactors the pubsub zmq admin.
CELIX-454: Refactors the pubsub zmq admin.
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/00454efd
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/00454efd
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/00454efd
Branch: refs/heads/develop
Commit: 00454efd94af591bc9ca36c80b4ba42b651cf5d7
Parents: ffb97ff
Author: Pepijn Noltes <pe...@gmail.com>
Authored: Thu Sep 27 16:57:48 2018 +0200
Committer: Pepijn Noltes <pe...@gmail.com>
Committed: Thu Sep 27 16:57:48 2018 +0200
----------------------------------------------------------------------
bundles/pubsub/CMakeLists.txt | 2 -
bundles/pubsub/examples/CMakeLists.txt | 37 +-
.../src/pubsub_udpmc_admin.c | 9 +-
.../src/pubsub_udpmc_admin.h | 2 +-
bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt | 15 +-
.../pubsub/pubsub_admin_zmq/src/psa_activator.c | 133 +---
.../pubsub_admin_zmq/src/pubsub_admin_impl.h | 3 +
.../pubsub_admin_zmq/src/pubsub_zmq_admin.c | 563 +++++++++++++++
.../pubsub_admin_zmq/src/pubsub_zmq_admin.h | 59 ++
.../pubsub_admin_zmq/src/pubsub_zmq_common.c | 40 ++
.../pubsub_admin_zmq/src/pubsub_zmq_common.h | 34 +
.../src/pubsub_zmq_topic_receiver.c | 684 +++++++++++++++++++
.../src/pubsub_zmq_topic_receiver.h | 41 ++
.../src/pubsub_zmq_topic_sender.c | 556 +++++++++++++++
.../src/pubsub_zmq_topic_sender.h | 46 ++
bundles/pubsub/pubsub_discovery/CMakeLists.txt | 5 +-
.../pubsub/pubsub_discovery/src/psd_activator.c | 14 +-
.../src/pubsub_discovery_impl.c | 137 +++-
.../src/pubsub_discovery_impl.h | 9 +-
.../pubsub/pubsub_spi/src/pubsub_utils_match.c | 6 +-
.../src/pubsub_topology_manager.c | 59 +-
libs/etcdlib/src/etcd.c | 20 +-
libs/utils/include/celix_array_list.h | 2 +
libs/utils/src/array_list.c | 20 +-
24 files changed, 2310 insertions(+), 186 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/bundles/pubsub/CMakeLists.txt b/bundles/pubsub/CMakeLists.txt
index d7ae3f8..e3db995 100644
--- a/bundles/pubsub/CMakeLists.txt
+++ b/bundles/pubsub/CMakeLists.txt
@@ -19,8 +19,6 @@ celix_subproject(PUBSUB "Option to build the pubsub bundles" OFF DEPS UTILS)
if (PUBSUB)
option(BUILD_PUBSUB_PSA_ZMQ "Build ZeroMQ PubSub Admin (LGPL License)" OFF)
- set(BUILD_PUBSUB_PSA_ZMQ OFF)
- message(WARNING "TODO enable PSA_ZMQ again, for now disabled because refactoring is needed")
if (BUILD_PUBSUB_PSA_ZMQ)
message(WARNING "Celix will now contain a dependency with a LGPL License (ZeroMQ). For more information about this, consult the pubsub/README.md file.")
option(BUILD_ZMQ_SECURITY "Build with security for ZeroMQ." OFF)
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/examples/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/bundles/pubsub/examples/CMakeLists.txt b/bundles/pubsub/examples/CMakeLists.txt
index 41a3406..9df2314 100644
--- a/bundles/pubsub/examples/CMakeLists.txt
+++ b/bundles/pubsub/examples/CMakeLists.txt
@@ -153,8 +153,8 @@ if (BUILD_PUBSUB_PSA_ZMQ)
target_link_libraries(pubsub_zmq PRIVATE ${PUBSUB_CONTAINER_LIBS})
add_celix_container("pubsub_publisher_zmq"
- GROUP "pubsub"
- BUNDLES
+ GROUP "pubsub"
+ BUNDLES
Celix::shell
Celix::shell_tui
Celix::pubsub_serializer_json
@@ -163,14 +163,16 @@ if (BUILD_PUBSUB_PSA_ZMQ)
Celix::pubsub_admin_zmq
celix_pubsub_poi_publisher
celix_pubsub_poi_publisher2
- PROPERTIES
- pubsub.scope=my_small_scope
- )
+ PROPERTIES
+ PSA_ZMQ_VERBOSE=true
+ PUBSUB_ETCD_DISCOVERY_VERBOSE=true
+ PUBSUB_TOPOLOGY_MANAGER_VERBOSE=true
+ )
target_link_libraries(pubsub_publisher_zmq PRIVATE ${PUBSUB_CONTAINER_LIBS})
- add_celix_container("pubsub_subscriber_zmq"
- GROUP "pubsub"
- BUNDLES
+ add_celix_container(pubsub_subscriber_zmq
+ GROUP "pubsub"
+ BUNDLES
Celix::shell
Celix::shell_tui
Celix::pubsub_serializer_json
@@ -178,12 +180,16 @@ if (BUILD_PUBSUB_PSA_ZMQ)
Celix::pubsub_topology_manager
Celix::pubsub_admin_zmq
celix_pubsub_poi_subscriber
- )
+ PROPERTIES
+ PSA_ZMQ_VERBOSE=true
+ PUBSUB_ETCD_DISCOVERY_VERBOSE=true
+ PUBSUB_TOPOLOGY_MANAGER_VERBOSE=true
+ )
target_link_libraries(pubsub_subscriber_zmq PRIVATE ${PUBSUB_CONTAINER_LIBS})
- add_celix_container("pubsub_subscriber2_zmq"
- GROUP "pubsub"
- BUNDLES
+ add_celix_container(pubsub_subscriber2_zmq
+ GROUP "pubsub"
+ BUNDLES
Celix::shell
Celix::shell_tui
Celix::pubsub_serializer_json
@@ -191,8 +197,11 @@ if (BUILD_PUBSUB_PSA_ZMQ)
Celix::pubsub_topology_manager
Celix::pubsub_admin_zmq
celix_pubsub_poi_subscriber
-
- )
+ PROPERTIES
+ PSA_ZMQ_VERBOSE=true
+ PUBSUB_ETCD_DISCOVERY_VERBOSE=true
+ PUBSUB_TOPOLOGY_MANAGER_VERBOSE=true
+ )
target_link_libraries(pubsub_subscriber2_zmq PRIVATE ${PUBSUB_CONTAINER_LIBS})
# ZMQ Multipart
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.c b/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.c
index 3e06334..1935c15 100644
--- a/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.c
+++ b/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.c
@@ -73,13 +73,6 @@ struct pubsub_udpmc_admin {
};
-typedef struct psa_udpmc_connected_endpoint {
- void *sender; //if connected endpoint is subscriber. todo type
- void *receiver; //if connected endpoint is publisher. TODO type
- char *endpointUUID;
-} psa_udpmc_connected_endpoint_t;
-
-
static celix_status_t udpmc_getIpAddress(const char* interface, char** ip);
static celix_status_t pubsub_udpmcAdmin_connectEndpointToReceiver(pubsub_udpmc_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint);
static celix_status_t pubsub_udpmcAdmin_disconnectEndpointFromReceiver(pubsub_udpmc_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint);
@@ -540,7 +533,7 @@ celix_status_t pubsub_udpmcAdmin_executeCommand(void *handle, char *commandLine
const char *serType = pubsub_udpmcTopicReceiver_serializerType(receiver);
const char *scope = pubsub_udpmcTopicReceiver_scope(receiver);
const char *topic = pubsub_udpmcTopicReceiver_topic(receiver);
- fprintf(out, "|- Topic Sender %s/%s\n", scope, topic);
+ fprintf(out, "|- Topic Receiver %s/%s\n", scope, topic);
fprintf(out, " |- psa type = %s\n", psaType);
fprintf(out, " |- serializer type = %s\n", serType);
}
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.h b/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.h
index 11c7e13..e050384 100644
--- a/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.h
+++ b/bundles/pubsub/pubsub_admin_udp_mc/src/pubsub_udpmc_admin.h
@@ -33,7 +33,7 @@
#define PUBSUB_UDPMC_VERBOSE_KEY "PSA_UDPMC_VERBOSE"
#define PUBSUB_UDPMC_MULTICAST_IP_PREFIX_DEFAULT "224.100"
-#define PUBSUB_UDPMC_VERBOSE_DEFAULT "true"
+#define PUBSUB_UDPMC_VERBOSE_DEFAULT true
typedef struct pubsub_udpmc_admin pubsub_udpmc_admin_t;
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt b/bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt
index 19ab58d..b960615 100644
--- a/bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt
+++ b/bundles/pubsub/pubsub_admin_zmq/CMakeLists.txt
@@ -31,14 +31,15 @@ if (BUILD_PUBSUB_PSA_ZMQ)
endif()
add_celix_bundle(celix_pubsub_admin_zmq
- BUNDLE_SYMBOLICNAME "apache_celix_pubsub_admin_zmq"
- VERSION "1.0.0"
- GROUP "Celix/PubSub"
- SOURCES
+ BUNDLE_SYMBOLICNAME "apache_celix_pubsub_admin_zmq"
+ VERSION "1.0.0"
+ GROUP "Celix/PubSub"
+ SOURCES
src/psa_activator.c
- src/pubsub_admin_impl.c
- src/topic_subscription.c
- src/topic_publication.c
+ src/pubsub_zmq_admin.c
+ src/pubsub_zmq_topic_sender.c
+ src/pubsub_zmq_topic_receiver.c
+ src/pubsub_zmq_common.c
${ZMQ_CRYPTO_C}
)
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/psa_activator.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/psa_activator.c b/bundles/pubsub/pubsub_admin_zmq/src/psa_activator.c
index 3eb887b..b7cc138 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/psa_activator.c
+++ b/bundles/pubsub/pubsub_admin_zmq/src/psa_activator.c
@@ -17,130 +17,68 @@
*under the License.
*/
-/**
- *Licensed to the Apache Software Foundation (ASF) under one
- *or more contributor license agreements. See the NOTICE file
- *distributed with this work for additional information
- *regarding copyright ownership. The ASF licenses this file
- *to you under the Apache License, Version 2.0 (the
- *"License"); you may not use this file except in compliance
- *with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- *Unless required by applicable law or agreed to in writing,
- *software distributed under the License is distributed on an
- *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- *specific language governing permissions and limitations
- *under the License.
- */
-
#include <stdlib.h>
#include "celix_api.h"
#include "pubsub_serializer.h"
+#include "log_helper.h"
-#include "pubsub_admin_impl.h"
+#include "pubsub_admin.h"
+#include "pubsub_zmq_admin.h"
+#include "../../../shell/shell/include/command.h"
typedef struct psa_zmq_activator {
- pubsub_admin_t *admin;
+ log_helper_t *logHelper;
+
+ pubsub_zmq_admin_t *admin;
pubsub_admin_service_t adminService;
long adminSvcId;
command_service_t cmdSvc;
long cmdSvcId;
-
- long serializerTrackerId;
} psa_zmq_activator_t;
-static celix_status_t shellCommand(void *handle, char * commandLine, FILE *outStream, FILE *errorStream) {
- psa_zmq_activator_t *act= handle;
- if (act->admin->externalPublications && !hashMap_isEmpty(act->admin->externalPublications)) {
- fprintf(outStream, "External Publications:\n");
- for(hash_map_iterator_t iter = hashMapIterator_construct(act->admin->externalPublications); hashMapIterator_hasNext(&iter);) {
- const char* key = (const char*)hashMapIterator_nextKey(&iter);
- fprintf(outStream, " %s\n", key);
- }
- }
- if (act->admin->localPublications && !hashMap_isEmpty(act->admin->localPublications)) {
- fprintf(outStream, "Local Publications:\n");
- for (hash_map_iterator_t iter = hashMapIterator_construct(
- act->admin->localPublications); hashMapIterator_hasNext(&iter);) {
- const char *key = (const char *) hashMapIterator_nextKey(&iter);
- fprintf(outStream, " %s\n", key);
- }
- }
- if (act->admin->subscriptions && !hashMap_isEmpty(act->admin->subscriptions)) {
- fprintf(outStream, "Active Subscriptions:\n");
- for (hash_map_iterator_t iter = hashMapIterator_construct(
- act->admin->subscriptions); hashMapIterator_hasNext(&iter);) {
- const char *key = (const char *) hashMapIterator_nextKey(&iter);
- fprintf(outStream, " %s\n", key);
- }
- }
- if (act->admin->pendingSubscriptions && !hashMap_isEmpty(act->admin->pendingSubscriptions)) {
- fprintf(outStream, "Pending Subscriptions:\n");
- for (hash_map_iterator_t iter = hashMapIterator_construct(
- act->admin->pendingSubscriptions); hashMapIterator_hasNext(&iter);) {
- const char *key = (const char *) hashMapIterator_nextKey(&iter);
- fprintf(outStream, " %s\n", key);
- }
- }
- return CELIX_SUCCESS;
-}
-
int psa_zmq_start(psa_zmq_activator_t *act, celix_bundle_context_t *ctx) {
act->adminSvcId = -1L;
- act->serializerTrackerId = -1l;
+ act->cmdSvcId = -1L;
- celix_status_t status = pubsubAdmin_create(ctx, &(act->admin));
+ logHelper_create(ctx, &act->logHelper);
+ logHelper_start(act->logHelper);
- //track pubsub serializers
- if (status == CELIX_SUCCESS) {
- celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
- opts.filter.serviceName = PUBSUB_SERIALIZER_SERVICE_NAME;
- opts.filter.ignoreServiceLanguage = true;
-
- opts.callbackHandle = act->admin;
- opts.addWithProperties = pubsubAdmin_addSerializer;
- opts.removeWithProperties = pubsubAdmin_removeSerializer;
- act->serializerTrackerId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
- }
+ act->admin = pubsub_zmqAdmin_create(ctx, act->logHelper);
+ celix_status_t status = act->admin != NULL ? CELIX_SUCCESS : CELIX_BUNDLE_EXCEPTION;
- //register psa service
+ //register pubsub admin service
if (status == CELIX_SUCCESS) {
pubsub_admin_service_t *psaSvc = &act->adminService;
psaSvc->handle = act->admin;
- psaSvc->matchPublisher = pubsubAdmin_matchPublisher;
- psaSvc->matchSubscriber = pubsubAdmin_matchSubscriber;
- psaSvc->matchEndpoint = pubsubAdmin_matchEndpoint;
- psaSvc->setupTopicSender = pubsubAdmin_setupTopicSender;
- psaSvc->teardownTopicSender = pubsubAdmin_teardownTopicSender;
- psaSvc->setupTopicReciever = pubsubAdmin_setupTopicReciever;
- psaSvc->teardownTopicReciever = pubsubAdmin_teardownTopicReciever;
- psaSvc->addEndpoint = pubsubAdmin_addEndpoint;
- psaSvc->removeEndpoint = pubsubAdmin_removeEndpoint;
+ psaSvc->matchPublisher = pubsub_zmqAdmin_matchPublisher;
+ psaSvc->matchSubscriber = pubsub_zmqAdmin_matchSubscriber;
+ psaSvc->matchEndpoint = pubsub_zmqAdmin_matchEndpoint;
+ psaSvc->setupTopicSender = pubsub_zmqAdmin_setupTopicSender;
+ psaSvc->teardownTopicSender = pubsub_zmqAdmin_teardownTopicSender;
+ psaSvc->setupTopicReciever = pubsub_zmqAdmin_setupTopicReceiver;
+ psaSvc->teardownTopicReciever = pubsub_zmqAdmin_teardownTopicReceiver;
+ psaSvc->addEndpoint = pubsub_zmqAdmin_addEndpoint;
+ psaSvc->removeEndpoint = pubsub_zmqAdmin_removeEndpoint;
celix_properties_t *props = celix_properties_create();
- celix_properties_set(props, PUBSUB_ADMIN_SERVICE_TYPE, PUBSUB_PSA_ZMQ_PSA_TYPE);
+ celix_properties_set(props, PUBSUB_ADMIN_SERVICE_TYPE, PUBSUB_ZMQ_ADMIN_TYPE);
act->adminSvcId = celix_bundleContext_registerService(ctx, psaSvc, PUBSUB_ADMIN_SERVICE_NAME, props);
}
- //register psa shell command
- if (status == CELIX_SUCCESS) {
- act->cmdSvc.handle = act;
- act->cmdSvc.executeCommand = shellCommand;
-
- celix_properties_t *shellProps = celix_properties_create();
- celix_properties_set(shellProps, OSGI_SHELL_COMMAND_NAME, "psa_zmq_info");
- celix_properties_set(shellProps, OSGI_SHELL_COMMAND_USAGE, "psa_zmq_info");
- celix_properties_set(shellProps, OSGI_SHELL_COMMAND_DESCRIPTION, "psa_zmq_info: Overview of PubSub ZMQ Admin");
-
- act->cmdSvcId = celix_bundleContext_registerService(ctx, &act->cmdSvc, OSGI_SHELL_COMMAND_SERVICE_NAME, shellProps);
+ //register shell command service
+ {
+ act->cmdSvc.handle = act->admin;
+ act->cmdSvc.executeCommand = pubsub_zmqAdmin_executeCommand;
+ celix_properties_t *props = celix_properties_create();
+ celix_properties_set(props, OSGI_SHELL_COMMAND_NAME, "psa_zmq");
+ celix_properties_set(props, OSGI_SHELL_COMMAND_USAGE, "psa_zmq");
+ celix_properties_set(props, OSGI_SHELL_COMMAND_DESCRIPTION, "Print the information about the TopicSender and TopicReceivers for the ZMQ PSA");
+ act->cmdSvcId = celix_bundleContext_registerService(ctx, &act->cmdSvc, OSGI_SHELL_COMMAND_SERVICE_NAME, props);
}
return status;
@@ -149,8 +87,11 @@ int psa_zmq_start(psa_zmq_activator_t *act, celix_bundle_context_t *ctx) {
int psa_zmq_stop(psa_zmq_activator_t *act, celix_bundle_context_t *ctx) {
celix_bundleContext_unregisterService(ctx, act->adminSvcId);
celix_bundleContext_unregisterService(ctx, act->cmdSvcId);
- celix_bundleContext_stopTracker(ctx, act->serializerTrackerId);
- pubsubAdmin_destroy(act->admin);
+ pubsub_zmqAdmin_destroy(act->admin);
+
+ logHelper_stop(act->logHelper);
+ logHelper_destroy(&act->logHelper);
+
return CELIX_SUCCESS;
}
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_admin_impl.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_admin_impl.h b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_admin_impl.h
index e8f80b1..47a946c 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_admin_impl.h
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_admin_impl.h
@@ -44,6 +44,9 @@
#include "command.h"
#define PUBSUB_PSA_ZMQ_PSA_TYPE "zmq"
+#define PUBSUB_PSA_ZMQ_IP "PSA_IP"
+#define PUBSUB_PSA_ZMQ_ITF "PSA_INTERFACE"
+
typedef struct pubsub_admin {
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c
new file mode 100644
index 0000000..d3ba3b5
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c
@@ -0,0 +1,563 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+
+#include <memory.h>
+#include <sys/socket.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <ifaddrs.h>
+#include <pubsub_endpoint.h>
+#include <czmq.h>
+
+#include "pubsub_utils.h"
+#include "pubsub_zmq_admin.h"
+#include "pubsub_psa_zmq_constants.h"
+#include "pubsub_zmq_topic_sender.h"
+#include "pubsub_zmq_topic_receiver.h"
+
+#define L_DEBUG(...) \
+ logHelper_log(psa->log, OSGI_LOGSERVICE_DEBUG, __VA_ARGS__)
+#define L_INFO(...) \
+ logHelper_log(psa->log, OSGI_LOGSERVICE_INFO, __VA_ARGS__)
+#define L_WARN(...) \
+ logHelper_log(psa->log, OSGI_LOGSERVICE_WARNING, __VA_ARGS__)
+#define L_ERROR(...) \
+ logHelper_log(psa->log, OSGI_LOGSERVICE_ERROR, __VA_ARGS__)
+
+struct pubsub_zmq_admin {
+ celix_bundle_context_t *ctx;
+ log_helper_t *log;
+ const char *fwUUID;
+
+ char* ipAddress;
+ zactor_t* zmq_auth;
+
+ unsigned int basePort;
+ unsigned int maxPort;
+
+ double qosSampleScore;
+ double qosControlScore;
+ double defaultScore;
+
+ bool verbose;
+
+ struct {
+ celix_thread_mutex_t mutex;
+ hash_map_t *map; //key = scope:topic key, value = pubsub_zmq_topic_sender_t
+ } topicSenders;
+
+ struct {
+ celix_thread_mutex_t mutex;
+ hash_map_t *map; //key = scope:topic key, value = pubsub_zmq_topic_sender_t
+ } topicReceivers;
+
+ struct {
+ celix_thread_mutex_t mutex;
+ hash_map_t *map; //key = endpoint uuid, value = psa_zmq_connected_endpoint_entry_t
+ } discoveredEndpoints;
+
+};
+
+static celix_status_t zmq_getIpAddress(const char* interface, const char** ip);
+static celix_status_t pubsub_zmqAdmin_connectEndpointToReceiver(pubsub_zmq_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint);
+static celix_status_t pubsub_zmqAdmin_disconnectEndpointFromReceiver(pubsub_zmq_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint);
+
+
+pubsub_zmq_admin_t* pubsub_zmqAdmin_create(celix_bundle_context_t *ctx, log_helper_t *logHelper) {
+ pubsub_zmq_admin_t *psa = calloc(1, sizeof(*psa));
+ psa->ctx = ctx;
+ psa->log = logHelper;
+ psa->verbose = celix_bundleContext_getPropertyAsBool(ctx, PUBSUB_ZMQ_VERBOSE_KEY, PUBSUB_ZMQ_VERBOSE_DEFAULT);
+ psa->fwUUID = celix_bundleContext_getProperty(ctx, OSGI_FRAMEWORK_FRAMEWORK_UUID, NULL);
+
+ const char *ip = celix_bundleContext_getProperty(ctx, PUBSUB_ZMQ_PSA_IP_KEY , NULL);
+
+ if (ip == NULL) {
+ //TODO try to get ip from subnet (CIDR)
+ }
+
+ if (ip == NULL) {
+ //try to get ip from itf
+ const char *interface = celix_bundleContext_getProperty(ctx, PUBSUB_ZMQ_PSA_ITF_KEY, NULL);
+ zmq_getIpAddress(interface, &ip);
+ }
+
+ if (ip == NULL) {
+ L_WARN("[PSA_ZMQ] Could not determine IP address for PSA, using default ip (%s)", PUBSUB_ZMQ_DEFAULT_IP);
+ ip = PUBSUB_ZMQ_DEFAULT_IP;
+ }
+
+ psa->ipAddress = strndup(ip, 1024*1024);
+ if (psa->verbose) {
+ L_INFO("[PSA_ZMQ] Using %s for service annunciation", ip);
+ }
+
+
+ long basePort = celix_bundleContext_getPropertyAsLong(ctx, PSA_ZMQ_BASE_PORT, PSA_ZMQ_DEFAULT_BASE_PORT);
+ long maxPort = celix_bundleContext_getPropertyAsLong(ctx, PSA_ZMQ_MAX_PORT, PSA_ZMQ_DEFAULT_MAX_PORT);
+ psa->basePort = (unsigned int)basePort;
+ psa->maxPort = (unsigned int)maxPort;
+ if (psa->verbose) {
+ L_INFO("[PSA_ZMQ] Using base till max port: %i till %i", psa->basePort, psa->maxPort);
+ }
+
+ // Disable Signal Handling by CZMQ
+ setenv("ZSYS_SIGHANDLER", "false", true);
+
+ long nrThreads = celix_bundleContext_getPropertyAsLong(ctx, PUBSUB_ZMQ_NR_THREADS_KEY, 0);
+ if (nrThreads > 0) {
+ zsys_set_io_threads((size_t)nrThreads);
+ L_INFO("[PSA_ZMQ] Using %d threads for ZMQ", (size_t)nrThreads);
+ }
+
+
+#ifdef BUILD_WITH_ZMQ_SECURITY
+ // Setup authenticator
+ zactor_t* auth = zactor_new (zauth, NULL);
+ zstr_sendx(auth, "VERBOSE", NULL);
+
+ // Load all public keys of subscribers into the application
+ // This step is done for authenticating subscribers
+ char curve_folder_path[MAX_KEY_FOLDER_PATH_LENGTH];
+ char* keys_bundle_dir = pubsub_getKeysBundleDir(context);
+ snprintf(curve_folder_path, MAX_KEY_FOLDER_PATH_LENGTH, "%s/META-INF/keys/subscriber/public", keys_bundle_dir);
+ zstr_sendx (auth, "CURVE", curve_folder_path, NULL);
+ free(keys_bundle_dir);
+
+ (*admin)->zmq_auth = auth;
+#endif
+
+ psa->defaultScore = celix_bundleContext_getPropertyAsDouble(ctx, PSA_ZMQ_DEFAULT_SCORE_KEY, PSA_ZMQ_DEFAULT_SCORE);
+ psa->qosSampleScore = celix_bundleContext_getPropertyAsDouble(ctx, PSA_ZMQ_QOS_SAMPLE_SCORE_KEY, PSA_ZMQ_DEFAULT_QOS_SAMPLE_SCORE);
+ psa->qosControlScore = celix_bundleContext_getPropertyAsDouble(ctx, PSA_ZMQ_QOS_CONTROL_SCORE_KEY, PSA_ZMQ_DEFAULT_QOS_CONTROL_SCORE);
+
+ celixThreadMutex_create(&psa->topicSenders.mutex, NULL);
+ psa->topicSenders.map = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
+
+ celixThreadMutex_create(&psa->topicReceivers.mutex, NULL);
+ psa->topicReceivers.map = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
+
+ celixThreadMutex_create(&psa->discoveredEndpoints.mutex, NULL);
+ psa->discoveredEndpoints.map = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
+
+ return psa;
+}
+
+void pubsub_zmqAdmin_destroy(pubsub_zmq_admin_t *psa) {
+ if (psa == NULL) {
+ return;
+ }
+
+ //note assuming al psa register services and service tracker are removed.
+
+ celixThreadMutex_lock(&psa->topicSenders.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicSenders.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_zmq_topic_sender_t *sender = hashMapIterator_nextValue(&iter);
+ pubsub_zmqTopicSender_destroy(sender);
+ }
+ celixThreadMutex_unlock(&psa->topicSenders.mutex);
+
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_updmc_topic_receiver_t *recv = hashMapIterator_nextValue(&iter);
+ pubsub_zmqTopicReceiver_destroy(recv);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
+
+ celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
+ iter = hashMapIterator_construct(psa->discoveredEndpoints.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ celix_properties_t *ep = hashMapIterator_nextValue(&iter);
+ celix_properties_destroy(ep);
+ }
+ celixThreadMutex_unlock(&psa->discoveredEndpoints.mutex);
+
+ celixThreadMutex_destroy(&psa->topicSenders.mutex);
+ hashMap_destroy(psa->topicSenders.map, true, false);
+
+ celixThreadMutex_destroy(&psa->topicReceivers.mutex);
+ hashMap_destroy(psa->topicReceivers.map, true, false);
+
+ celixThreadMutex_destroy(&psa->discoveredEndpoints.mutex);
+ hashMap_destroy(psa->discoveredEndpoints.map, true, false);
+
+ if (psa->zmq_auth != NULL){
+ zactor_destroy(&psa->zmq_auth);
+ }
+
+ free(psa->ipAddress);
+
+ free(psa);
+}
+
+celix_status_t pubsub_zmqAdmin_matchPublisher(void *handle, long svcRequesterBndId, const celix_filter_t *svcFilter, double *outScore, long *outSerializerSvcId) {
+ pubsub_zmq_admin_t *psa = handle;
+ L_DEBUG("[PSA_ZMQ] pubsub_zmqAdmin_matchPublisher");
+ celix_status_t status = CELIX_SUCCESS;
+ double score = pubsub_utils_matchPublisher(psa->ctx, svcRequesterBndId, svcFilter->filterStr, PUBSUB_ZMQ_ADMIN_TYPE,
+ psa->qosSampleScore, psa->qosControlScore, psa->defaultScore, outSerializerSvcId);
+ *outScore = score;
+
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_matchSubscriber(void *handle, long svcProviderBndId, const celix_properties_t *svcProperties, double *outScore, long *outSerializerSvcId) {
+ pubsub_zmq_admin_t *psa = handle;
+ L_DEBUG("[PSA_ZMQ] pubsub_zmqAdmin_matchSubscriber");
+ celix_status_t status = CELIX_SUCCESS;
+ double score = pubsub_utils_matchSubscriber(psa->ctx, svcProviderBndId, svcProperties, PUBSUB_ZMQ_ADMIN_TYPE,
+ psa->qosSampleScore, psa->qosControlScore, psa->defaultScore, outSerializerSvcId);
+ if (outScore != NULL) {
+ *outScore = score;
+ }
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_matchEndpoint(void *handle, const celix_properties_t *endpoint, bool *outMatch) {
+ pubsub_zmq_admin_t *psa = handle;
+ L_DEBUG("[PSA_ZMQ] pubsub_zmqAdmin_matchEndpoint");
+ celix_status_t status = CELIX_SUCCESS;
+ bool match = pubsub_utils_matchEndpoint(psa->ctx, endpoint, PUBSUB_ZMQ_ADMIN_TYPE, NULL);
+ if (outMatch != NULL) {
+ *outMatch = match;
+ }
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_setupTopicSender(void *handle, const char *scope, const char *topic, long serializerSvcId, celix_properties_t **outPublisherEndpoint) {
+ pubsub_zmq_admin_t *psa = handle;
+ celix_status_t status = CELIX_SUCCESS;
+
+ //1) Create TopicSender
+ //2) Store TopicSender
+ //3) Connect existing endpoints
+ //4) set outPublisherEndpoint
+
+ celix_properties_t *newEndpoint = NULL;
+
+ char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
+ celixThreadMutex_lock(&psa->topicSenders.mutex);
+ pubsub_zmq_topic_sender_t *sender = hashMap_get(psa->topicSenders.map, key);
+ if (sender == NULL) {
+ sender = pubsub_zmqTopicSender_create(psa->ctx, psa->log, scope, topic, serializerSvcId, psa->ipAddress, psa->basePort, psa->maxPort);
+ if (sender != NULL) {
+ const char *psaType = pubsub_zmqTopicSender_psaType(sender);
+ const char *serType = pubsub_zmqTopicSender_serializerType(sender);
+ newEndpoint = pubsubEndpoint_create(psa->fwUUID, scope, topic, PUBSUB_PUBLISHER_ENDPOINT_TYPE, psaType,
+ serType, NULL);
+ celix_properties_set(newEndpoint, PUBSUB_ZMQ_URL_KEY, pubsub_zmqTopicSender_url(sender));
+ //if available also set container name
+ const char *cn = celix_bundleContext_getProperty(psa->ctx, "CELIX_CONTAINER_NAME", NULL);
+ if (cn != NULL) {
+ celix_properties_set(newEndpoint, "container_name", cn);
+ }
+
+ //TODO connect endpoints to sender, NOTE is this needed for a zmq topic sender?
+ hashMap_put(psa->topicSenders.map, key, sender);
+ } else {
+ L_ERROR("[PSA ZMQ] Error creating a TopicSender");
+ free(key);
+ }
+ } else {
+ free(key);
+ L_ERROR("[PSA_ZMQ] Cannot setup already existing TopicSender for scope/topic %s/%s!", scope, topic);
+ }
+ celixThreadMutex_unlock(&psa->topicSenders.mutex);
+
+
+ if (newEndpoint != NULL && outPublisherEndpoint != NULL) {
+ *outPublisherEndpoint = newEndpoint;
+ }
+
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_teardownTopicSender(void *handle, const char *scope, const char *topic) {
+ pubsub_zmq_admin_t *psa = handle;
+ celix_status_t status = CELIX_SUCCESS;
+
+ //1) Find and remove TopicSender from map
+ //2) destroy topic sender
+
+ char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
+ celixThreadMutex_lock(&psa->topicSenders.mutex);
+ hash_map_entry_t *entry = hashMap_getEntry(psa->topicSenders.map, key);
+ if (entry != NULL) {
+ char *mapKey = hashMapEntry_getKey(entry);
+ pubsub_zmq_topic_sender_t *sender = hashMap_remove(psa->topicSenders.map, key);
+ free(mapKey);
+ //TODO disconnect endpoints to sender. note is this needed for a zmq topic sender?
+ pubsub_zmqTopicSender_destroy(sender);
+ } else {
+ L_ERROR("[PSA ZMQ] Cannot teardown TopicSender with scope/topic %s/%s. Does not exists", scope, topic);
+ }
+ celixThreadMutex_unlock(&psa->topicSenders.mutex);
+ free(key);
+
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_setupTopicReceiver(void *handle, const char *scope, const char *topic, long serializerSvcId, celix_properties_t **outSubscriberEndpoint) {
+ pubsub_zmq_admin_t *psa = handle;
+
+ celix_properties_t *newEndpoint = NULL;
+
+ char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ pubsub_updmc_topic_receiver_t *receiver = hashMap_get(psa->topicReceivers.map, key);
+ if (receiver == NULL) {
+ receiver = pubsub_zmqTopicReceiver_create(psa->ctx, psa->log, scope, topic, serializerSvcId);
+ if (receiver != NULL) {
+ const char *psaType = pubsub_zmqTopicReceiver_psaType(receiver);
+ const char *serType = pubsub_zmqTopicReceiver_serializerType(receiver);
+ newEndpoint = pubsubEndpoint_create(psa->fwUUID, scope, topic,
+ PUBSUB_SUBSCRIBER_ENDPOINT_TYPE, psaType, serType, NULL);
+
+ //if available also set container name
+ const char *cn = celix_bundleContext_getProperty(psa->ctx, "CELIX_CONTAINER_NAME", NULL);
+ if (cn != NULL) {
+ celix_properties_set(newEndpoint, "container_name", cn);
+ }
+
+ celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->discoveredEndpoints.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ celix_properties_t *endpoint = hashMapIterator_nextValue(&iter);
+ pubsub_zmqAdmin_connectEndpointToReceiver(psa, receiver, endpoint);
+ }
+ celixThreadMutex_unlock(&psa->discoveredEndpoints.mutex);
+
+ hashMap_put(psa->topicReceivers.map, key, receiver);
+ } else {
+ L_ERROR("[PSA ZMQ] Error creating a TopicReceiver.");
+ free(key);
+ }
+ } else {
+ free(key);
+ L_ERROR("[PSA_ZMQ] Cannot setup already existing TopicReceiver for scope/topic %s/%s!", scope, topic);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
+
+
+ if (newEndpoint != NULL && outSubscriberEndpoint != NULL) {
+ *outSubscriberEndpoint = newEndpoint;
+ }
+
+ celix_status_t status = CELIX_SUCCESS;
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_teardownTopicReceiver(void *handle, const char *scope, const char *topic) {
+ pubsub_zmq_admin_t *psa = handle;
+
+ char *key = pubsubEndpoint_createScopeTopicKey(scope, topic);
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ hash_map_entry_t *entry = hashMap_getEntry(psa->topicReceivers.map, key);
+ free(key);
+ if (entry != NULL) {
+ char *receiverKey = hashMapEntry_getKey(entry);
+ pubsub_updmc_topic_receiver_t *receiver = hashMapEntry_getValue(entry);
+ hashMap_remove(psa->topicReceivers.map, receiverKey);
+
+ free(receiverKey);
+ pubsub_zmqTopicReceiver_destroy(receiver);
+ }
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+
+ celix_status_t status = CELIX_SUCCESS;
+ return status;
+}
+
+static celix_status_t pubsub_zmqAdmin_connectEndpointToReceiver(pubsub_zmq_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint) {
+ //note can be called with discoveredEndpoint.mutex lock
+ celix_status_t status = CELIX_SUCCESS;
+
+ const char *scope = pubsub_zmqTopicReceiver_scope(receiver);
+ const char *topic = pubsub_zmqTopicReceiver_topic(receiver);
+
+ const char *type = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TYPE, NULL);
+ const char *eScope = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_SCOPE, NULL);
+ const char *eTopic = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_NAME, NULL);
+ const char *url = celix_properties_get(endpoint, PUBSUB_ZMQ_URL_KEY, NULL);
+
+ if (type == NULL || url == NULL) {
+ fprintf(stderr, "[PSA UPDMC] Error got endpoint without zmq url or endpoint type");
+ status = CELIX_BUNDLE_EXCEPTION;
+ } else {
+ if (eScope != NULL && eTopic != NULL && type != NULL &&
+ strncmp(eScope, scope, 1024 * 1024) == 0 &&
+ strncmp(eTopic, topic, 1024 * 1024) == 0 &&
+ strncmp(type, PUBSUB_PUBLISHER_ENDPOINT_TYPE, strlen(PUBSUB_PUBLISHER_ENDPOINT_TYPE)) == 0) {
+ pubsub_zmqTopicReceiver_connectTo(receiver, url);
+ }
+ }
+
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_addEndpoint(void *handle, const celix_properties_t *endpoint) {
+ pubsub_zmq_admin_t *psa = handle;
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_updmc_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
+ pubsub_zmqAdmin_connectEndpointToReceiver(psa, receiver, endpoint);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
+
+ celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
+ celix_properties_t *cpy = celix_properties_copy(endpoint);
+ const char *uuid = celix_properties_get(cpy, PUBSUB_ENDPOINT_UUID, NULL);
+ hashMap_put(psa->discoveredEndpoints.map, (void*)uuid, cpy);
+ celixThreadMutex_unlock(&psa->discoveredEndpoints.mutex);
+
+ celix_status_t status = CELIX_SUCCESS;
+ return status;
+}
+
+
+static celix_status_t pubsub_zmqAdmin_disconnectEndpointFromReceiver(pubsub_zmq_admin_t* psa, pubsub_updmc_topic_receiver_t *receiver, const celix_properties_t *endpoint) {
+ //note can be called with discoveredEndpoint.mutex lock
+ celix_status_t status = CELIX_SUCCESS;
+
+ const char *scope = pubsub_zmqTopicReceiver_scope(receiver);
+ const char *topic = pubsub_zmqTopicReceiver_topic(receiver);
+
+ const char *type = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TYPE, NULL);
+ const char *eScope = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_SCOPE, NULL);
+ const char *eTopic = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TOPIC_NAME, NULL);
+ const char *url = celix_properties_get(endpoint, PUBSUB_ZMQ_URL_KEY, NULL);
+
+ if (type == NULL || url == NULL) {
+ fprintf(stderr, "[PSA UPDMC] Error got endpoint without zmq url or endpoint type");
+ status = CELIX_BUNDLE_EXCEPTION;
+ } else {
+ if (eScope != NULL && eTopic != NULL && type != NULL &&
+ strncmp(eScope, scope, 1024 * 1024) == 0 &&
+ strncmp(eTopic, topic, 1024 * 1024) == 0 &&
+ strncmp(type, PUBSUB_PUBLISHER_ENDPOINT_TYPE, strlen(PUBSUB_PUBLISHER_ENDPOINT_TYPE)) == 0) {
+ pubsub_zmqTopicReceiver_disconnectFrom(receiver, url);
+ }
+ }
+
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_removeEndpoint(void *handle, const celix_properties_t *endpoint) {
+ pubsub_zmq_admin_t *psa = handle;
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_updmc_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
+ pubsub_zmqAdmin_disconnectEndpointFromReceiver(psa, receiver, endpoint);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
+
+ celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
+ const char *uuid = celix_properties_get(endpoint, PUBSUB_ENDPOINT_UUID, NULL);
+ celix_properties_t *found = hashMap_remove(psa->discoveredEndpoints.map, (void*)uuid);
+ celixThreadMutex_unlock(&psa->discoveredEndpoints.mutex);
+
+ if (found != NULL) {
+ celix_properties_destroy(found);
+ }
+
+ celix_status_t status = CELIX_SUCCESS;
+ return status;
+}
+
+celix_status_t pubsub_zmqAdmin_executeCommand(void *handle, char *commandLine __attribute__((unused)), FILE *out, FILE *errStream __attribute__((unused))) {
+ pubsub_zmq_admin_t *psa = handle;
+ celix_status_t status = CELIX_SUCCESS;
+
+ fprintf(out, "\n");
+ fprintf(out, "Topic Senders:\n");
+ celixThreadMutex_lock(&psa->topicSenders.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicSenders.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_zmq_topic_sender_t *sender = hashMapIterator_nextValue(&iter);
+ const char *psaType = pubsub_zmqTopicSender_psaType(sender);
+ const char *serType = pubsub_zmqTopicSender_serializerType(sender);
+ const char *scope = pubsub_zmqTopicSender_scope(sender);
+ const char *topic = pubsub_zmqTopicSender_topic(sender);
+ const char *url = pubsub_zmqTopicSender_url(sender);
+ fprintf(out, "|- Topic Sender %s/%s\n", scope, topic);
+ fprintf(out, " |- psa type = %s\n", psaType);
+ fprintf(out, " |- serializer type = %s\n", serType);
+ fprintf(out, " |- url = %s\n", url);
+ }
+ celixThreadMutex_unlock(&psa->topicSenders.mutex);
+
+ fprintf(out, "\n");
+ fprintf(out, "\nTopic Receivers:\n");
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_updmc_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
+ const char *psaType = pubsub_zmqTopicReceiver_psaType(receiver);
+ const char *serType = pubsub_zmqTopicReceiver_serializerType(receiver);
+ const char *scope = pubsub_zmqTopicReceiver_scope(receiver);
+ const char *topic = pubsub_zmqTopicReceiver_topic(receiver);
+ fprintf(out, "|- Topic Receiver %s/%s\n", scope, topic);
+ fprintf(out, " |- psa type = %s\n", psaType);
+ fprintf(out, " |- serializer type = %s\n", serType);
+ }
+ celixThreadMutex_unlock(&psa->topicSenders.mutex);
+ fprintf(out, "\n");
+
+ //TODO topic receivers/senders connection count
+
+ return status;
+}
+
+#ifndef ANDROID
+static celix_status_t zmq_getIpAddress(const char* interface, const char** ip) {
+ celix_status_t status = CELIX_BUNDLE_EXCEPTION;
+
+ struct ifaddrs *ifaddr, *ifa;
+ char host[NI_MAXHOST];
+
+ if (getifaddrs(&ifaddr) != -1)
+ {
+ for (ifa = ifaddr; ifa != NULL && status != CELIX_SUCCESS; ifa = ifa->ifa_next)
+ {
+ if (ifa->ifa_addr == NULL)
+ continue;
+
+ if ((getnameinfo(ifa->ifa_addr,sizeof(struct sockaddr_in), host, NI_MAXHOST, NULL, 0, NI_NUMERICHOST) == 0) && (ifa->ifa_addr->sa_family == AF_INET)) {
+ if (interface == NULL) {
+ *ip = strdup(host);
+ status = CELIX_SUCCESS;
+ }
+ else if (strcmp(ifa->ifa_name, interface) == 0) {
+ *ip = strdup(host);
+ status = CELIX_SUCCESS;
+ }
+ }
+ }
+
+ freeifaddrs(ifaddr);
+ }
+
+ return status;
+}
+#endif
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.h b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.h
new file mode 100644
index 0000000..a1c5a75
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.h
@@ -0,0 +1,59 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+
+#ifndef CELIX_PUBSUB_ZMQ_ADMIN_H
+#define CELIX_PUBSUB_ZMQ_ADMIN_H
+
+#include "celix_api.h"
+#include "log_helper.h"
+
+#define PUBSUB_ZMQ_ADMIN_TYPE "zmq"
+#define PUBSUB_ZMQ_URL_KEY "zmq.url"
+
+#define PUBSUB_ZMQ_VERBOSE_KEY "PSA_ZMQ_VERBOSE"
+#define PUBSUB_ZMQ_VERBOSE_DEFAULT true
+
+#define PUBSUB_ZMQ_PSA_IP_KEY "PSA_IP"
+#define PUBSUB_ZMQ_PSA_ITF_KEY "PSA_INTERFACE"
+#define PUBSUB_ZMQ_NR_THREADS_KEY "PSA_ZMQ_NR_THREADS"
+
+#define PUBSUB_ZMQ_DEFAULT_IP "127.0.0.1"
+
+typedef struct pubsub_zmq_admin pubsub_zmq_admin_t;
+
+pubsub_zmq_admin_t* pubsub_zmqAdmin_create(celix_bundle_context_t *ctx, log_helper_t *logHelper);
+void pubsub_zmqAdmin_destroy(pubsub_zmq_admin_t *psa);
+
+celix_status_t pubsub_zmqAdmin_matchPublisher(void *handle, long svcRequesterBndId, const celix_filter_t *svcFilter, double *score, long *serializerSvcId);
+celix_status_t pubsub_zmqAdmin_matchSubscriber(void *handle, long svcProviderBndId, const celix_properties_t *svcProperties, double *score, long *serializerSvcId);
+celix_status_t pubsub_zmqAdmin_matchEndpoint(void *handle, const celix_properties_t *endpoint, bool *match);
+
+celix_status_t pubsub_zmqAdmin_setupTopicSender(void *handle, const char *scope, const char *topic, long serializerSvcId, celix_properties_t **publisherEndpoint);
+celix_status_t pubsub_zmqAdmin_teardownTopicSender(void *handle, const char *scope, const char *topic);
+
+celix_status_t pubsub_zmqAdmin_setupTopicReceiver(void *handle, const char *scope, const char *topic, long serializerSvcId, celix_properties_t **subscriberEndpoint);
+celix_status_t pubsub_zmqAdmin_teardownTopicReceiver(void *handle, const char *scope, const char *topic);
+
+celix_status_t pubsub_zmqAdmin_addEndpoint(void *handle, const celix_properties_t *endpoint);
+celix_status_t pubsub_zmqAdmin_removeEndpoint(void *handle, const celix_properties_t *endpoint);
+
+celix_status_t pubsub_zmqAdmin_executeCommand(void *handle, char *commandLine, FILE *outStream, FILE *errStream);
+
+#endif //CELIX_PUBSUB_ZMQ_ADMIN_H
+
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.c b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.c
new file mode 100644
index 0000000..8726faa
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.c
@@ -0,0 +1,40 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+
+#include "pubsub_zmq_common.h"
+
+int psa_zmq_localMsgTypeIdForMsgType(void* handle __attribute__((unused)), const char* msgType, unsigned int* msgTypeId) {
+ *msgTypeId = utils_stringHash(msgType);
+ return 0;
+}
+
+bool psa_zmq_checkVersion(version_pt msgVersion, pubsub_msg_header_t *hdr) {
+ bool check=false;
+ int major=0,minor=0;
+
+ if(msgVersion!=NULL){
+ version_getMajor(msgVersion,&major);
+ version_getMinor(msgVersion,&minor);
+ if(hdr->major==((unsigned char)major)){ /* Different major means incompatible */
+ check = (hdr->minor>=((unsigned char)minor)); /* Compatible only if the provider has a minor equals or greater (means compatible update) */
+ }
+ }
+
+ return check;
+}
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.h b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.h
new file mode 100644
index 0000000..d36dc18
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_common.h
@@ -0,0 +1,34 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+
+#ifndef CELIX_PUBSUB_ZMQ_COMMON_H
+#define CELIX_PUBSUB_ZMQ_COMMON_H
+
+#include <utils.h>
+
+#include "version.h"
+#include "pubsub_common.h"
+
+
+int psa_zmq_localMsgTypeIdForMsgType(void* handle __attribute__((unused)), const char* msgType, unsigned int* msgTypeId);
+
+bool psa_zmq_checkVersion(version_pt msgVersion, pubsub_msg_header_t *hdr);
+
+
+#endif //CELIX_PUBSUB_ZMQ_COMMON_H
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c
new file mode 100644
index 0000000..e26754c
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c
@@ -0,0 +1,684 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+
+#include <pubsub_serializer.h>
+#include <stdlib.h>
+#include <pubsub/subscriber.h>
+#include <memory.h>
+#include <pubsub_constants.h>
+#include <sys/epoll.h>
+#include <assert.h>
+#include <pubsub_endpoint.h>
+#include <arpa/inet.h>
+#include <czmq.h>
+#include <log_helper.h>
+#include "pubsub_zmq_topic_receiver.h"
+#include "pubsub_psa_zmq_constants.h"
+#include "pubsub_zmq_common.h"
+
+//TODO see if block and wakeup (reset) also works
+#define PSA_ZMQ_RECV_TIMEOUT 1000
+
+
+#define L_DEBUG(...) \
+ logHelper_log(receiver->logHelper, OSGI_LOGSERVICE_DEBUG, __VA_ARGS__)
+#define L_INFO(...) \
+ logHelper_log(receiver->logHelper, OSGI_LOGSERVICE_INFO, __VA_ARGS__)
+#define L_WARN(...) \
+ logHelper_log(receiver->logHelper, OSGI_LOGSERVICE_WARNING, __VA_ARGS__)
+#define L_ERROR(...) \
+ logHelper_log(receiver->logHelper, OSGI_LOGSERVICE_ERROR, __VA_ARGS__)
+
+struct pubsub_updmc_topic_receiver {
+ celix_bundle_context_t *ctx;
+ log_helper_t *logHelper;
+ char *scope;
+ char *topic;
+
+ zsock_t *zmqSocket;
+
+ //serialiser svc
+ long serializerTrackerId;
+ struct {
+
+ celix_thread_mutex_t mutex; //protect svc
+ pubsub_serializer_service_t *svc;
+ const celix_properties_t *props;
+ } serializer;
+
+ struct {
+ celix_thread_t thread;
+ celix_thread_mutex_t mutex;
+ bool running;
+ } recvThread;
+
+ struct {
+ celix_thread_mutex_t mutex;
+ hash_map_t *map; //key = zmq url, value = psa_zmq_requested_connection_entry_t*
+ } requestedConnections;
+
+ long subscriberTrackerId;
+ struct {
+ celix_thread_mutex_t mutex;
+ hash_map_t *map; //key = bnd id, value = psa_zmq_subscriber_entry_t
+ } subscribers;
+};
+
+typedef struct psa_zmq_requested_connection_entry {
+ char *url;
+ bool connected;
+} psa_zmq_requested_connection_entry_t;
+
+typedef struct psa_zmq_subscriber_entry {
+ int usageCount;
+ hash_map_t *msgTypes; //map from serializer svc
+ pubsub_subscriber_t *svc;
+} psa_zmq_subscriber_entry_t;
+
+typedef struct complete_zmq_msg {
+ zframe_t* header;
+ zframe_t* payload;
+} complete_zmq_msg_t;
+
+typedef struct mp_handle {
+ hash_map_pt svc_msg_db;
+ hash_map_pt rcv_msg_map;
+} mp_handle_t;
+
+typedef struct msg_map_entry {
+ bool retain;
+ void* msgInst;
+} msg_map_entry_t;
+
+
+static void pubsub_zmqTopicReceiver_setSerializer(void *handle, void *svc, const celix_properties_t *props);
+static void pubsub_zmqTopicReceiver_addSubscriber(void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *owner);
+static void pubsub_zmqTopicReceiver_removeSubscriber(void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *owner);
+static void psa_zmq_processMsg(pubsub_updmc_topic_receiver_t *receiver, celix_array_list_t *messages);
+static void* psa_zmq_recvThread(void * data);
+
+
+pubsub_updmc_topic_receiver_t* pubsub_zmqTopicReceiver_create(celix_bundle_context_t *ctx,
+ log_helper_t *logHelper,
+ const char *scope,
+ const char *topic,
+ long serializerSvcId) {
+ pubsub_updmc_topic_receiver_t *receiver = calloc(1, sizeof(*receiver));
+ receiver->ctx = ctx;
+ receiver->logHelper = logHelper;
+
+
+#ifdef BUILD_WITH_ZMQ_SECURITY
+ char* keys_bundle_dir = pubsub_getKeysBundleDir(bundle_context);
+ if (keys_bundle_dir == NULL){
+ return CELIX_SERVICE_EXCEPTION;
+ }
+
+ const char* keys_file_path = NULL;
+ const char* keys_file_name = NULL;
+ bundleContext_getProperty(bundle_context, PROPERTY_KEYS_FILE_PATH, &keys_file_path);
+ bundleContext_getProperty(bundle_context, PROPERTY_KEYS_FILE_NAME, &keys_file_name);
+
+ char sub_cert_path[MAX_CERT_PATH_LENGTH];
+ char pub_cert_path[MAX_CERT_PATH_LENGTH];
+
+ //certificate path ".cache/bundle{id}/version0.0/./META-INF/keys/subscriber/private/sub_{topic}.key.enc"
+ snprintf(sub_cert_path, MAX_CERT_PATH_LENGTH, "%s/META-INF/keys/subscriber/private/sub_%s.key.enc", keys_bundle_dir, topic);
+ snprintf(pub_cert_path, MAX_CERT_PATH_LENGTH, "%s/META-INF/keys/publisher/public/pub_%s.pub", keys_bundle_dir, topic);
+ free(keys_bundle_dir);
+
+ printf("PSA_ZMQ_PSA_ZMQ_TS: Loading subscriber key '%s'\n", sub_cert_path);
+ printf("PSA_ZMQ_PSA_ZMQ_TS: Loading publisher key '%s'\n", pub_cert_path);
+
+ zcert_t* sub_cert = get_zcert_from_encoded_file((char *) keys_file_path, (char *) keys_file_name, sub_cert_path);
+ if (sub_cert == NULL){
+ printf("PSA_ZMQ_PSA_ZMQ_TS: Cannot load key '%s'\n", sub_cert_path);
+ return CELIX_SERVICE_EXCEPTION;
+ }
+
+ zcert_t* pub_cert = zcert_load(pub_cert_path);
+ if (pub_cert == NULL){
+ zcert_destroy(&sub_cert);
+ printf("PSA_ZMQ_PSA_ZMQ_TS: Cannot load key '%s'\n", pub_cert_path);
+ return CELIX_SERVICE_EXCEPTION;
+ }
+
+ const char* pub_key = zcert_public_txt(pub_cert);
+#endif
+ receiver->zmqSocket = zsock_new (ZMQ_SUB);
+ if (receiver->zmqSocket == NULL) {
+#ifdef BUILD_WITH_ZMQ_SECURITY
+ zcert_destroy(&sub_cert);
+ zcert_destroy(&pub_cert);
+#endif
+ }
+
+ if (receiver->zmqSocket != NULL) {
+#ifdef BUILD_WITH_ZMQ_SECURITY
+ zcert_apply (sub_cert, zmq_s);
+ zsock_set_curve_serverkey (zmq_s, pub_key); //apply key of publisher to socket of subscriber
+#endif
+ zsock_set_subscribe(receiver->zmqSocket, topic);
+
+#ifdef BUILD_WITH_ZMQ_SECURITY
+ ts->zmq_cert = sub_cert;
+ ts->zmq_pub_cert = pub_cert;
+#endif
+ }
+
+ if (receiver->zmqSocket != NULL) {
+ receiver->scope = strndup(scope, 1024 * 1024);
+ receiver->topic = strndup(topic, 1024 * 1024);
+
+ celixThreadMutex_create(&receiver->serializer.mutex, NULL);
+ celixThreadMutex_create(&receiver->subscribers.mutex, NULL);
+ celixThreadMutex_create(&receiver->requestedConnections.mutex, NULL);
+ celixThreadMutex_create(&receiver->recvThread.mutex, NULL);
+
+ receiver->subscribers.map = hashMap_create(NULL, NULL, NULL, NULL);
+ receiver->requestedConnections.map = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
+ }
+
+ //track serializer svc based on the provided serializerSvcId
+ if (receiver->zmqSocket != NULL ) {
+ char filter[64];
+ snprintf(filter, 64, "(service.id=%li)", serializerSvcId);
+
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.serviceName = PUBSUB_SERIALIZER_SERVICE_NAME;
+ opts.filter.filter = filter;
+ opts.filter.ignoreServiceLanguage = true;
+ opts.callbackHandle = receiver;
+ opts.setWithProperties = pubsub_zmqTopicReceiver_setSerializer;
+ receiver->serializerTrackerId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
+ }
+
+ //track subscribers
+ if (receiver->zmqSocket != NULL ) {
+ int size = snprintf(NULL, 0, "(%s=%s)", PUBSUB_SUBSCRIBER_TOPIC, topic);
+ char buf[size+1];
+ snprintf(buf, (size_t)size+1, "(%s=%s)", PUBSUB_SUBSCRIBER_TOPIC, topic);
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.ignoreServiceLanguage = true;
+ opts.filter.serviceName = PUBSUB_SUBSCRIBER_SERVICE_NAME;
+ opts.filter.filter = buf;
+ opts.callbackHandle = receiver;
+ opts.addWithOwner = pubsub_zmqTopicReceiver_addSubscriber;
+ opts.removeWithOwner = pubsub_zmqTopicReceiver_removeSubscriber;
+
+ receiver->subscriberTrackerId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
+ }
+
+ if (receiver->zmqSocket != NULL ) {
+ receiver->recvThread.running = true;
+ celixThread_create(&receiver->recvThread.thread, NULL, psa_zmq_recvThread, receiver);
+ }
+
+ if (receiver->zmqSocket == NULL) {
+ free(receiver);
+ receiver = NULL;
+ L_ERROR("[PSA_ZMQ] Cannot create TopicReceiver for %s/%s", scope, topic);
+ }
+
+ return receiver;
+}
+
+void pubsub_zmqTopicReceiver_destroy(pubsub_updmc_topic_receiver_t *receiver) {
+ if (receiver != NULL) {
+
+ celixThreadMutex_lock(&receiver->subscribers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(receiver->subscribers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_zmq_subscriber_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (entry != NULL) {
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ if (receiver->serializer.svc != NULL) {
+ if (entry->msgTypes != NULL) {
+ receiver->serializer.svc->destroySerializerMap(receiver->serializer.svc->handle,
+ entry->msgTypes);
+ }
+ } else {
+ L_ERROR("Cannot find serializer for TopicReceiver %s/%s", receiver->scope, receiver->topic);
+ }
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+ free(entry);
+ }
+ }
+ hashMap_destroy(receiver->subscribers.map, false, false);
+ celixThreadMutex_unlock(&receiver->subscribers.mutex);
+
+
+ celix_bundleContext_stopTracker(receiver->ctx, receiver->subscriberTrackerId);
+ celix_bundleContext_stopTracker(receiver->ctx, receiver->serializerTrackerId);
+
+ celixThreadMutex_lock(&receiver->recvThread.mutex);
+ receiver->recvThread.running = false;
+ celixThreadMutex_unlock(&receiver->recvThread.mutex);
+ celixThread_join(receiver->recvThread.thread, NULL);
+
+ celixThreadMutex_lock(&receiver->requestedConnections.mutex);
+ iter = hashMapIterator_construct(receiver->requestedConnections.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_zmq_requested_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (entry != NULL) {
+ free(entry->url);
+ free(entry);
+ }
+ }
+ hashMap_destroy(receiver->requestedConnections.map, false, false);
+ celixThreadMutex_unlock(&receiver->requestedConnections.mutex);
+
+ celixThreadMutex_destroy(&receiver->serializer.mutex);
+ celixThreadMutex_destroy(&receiver->subscribers.mutex);
+ celixThreadMutex_destroy(&receiver->requestedConnections.mutex);
+ celixThreadMutex_destroy(&receiver->recvThread.mutex);
+
+ zsock_destroy(&receiver->zmqSocket);
+ }
+ free(receiver);
+}
+
+const char* pubsub_zmqTopicReceiver_psaType(pubsub_updmc_topic_receiver_t *receiver) {
+ return PSA_ZMQ_PUBSUB_ADMIN_TYPE;
+}
+
+const char* pubsub_zmqTopicReceiver_serializerType(pubsub_updmc_topic_receiver_t *receiver) {
+ const char *result = NULL;
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ if (receiver->serializer.props != NULL) {
+ result = celix_properties_get(receiver->serializer.props, PUBSUB_SERIALIZER_TYPE_KEY, NULL);
+ }
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+ return result;
+}
+
+const char* pubsub_zmqTopicReceiver_scope(pubsub_updmc_topic_receiver_t *receiver) {
+ return receiver->scope;
+}
+const char* pubsub_zmqTopicReceiver_topic(pubsub_updmc_topic_receiver_t *receiver) {
+ return receiver->topic;
+}
+
+void pubsub_zmqTopicReceiver_connectTo(
+ pubsub_updmc_topic_receiver_t *receiver,
+ const char *url) {
+ L_DEBUG("[PSA_ZMQ] TopicReceiver %s/%s connecting to zmq url %s", receiver->scope, receiver->topic, url);
+
+ celixThreadMutex_lock(&receiver->requestedConnections.mutex);
+ psa_zmq_requested_connection_entry_t *entry = hashMap_get(receiver->requestedConnections.map, url);
+ if (entry == NULL) {
+ entry = calloc(1, sizeof(*entry));
+ entry->url = strndup(url, 1024*1024);
+ entry->connected = false;
+ hashMap_put(receiver->requestedConnections.map, (void*)entry->url, entry);
+ }
+ if (!entry->connected) {
+ if (zsock_connect(receiver->zmqSocket, "%s", url) == 0) {
+ entry->connected = true;
+ } else {
+ L_WARN("[PSA_ZMQ] Error connecting to zmq url %s. (%s)", url, strerror(errno));
+ }
+ }
+ celixThreadMutex_unlock(&receiver->requestedConnections.mutex);
+}
+
+void pubsub_zmqTopicReceiver_disconnectFrom(pubsub_updmc_topic_receiver_t *receiver, const char *url) {
+ L_DEBUG("[PSA ZMQ] TopicReceiver %s/%s disconnect from zmq url %s", receiver->scope, receiver->topic, url);
+
+ celixThreadMutex_lock(&receiver->requestedConnections.mutex);
+ psa_zmq_requested_connection_entry_t *entry = hashMap_remove(receiver->requestedConnections.map, url);
+ if (entry != NULL && entry->connected) {
+ if (zsock_disconnect(receiver->zmqSocket,"%s",url) == 0) {
+ entry->connected = false;
+ } else {
+ L_WARN("[PSA_ZMQ] Error disconnecting from zmq url %s. (%s)", url, strerror(errno));
+ }
+ }
+ if (entry != NULL) {
+ free(entry->url);
+ free(entry);
+ }
+ celixThreadMutex_unlock(&receiver->requestedConnections.mutex);
+}
+
+
+static void pubsub_zmqTopicReceiver_setSerializer(void *handle, void *svc, const celix_properties_t *props) {
+ pubsub_updmc_topic_receiver_t *receiver = handle;
+ pubsub_serializer_service_t *ser = svc;
+
+ //check if current serializer is set, if so destroy existing type maps
+ celixThreadMutex_lock(&receiver->subscribers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(receiver->subscribers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_zmq_subscriber_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (entry != NULL && entry->msgTypes != NULL) {
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ if (receiver->serializer.svc != NULL) {
+ receiver->serializer.svc->destroySerializerMap(receiver->serializer.svc->handle, entry->msgTypes);
+ }
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+ entry->msgTypes = NULL;
+ }
+ }
+ celixThreadMutex_unlock(&receiver->subscribers.mutex);
+
+ if (ser == NULL) {
+ //TODO -> no serializer -> remove all publishers
+ }
+
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ receiver->serializer.svc = ser;
+ receiver->serializer.props = props;
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+}
+
+static void pubsub_zmqTopicReceiver_addSubscriber(void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *bnd) {
+ pubsub_updmc_topic_receiver_t *receiver = handle;
+
+ long bndId = celix_bundle_getId(bnd);
+ const char *subScope = celix_properties_get(props, PUBSUB_SUBSCRIBER_SCOPE, "default");
+ if (strncmp(subScope, receiver->scope, strlen(receiver->scope)) != 0) {
+ //not the same scope. ignore
+ return;
+ }
+
+ celixThreadMutex_lock(&receiver->subscribers.mutex);
+ psa_zmq_subscriber_entry_t *entry = hashMap_get(receiver->subscribers.map, (void*)bndId);
+ if (entry != NULL) {
+ entry->usageCount += 1;
+ } else {
+ //new create entry
+ entry = calloc(1, sizeof(*entry));
+ entry->usageCount = 1;
+ entry->svc = svc;
+
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ if (receiver->serializer.svc != NULL) {
+ receiver->serializer.svc->createSerializerMap(receiver->serializer.svc->handle, (celix_bundle_t*)bnd, &entry->msgTypes);
+ } else {
+ fprintf(stderr, "Cannot find serializer for TopicReceiver %s/%s", receiver->scope, receiver->topic);
+ }
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+
+ hashMap_put(receiver->subscribers.map, (void*)bndId, entry);
+
+ }
+ celixThreadMutex_unlock(&receiver->subscribers.mutex);
+}
+
+static void pubsub_zmqTopicReceiver_removeSubscriber(void *handle, void *svc, const celix_properties_t *props, const celix_bundle_t *bnd) {
+ pubsub_updmc_topic_receiver_t *receiver = handle;
+
+ long bndId = celix_bundle_getId(bnd);
+
+ celixThreadMutex_lock(&receiver->subscribers.mutex);
+ psa_zmq_subscriber_entry_t *entry = hashMap_get(receiver->subscribers.map, (void*)bndId);
+ if (entry != NULL) {
+ entry->usageCount -= 1;
+ }
+ if (entry != NULL && entry->usageCount <= 0) {
+ //remove entry
+ hashMap_remove(receiver->subscribers.map, (void*)bndId);
+ celixThreadMutex_lock(&receiver->serializer.mutex);
+ if (receiver->serializer.svc != NULL) {
+ receiver->serializer.svc->destroySerializerMap(receiver->serializer.svc->handle, entry->msgTypes);
+ } else {
+ fprintf(stderr, "Cannot find serializer for TopicReceiver %s/%s", receiver->scope, receiver->topic);
+ }
+ celixThreadMutex_unlock(&receiver->serializer.mutex);
+ free(entry);
+ }
+ celixThreadMutex_unlock(&receiver->subscribers.mutex);
+}
+
+static void* psa_zmq_recvThread(void * data) {
+ pubsub_updmc_topic_receiver_t *receiver = data;
+
+ celixThreadMutex_lock(&receiver->recvThread.mutex);
+ bool running = receiver->recvThread.running;
+ celixThreadMutex_unlock(&receiver->recvThread.mutex);
+
+ while (running) {
+ zframe_t* headerMsg = zframe_recv_nowait(receiver->zmqSocket);
+ if (headerMsg == NULL) {
+ if(errno == EAGAIN) {
+ usleep(PSA_ZMQ_RECV_TIMEOUT);
+ } else if (errno == EINTR) {
+ //It means we got a signal and we have to exit...
+ L_INFO("PSA_ZMQ_TS: header_recv thread for topic got a signal and will exit.\n");
+ } else {
+ L_ERROR("PSA_ZMQ_TS: header_recv thread (%s)", strerror(errno));
+ }
+ } else {
+ pubsub_msg_header_t *hdr = (pubsub_msg_header_t*)zframe_data(headerMsg);
+
+ if (zframe_more(headerMsg)) {
+ zframe_t* payloadMsg = zframe_recv(receiver->zmqSocket);
+ if (payloadMsg == NULL) {
+ if (errno == EINTR) {
+ //It means we got a signal and we have to exit...
+ L_INFO("[PSA_ZMQ] TopicSender payload_recv thread for topic got a signal and will exit.\n");
+ } else {
+ L_WARN("[PSA_ZMQ] TopicSender: payload_recv (%s)", strerror(errno));
+ }
+ zframe_destroy(&headerMsg);
+ } else {
+
+ //Let's fetch all the messages from the socket
+ celix_array_list_t *msg_list = celix_arrayList_create();
+ complete_zmq_msg_t *firstMsg = calloc(1, sizeof(struct complete_zmq_msg));
+ firstMsg->header = headerMsg;
+ firstMsg->payload = payloadMsg;
+ arrayList_add(msg_list, firstMsg);
+
+ bool more = (bool)zframe_more(payloadMsg);
+ while (more) {
+
+ zframe_t* h_msg = zframe_recv(receiver->zmqSocket);
+ if (h_msg == NULL) {
+ if (errno == EINTR) {
+ //It means we got a signal and we have to exit...
+ L_INFO("[PSA_ZMQ] TopicSender payload_recv thread for topic got a signal and will exit.\n");
+ } else {
+ L_WARN("[PSA_ZMQ] TopicSender: payload_recv (%s)", strerror(errno));
+ }
+ break;
+ }
+
+ zframe_t* p_msg = zframe_recv(receiver->zmqSocket);
+ if (p_msg == NULL) {
+ if (errno == EINTR) {
+ //It means we got a signal and we have to exit...
+ L_INFO("[PSA_ZMQ] TopicSender payload_recv thread for topic got a signal and will exit.\n");
+ } else {
+ L_WARN("[PSA_ZMQ] TopicSender: payload_recv (%s)", strerror(errno));
+ }
+ zframe_destroy(&h_msg);
+ break;
+ }
+
+ complete_zmq_msg_t *c_msg = calloc(1, sizeof(struct complete_zmq_msg));
+ c_msg->header = h_msg;
+ c_msg->payload = p_msg;
+ arrayList_add(msg_list, c_msg);
+
+ if (!zframe_more(p_msg)) {
+ more = false;
+ }
+ }
+
+ psa_zmq_processMsg(receiver, msg_list);
+ }
+
+ } else {
+ free(headerMsg);
+ L_WARN("PSA_ZMQ_TS: received message %u for topic %s without payload!\n", hdr->type, hdr->topic);
+ }
+
+ }
+
+ celixThreadMutex_lock(&receiver->recvThread.mutex);
+ running = receiver->recvThread.running;
+ celixThreadMutex_unlock(&receiver->recvThread.mutex);
+ } // while
+
+ return NULL;
+}
+
+static int psa_zmq_getMultipart(void *handle, unsigned int msgTypeId, bool retain, void **part){
+
+ if(handle==NULL){
+ *part = NULL;
+ return -1;
+ }
+
+ mp_handle_t *mp_handle = handle;
+ msg_map_entry_t *entry = hashMap_get(mp_handle->rcv_msg_map, (void*)(uintptr_t) msgTypeId);
+ if(entry!=NULL){
+ entry->retain = retain;
+ *part = entry->msgInst;
+ }
+ else{
+ printf("TP: getMultipart cannot find msg '%u'\n",msgTypeId);
+ *part=NULL;
+ return -2;
+ }
+
+ return 0;
+
+}
+
+static void psa_zmq_destroyMultipartHandle(mp_handle_t *mp_handle) {
+
+ hash_map_iterator_pt iter = hashMapIterator_create(mp_handle->rcv_msg_map);
+ while(hashMapIterator_hasNext(iter)){
+ hash_map_entry_pt entry = hashMapIterator_nextEntry(iter);
+ unsigned int msgId = (unsigned int)(uintptr_t)hashMapEntry_getKey(entry);
+ msg_map_entry_t *msgEntry = hashMapEntry_getValue(entry);
+ pubsub_msg_serializer_t* msgSer = hashMap_get(mp_handle->svc_msg_db, (void*)(uintptr_t)msgId);
+
+ if(msgSer!=NULL){
+ if(!msgEntry->retain){
+ msgSer->freeMsg(msgSer->handle,msgEntry->msgInst);
+ }
+ }
+ else{
+ printf("PSA_ZMQ_TS: ERROR: Cannot find messageSerializer for msg %u, so cannot destroy it!\n",msgId);
+ }
+
+ free(msgEntry);
+ }
+ hashMapIterator_destroy(iter);
+
+ hashMap_destroy(mp_handle->rcv_msg_map,false,false);
+ free(mp_handle);
+}
+
+static mp_handle_t* psa_zmq_createMultipartHandle(hash_map_pt svc_msg_db,array_list_pt rcv_msg_list){
+
+ if(arrayList_size(rcv_msg_list)==1){ //Means it's not a multipart message
+ return NULL;
+ }
+
+ mp_handle_t *mp_handle = calloc(1,sizeof(struct mp_handle));
+ mp_handle->svc_msg_db = svc_msg_db;
+ mp_handle->rcv_msg_map = hashMap_create(NULL, NULL, NULL, NULL);
+
+ int i=1; //We skip the first message, it will be handle differently
+ for(;i<arrayList_size(rcv_msg_list);i++){
+ complete_zmq_msg_t *c_msg = celix_arrayList_get(rcv_msg_list,i);
+ pubsub_msg_header_t *header = (pubsub_msg_header_t*)zframe_data(c_msg->header);
+
+ pubsub_msg_serializer_t* msgSer = hashMap_get(svc_msg_db, (void*)(uintptr_t)(header->type));
+
+ if (msgSer!= NULL) {
+ void *msgInst = NULL;
+
+ bool validVersion = psa_zmq_checkVersion(msgSer->msgVersion,header);
+
+ if(validVersion){
+ celix_status_t status = msgSer->deserialize(msgSer, (const void*)zframe_data(c_msg->payload), 0, &msgInst);
+
+ if(status == CELIX_SUCCESS){
+ msg_map_entry_t *entry = calloc(1,sizeof(struct msg_map_entry));
+ entry->msgInst = msgInst;
+ hashMap_put(mp_handle->rcv_msg_map, (void*)(uintptr_t)header->type,entry);
+ }
+ }
+ }
+ }
+
+ return mp_handle;
+
+}
+
+
+static void psa_zmq_processMsg(pubsub_updmc_topic_receiver_t *receiver, celix_array_list_t *messages) {
+
+ pubsub_msg_header_t *first_msg_hdr = (pubsub_msg_header_t*)zframe_data(((complete_zmq_msg_t*)celix_arrayList_get(messages,0))->header);
+
+ celixThreadMutex_lock(&receiver->subscribers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(receiver->subscribers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_zmq_subscriber_entry_t *entry = hashMapIterator_nextValue(&iter);
+
+ pubsub_msg_serializer_t *msgSer = NULL;
+ if (entry->msgTypes != NULL) {
+ msgSer = hashMap_get(entry->msgTypes, (void *) (uintptr_t) first_msg_hdr->type);
+ }
+ if (msgSer == NULL) {
+ printf("[PSA_ZMQ] Serializer not available for message %d.\n", first_msg_hdr->type);
+ } else {
+ void *msgInst = NULL;
+ bool validVersion = psa_zmq_checkVersion(msgSer->msgVersion,first_msg_hdr);
+
+ if (validVersion) {
+
+ celix_status_t status = msgSer->deserialize(msgSer, (const void *) zframe_data(((complete_zmq_msg_t*)celix_arrayList_get(messages,0))->payload), 0, &msgInst);
+ if (status == CELIX_SUCCESS) {
+ bool release = true;
+ mp_handle_t *mp_handle = psa_zmq_createMultipartHandle(entry->msgTypes, messages);
+ pubsub_multipart_callbacks_t mp_callbacks;
+ mp_callbacks.handle = mp_handle;
+ mp_callbacks.localMsgTypeIdForMsgType = psa_zmq_localMsgTypeIdForMsgType;
+ mp_callbacks.getMultipart = psa_zmq_getMultipart;
+ entry->svc->receive(entry->svc->handle, msgSer->msgName, first_msg_hdr->type, msgInst, &mp_callbacks, &release);
+
+ if (release) {
+ msgSer->freeMsg(msgSer,msgInst); // pubsubSerializer_freeMsg(msgType, msgInst);
+ }
+ if (mp_handle!=NULL) {
+ psa_zmq_destroyMultipartHandle(mp_handle);
+ }
+ }
+ else{
+ L_WARN("[PSA_ZMQ] TopicReceiver Cannot deserialize msgType %s.\n",msgSer->msgName);
+ }
+
+ } else {
+ int major=0,minor=0;
+ version_getMajor(msgSer->msgVersion,&major);
+ version_getMinor(msgSer->msgVersion,&minor);
+ L_WARN("[PSA_ZMQ] TopicReceiver: Version mismatch for primary message '%s' (have %d.%d, received %u.%u). NOT sending any part of the whole message.\n",
+ msgSer->msgName,major,minor,first_msg_hdr->major,first_msg_hdr->minor);
+ }
+ }
+ }
+ celixThreadMutex_unlock(&receiver->subscribers.mutex);
+}
http://git-wip-us.apache.org/repos/asf/celix/blob/00454efd/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h
new file mode 100644
index 0000000..269d1db
--- /dev/null
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h
@@ -0,0 +1,41 @@
+/**
+ *Licensed to the Apache Software Foundation (ASF) under one
+ *or more contributor license agreements. See the NOTICE file
+ *distributed with this work for additional information
+ *regarding copyright ownership. The ASF licenses this file
+ *to you under the Apache License, Version 2.0 (the
+ *"License"); you may not use this file except in compliance
+ *with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *Unless required by applicable law or agreed to in writing,
+ *software distributed under the License is distributed on an
+ *"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ *specific language governing permissions and limitations
+ *under the License.
+ */
+#ifndef CELIX_PUBSUB_ZMQ_TOPIC_RECEIVER_H
+#define CELIX_PUBSUB_ZMQ_TOPIC_RECEIVER_H
+
+#include "celix_bundle_context.h"
+
+typedef struct pubsub_updmc_topic_receiver pubsub_updmc_topic_receiver_t;
+
+pubsub_updmc_topic_receiver_t* pubsub_zmqTopicReceiver_create(celix_bundle_context_t *ctx,
+ log_helper_t *logHelper,
+ const char *scope,
+ const char *topic,
+ long serializerSvcId);
+void pubsub_zmqTopicReceiver_destroy(pubsub_updmc_topic_receiver_t *receiver);
+
+const char* pubsub_zmqTopicReceiver_psaType(pubsub_updmc_topic_receiver_t *receiver);
+const char* pubsub_zmqTopicReceiver_serializerType(pubsub_updmc_topic_receiver_t *receiver);
+const char* pubsub_zmqTopicReceiver_scope(pubsub_updmc_topic_receiver_t *receiver);
+const char* pubsub_zmqTopicReceiver_topic(pubsub_updmc_topic_receiver_t *receiver);
+
+void pubsub_zmqTopicReceiver_connectTo(pubsub_updmc_topic_receiver_t *receiver, const char *url);
+void pubsub_zmqTopicReceiver_disconnectFrom(pubsub_updmc_topic_receiver_t *receiver, const char *url);
+
+#endif //CELIX_PUBSUB_ZMQ_TOPIC_RECEIVER_H