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/09/28 17:33:38 UTC
[1/2] celix git commit: CELIX-434: Updates cache behaviour.
Repository: celix
Updated Branches:
refs/heads/feature/CELIX-454-pubsub-disc b635e2749 -> dac110224
CELIX-434: Updates cache behaviour.
If no cache dir is specified a tmp dir in /tmp/celix-cache-<progname>-<fwUUID> will be created
and used for the bundle caches. This directory is also removed if the program exists normally.
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/d311fa7e
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/d311fa7e
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/d311fa7e
Branch: refs/heads/feature/CELIX-454-pubsub-disc
Commit: d311fa7edfc39eed52807b49e32573a21ea4b20c
Parents: b635e27
Author: Pepijn Noltes <pe...@gmail.com>
Authored: Fri Sep 28 19:31:45 2018 +0200
Committer: Pepijn Noltes <pe...@gmail.com>
Committed: Fri Sep 28 19:31:45 2018 +0200
----------------------------------------------------------------------
libs/framework/private/mock/bundle_cache_mock.c | 2 +-
.../private/test/bundle_cache_test.cpp | 2 +-
libs/framework/src/bundle_cache.c | 44 +++++++++++++++-----
libs/framework/src/bundle_cache.h | 3 +-
libs/framework/src/bundle_cache_private.h | 1 +
libs/framework/src/framework.c | 20 ++++-----
6 files changed, 47 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/private/mock/bundle_cache_mock.c
----------------------------------------------------------------------
diff --git a/libs/framework/private/mock/bundle_cache_mock.c b/libs/framework/private/mock/bundle_cache_mock.c
index 169a256..94ae731 100644
--- a/libs/framework/private/mock/bundle_cache_mock.c
+++ b/libs/framework/private/mock/bundle_cache_mock.c
@@ -27,7 +27,7 @@
#include "bundle_cache.h"
-celix_status_t bundleCache_create(properties_pt configurationMap, bundle_cache_pt *bundle_cache) {
+celix_status_t bundleCache_create(const char *fwUUID, properties_pt configurationMap, bundle_cache_pt *bundle_cache) {
mock_c()->actualCall("bundleCache_create")
->withPointerParameters("configurationMap", configurationMap)
->withOutputParameter("bundle_cache", bundle_cache);
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/private/test/bundle_cache_test.cpp
----------------------------------------------------------------------
diff --git a/libs/framework/private/test/bundle_cache_test.cpp b/libs/framework/private/test/bundle_cache_test.cpp
index 13e0e80..bfdc6af 100644
--- a/libs/framework/private/test/bundle_cache_test.cpp
+++ b/libs/framework/private/test/bundle_cache_test.cpp
@@ -64,7 +64,7 @@ TEST(bundle_cache, create) {
.andReturnValue((char *) NULL);
bundle_cache_pt cache = NULL;
- LONGS_EQUAL(CELIX_SUCCESS, bundleCache_create(configuration, &cache));
+ LONGS_EQUAL(CELIX_SUCCESS, bundleCache_create("test-uuid", configuration, &cache));
LONGS_EQUAL(CELIX_SUCCESS, bundleCache_destroy(&cache));
}
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/src/bundle_cache.c
----------------------------------------------------------------------
diff --git a/libs/framework/src/bundle_cache.c b/libs/framework/src/bundle_cache.c
index 6de8ab7..054aa80 100644
--- a/libs/framework/src/bundle_cache.c
+++ b/libs/framework/src/bundle_cache.c
@@ -16,28 +16,34 @@
*specific language governing permissions and limitations
*under the License.
*/
-/*
- * bundle_cache.c
- *
- * \date Aug 6, 2010
- * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- * \copyright Apache License, Version 2.0
- */
+
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <dirent.h>
#include <unistd.h>
+#include <fcntl.h>
#include "bundle_cache_private.h"
#include "bundle_archive.h"
#include "constants.h"
#include "celix_log.h"
+#include "celix_properties.h"
static celix_status_t bundleCache_deleteTree(bundle_cache_pt cache, char * directory);
-celix_status_t bundleCache_create(properties_pt configurationMap, bundle_cache_pt *bundle_cache) {
+static const char* bundleCache_progamName() {
+#if defined(__APPLE__) || defined(__FreeBSD__)
+ return getprogname();
+#elif defined(_GNU_SOURCE)
+ return program_invocation_short_name;
+#else
+ return "";
+#endif
+}
+
+celix_status_t bundleCache_create(const char *fwUUID, celix_properties_t *configurationMap, bundle_cache_pt *bundle_cache) {
celix_status_t status;
bundle_cache_pt cache;
@@ -52,9 +58,22 @@ celix_status_t bundleCache_create(properties_pt configurationMap, bundle_cache_p
char* cacheDir = (char*)properties_get(configurationMap, (char *) OSGI_FRAMEWORK_FRAMEWORK_STORAGE);
cache->configurationMap = configurationMap;
if (cacheDir == NULL) {
- cacheDir = ".cache";
+ //Using /tmp dir for cache, so that multiple frameworks can be launched
+ //instead of cacheDir = ".cache";
+ const char *pg = bundleCache_progamName();
+ if (pg == NULL) {
+ pg = "";
+ }
+ size_t len = (size_t)snprintf(NULL, 0, "/tmp/celix-cache-%s-%s",pg, fwUUID) + 1;
+ char *tmpdir = calloc(len, sizeof(char));
+ snprintf(tmpdir, len, "/tmp/celix-cache-%s-%s", pg, fwUUID);
+
+ cache->cacheDir = tmpdir;
+ cache->deleteOnDestroy = true;
+ } else {
+ cache->cacheDir = strdup(cacheDir);
+ cache->deleteOnDestroy = false;
}
- cache->cacheDir = cacheDir;
*bundle_cache = cache;
status = CELIX_SUCCESS;
@@ -66,7 +85,10 @@ celix_status_t bundleCache_create(properties_pt configurationMap, bundle_cache_p
}
celix_status_t bundleCache_destroy(bundle_cache_pt *cache) {
-
+ if ((*cache)->deleteOnDestroy) {
+ bundleCache_delete(*cache);
+ }
+ free((*cache)->cacheDir);
free(*cache);
*cache = NULL;
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/src/bundle_cache.h
----------------------------------------------------------------------
diff --git a/libs/framework/src/bundle_cache.h b/libs/framework/src/bundle_cache.h
index 350f33b..c9de996 100644
--- a/libs/framework/src/bundle_cache.h
+++ b/libs/framework/src/bundle_cache.h
@@ -47,14 +47,13 @@ typedef struct bundleCache *bundle_cache_pt;
* Creates the bundle cache using the supplied configuration map.
*
* @param configurationMap Set with properties to use for this cache
- * @param mp The memory pool to use for allocation the cache
* @param bundle_cache Output parameter for the created cache
* @return Status code indication failure or success:
* - CELIX_SUCCESS when no errors are encountered.
* - CELIX_ILLEGAL_ARGUMENT If <code>bundle_cache</code> not is null.
* - CELIX_ENOMEM If allocating memory for <code>bundle_cache</code> failed.
*/
-celix_status_t bundleCache_create(properties_pt configurationMap, bundle_cache_pt *bundle_cache);
+celix_status_t bundleCache_create(const char *fwUUID, properties_pt configurationMap, bundle_cache_pt *bundle_cache);
/**
* Frees the bundle_cache memory allocated in bundleCache_create
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/src/bundle_cache_private.h
----------------------------------------------------------------------
diff --git a/libs/framework/src/bundle_cache_private.h b/libs/framework/src/bundle_cache_private.h
index 69e26ee..bb701d9 100644
--- a/libs/framework/src/bundle_cache_private.h
+++ b/libs/framework/src/bundle_cache_private.h
@@ -33,6 +33,7 @@
struct bundleCache {
properties_pt configurationMap;
char * cacheDir;
+ bool deleteOnDestroy;
};
http://git-wip-us.apache.org/repos/asf/celix/blob/d311fa7e/libs/framework/src/framework.c
----------------------------------------------------------------------
diff --git a/libs/framework/src/framework.c b/libs/framework/src/framework.c
index b0d505d..4284f9a 100644
--- a/libs/framework/src/framework.c
+++ b/libs/framework/src/framework.c
@@ -433,6 +433,15 @@ celix_status_t fw_init(framework_pt framework) {
array_list_pt archives = NULL;
bundle_archive_pt archive = NULL;
+ /*create and store framework uuid*/
+ char uuid[37];
+
+ uuid_t uid;
+ uuid_generate(uid);
+ uuid_unparse(uid, uuid);
+
+ properties_set(framework->configurationMap, (char*) OSGI_FRAMEWORK_FRAMEWORK_UUID, uuid);
+
celix_status_t status = CELIX_SUCCESS;
status = CELIX_DO_IF(status, framework_acquireBundleLock(framework, framework->bundle, OSGI_FRAMEWORK_BUNDLE_INSTALLED|OSGI_FRAMEWORK_BUNDLE_RESOLVED|OSGI_FRAMEWORK_BUNDLE_STARTING|OSGI_FRAMEWORK_BUNDLE_ACTIVE));
status = CELIX_DO_IF(status, arrayList_create(&framework->serviceListeners)); //entry is celix_fw_service_listener_entry_t
@@ -444,7 +453,7 @@ celix_status_t fw_init(framework_pt framework) {
if (status == CELIX_SUCCESS) {
if ((state == OSGI_FRAMEWORK_BUNDLE_INSTALLED) || (state == OSGI_FRAMEWORK_BUNDLE_RESOLVED)) {
bundle_state_e state;
- status = CELIX_DO_IF(status, bundleCache_create(framework->configurationMap,&framework->cache));
+ status = CELIX_DO_IF(status, bundleCache_create(uuid, framework->configurationMap,&framework->cache));
status = CELIX_DO_IF(status, bundle_getState(framework->bundle, &state));
if (status == CELIX_SUCCESS) {
if (state == OSGI_FRAMEWORK_BUNDLE_INSTALLED) {
@@ -458,15 +467,6 @@ celix_status_t fw_init(framework_pt framework) {
}
if (status == CELIX_SUCCESS) {
- /*create and store framework uuid*/
- char uuid[37];
-
- uuid_t uid;
- uuid_generate(uid);
- uuid_unparse(uid, uuid);
-
- properties_set(framework->configurationMap, (char*) OSGI_FRAMEWORK_FRAMEWORK_UUID, uuid);
-
framework->installedBundleMap = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
}
[2/2] celix git commit: CELIX-454: More bug and mem leak fixing
Posted by pn...@apache.org.
CELIX-454: More bug and mem leak fixing
Project: http://git-wip-us.apache.org/repos/asf/celix/repo
Commit: http://git-wip-us.apache.org/repos/asf/celix/commit/dac11022
Tree: http://git-wip-us.apache.org/repos/asf/celix/tree/dac11022
Diff: http://git-wip-us.apache.org/repos/asf/celix/diff/dac11022
Branch: refs/heads/feature/CELIX-454-pubsub-disc
Commit: dac110224355502a22f27987045ab59fe87d1372
Parents: d311fa7
Author: Pepijn Noltes <pe...@gmail.com>
Authored: Fri Sep 28 19:33:19 2018 +0200
Committer: Pepijn Noltes <pe...@gmail.com>
Committed: Fri Sep 28 19:33:19 2018 +0200
----------------------------------------------------------------------
.../pubsub_admin_zmq/src/pubsub_zmq_admin.c | 89 +++++++++++++-------
.../src/pubsub_zmq_topic_receiver.c | 27 ++++++
.../src/pubsub_zmq_topic_receiver.h | 1 +
.../src/pubsub_zmq_topic_sender.c | 3 +-
.../src/pubsub_discovery_impl.c | 8 +-
.../src/pstm_activator.c | 4 +
.../src/pubsub_topology_manager.c | 5 +-
libs/etcdlib/src/etcd.c | 1 +
libs/utils/src/celix_threads.c | 3 +-
9 files changed, 104 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/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
index 5b6c384..90788e1 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_admin.c
@@ -87,7 +87,7 @@ typedef struct psa_zmq_serializer_entry {
pubsub_serializer_service_t *svc;
} psa_zmq_serializer_entry_t;
-static celix_status_t zmq_getIpAddress(const char* interface, const char** ip);
+static celix_status_t zmq_getIpAddress(const char* interface, char** ip);
static celix_status_t pubsub_zmqAdmin_connectEndpointToReceiver(pubsub_zmq_admin_t* psa, pubsub_zmq_topic_receiver_t *receiver, const celix_properties_t *endpoint);
static celix_status_t pubsub_zmqAdmin_disconnectEndpointFromReceiver(pubsub_zmq_admin_t* psa, pubsub_zmq_topic_receiver_t *receiver, const celix_properties_t *endpoint);
@@ -99,7 +99,11 @@ pubsub_zmq_admin_t* pubsub_zmqAdmin_create(celix_bundle_context_t *ctx, log_help
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);
+ char *ip = NULL;
+ const char *confIp = celix_bundleContext_getProperty(ctx, PUBSUB_ZMQ_PSA_IP_KEY , NULL);
+ if (confIp != NULL) {
+ ip = strndup(confIp, 1024);
+ }
if (ip == NULL) {
//TODO try to get ip from subnet (CIDR)
@@ -113,10 +117,10 @@ pubsub_zmq_admin_t* pubsub_zmqAdmin_create(celix_bundle_context_t *ctx, log_help
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;
+ ip = strndup(PUBSUB_ZMQ_DEFAULT_IP, 1024);
}
- psa->ipAddress = strndup(ip, 1024*1024);
+ psa->ipAddress = ip;
if (psa->verbose) {
L_INFO("[PSA_ZMQ] Using %s for service annunciation", ip);
}
@@ -298,6 +302,8 @@ void pubsub_zmqAdmin_removeSerializerSvc(void *handle, void *svc, const celix_pr
}
}
celixThreadMutex_unlock(&psa->topicReceivers.mutex);
+
+ free(entry);
}
celixThreadMutex_unlock(&psa->serializers.mutex);
}
@@ -501,19 +507,17 @@ static celix_status_t pubsub_zmqAdmin_connectEndpointToReceiver(pubsub_zmq_admin
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");
+ if (url == NULL) {
+ L_WARN("[PSA ZMQ] Error got endpoint without a zmq url");
status = CELIX_BUNDLE_EXCEPTION;
} else {
- if (eScope != NULL && eTopic != NULL && type != NULL &&
+ if (eScope != NULL && eTopic != 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) {
+ strncmp(eTopic, topic, 1024 * 1024) == 0) {
pubsub_zmqTopicReceiver_connectTo(receiver, url);
}
}
@@ -523,13 +527,18 @@ static celix_status_t pubsub_zmqAdmin_connectEndpointToReceiver(pubsub_zmq_admin
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_zmq_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
- pubsub_zmqAdmin_connectEndpointToReceiver(psa, receiver, endpoint);
+
+ const char *type = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TYPE, NULL);
+
+ if (type != NULL && strncmp(PUBSUB_PUBLISHER_ENDPOINT_TYPE, type, strlen(PUBSUB_PUBLISHER_ENDPOINT_TYPE)) == 0) {
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_zmq_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
+ pubsub_zmqAdmin_connectEndpointToReceiver(psa, receiver, endpoint);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
}
- celixThreadMutex_unlock(&psa->topicReceivers.mutex);
celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
celix_properties_t *cpy = celix_properties_copy(endpoint);
@@ -549,19 +558,17 @@ static celix_status_t pubsub_zmqAdmin_disconnectEndpointFromReceiver(pubsub_zmq_
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");
+ if (url == NULL) {
+ L_WARN("[PSA ZMQ] Error got endpoint without zmq url");
status = CELIX_BUNDLE_EXCEPTION;
} else {
- if (eScope != NULL && eTopic != NULL && type != NULL &&
+ if (eScope != NULL && eTopic != 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) {
+ strncmp(eTopic, topic, 1024 * 1024) == 0) {
pubsub_zmqTopicReceiver_disconnectFrom(receiver, url);
}
}
@@ -571,13 +578,18 @@ static celix_status_t pubsub_zmqAdmin_disconnectEndpointFromReceiver(pubsub_zmq_
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_zmq_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
- pubsub_zmqAdmin_disconnectEndpointFromReceiver(psa, receiver, endpoint);
+
+ const char *type = celix_properties_get(endpoint, PUBSUB_ENDPOINT_TYPE, NULL);
+
+ if (type != NULL && strncmp(PUBSUB_PUBLISHER_ENDPOINT_TYPE, type, strlen(PUBSUB_PUBLISHER_ENDPOINT_TYPE)) == 0) {
+ celixThreadMutex_lock(&psa->topicReceivers.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(psa->topicReceivers.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ pubsub_zmq_topic_receiver_t *receiver = hashMapIterator_nextValue(&iter);
+ pubsub_zmqAdmin_disconnectEndpointFromReceiver(psa, receiver, endpoint);
+ }
+ celixThreadMutex_unlock(&psa->topicReceivers.mutex);
}
- celixThreadMutex_unlock(&psa->topicReceivers.mutex);
celixThreadMutex_lock(&psa->discoveredEndpoints.mutex);
const char *uuid = celix_properties_get(endpoint, PUBSUB_ENDPOINT_UUID, NULL);
@@ -628,20 +640,35 @@ celix_status_t pubsub_zmqAdmin_executeCommand(void *handle, char *commandLine __
const char *serType = serEntry == NULL ? "!Error!" : serEntry->serType;
const char *scope = pubsub_zmqTopicReceiver_scope(receiver);
const char *topic = pubsub_zmqTopicReceiver_topic(receiver);
+
+ celix_array_list_t *connected = celix_arrayList_create();
+ celix_array_list_t *unconnected = celix_arrayList_create();
+ pubsub_zmqTopicReceiver_listConnections(receiver, connected, unconnected);
+
fprintf(out, "|- Topic Receiver %s/%s\n", scope, topic);
fprintf(out, " |- serializer type = %s\n", serType);
+ for (int i = 0; i < celix_arrayList_size(connected); ++i) {
+ char *url = celix_arrayList_get(connected, i);
+ fprintf(out, " |- connected url = %s\n", url);
+ free(url);
+ }
+ for (int i = 0; i < celix_arrayList_size(unconnected); ++i) {
+ char *url = celix_arrayList_get(unconnected, i);
+ fprintf(out, " |- unconnected url = %s\n", url);
+ free(url);
+ }
+ celix_arrayList_destroy(connected);
+ celix_arrayList_destroy(unconnected);
}
celixThreadMutex_unlock(&psa->topicReceivers.mutex);
celixThreadMutex_unlock(&psa->serializers.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) {
+static celix_status_t zmq_getIpAddress(const char* interface, char** ip) {
celix_status_t status = CELIX_BUNDLE_EXCEPTION;
struct ifaddrs *ifaddr, *ifa;
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/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
index f0c072d..7a07827 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.c
@@ -208,6 +208,9 @@ pubsub_zmq_topic_receiver_t* pubsub_zmqTopicReceiver_create(celix_bundle_context
if (receiver->zmqSocket != NULL ) {
receiver->recvThread.running = true;
celixThread_create(&receiver->recvThread.thread, NULL, psa_zmq_recvThread, receiver);
+ char name[64];
+ snprintf(name, 64, "ZMQ TR %s/%s", scope, topic);
+ celixThread_setName(&receiver->recvThread.thread, name);
}
if (receiver->zmqSocket == NULL) {
@@ -258,6 +261,9 @@ void pubsub_zmqTopicReceiver_destroy(pubsub_zmq_topic_receiver_t *receiver) {
celixThreadMutex_destroy(&receiver->recvThread.mutex);
zsock_destroy(&receiver->zmqSocket);
+
+ free(receiver->scope);
+ free(receiver->topic);
}
free(receiver);
}
@@ -273,6 +279,20 @@ long pubsub_zmqTopicReceiver_serializerSvcId(pubsub_zmq_topic_receiver_t *receiv
return receiver->serializerSvcId;
}
+void pubsub_zmqTopicReceiver_listConnections(pubsub_zmq_topic_receiver_t *receiver, celix_array_list_t *connectedUrls, celix_array_list_t *unconnectedUrls) {
+ celixThreadMutex_lock(&receiver->requestedConnections.mutex);
+ hash_map_iterator_t iter = hashMapIterator_construct(receiver->requestedConnections.map);
+ while (hashMapIterator_hasNext(&iter)) {
+ psa_zmq_requested_connection_entry_t *entry = hashMapIterator_nextValue(&iter);
+ if (entry->connected) {
+ celix_arrayList_add(connectedUrls, strndup(entry->url, 1024));
+ } else {
+ celix_arrayList_add(unconnectedUrls, strndup(entry->url, 1024));
+ }
+ }
+ celixThreadMutex_unlock(&receiver->requestedConnections.mutex);
+}
+
void pubsub_zmqTopicReceiver_connectTo(
pubsub_zmq_topic_receiver_t *receiver,
@@ -446,6 +466,13 @@ static void* psa_zmq_recvThread(void * data) {
}
psa_zmq_processMsg(receiver, msg_list);
+ for (int i = 0; i < celix_arrayList_size(msg_list); ++i) {
+ complete_zmq_msg_t *msg = celix_arrayList_get(msg_list, i);
+ zframe_destroy(&msg->header);
+ zframe_destroy(&msg->payload);
+ free(msg);
+ }
+ celix_arrayList_destroy(msg_list);
}
} else {
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/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
index ea931a1..9049153 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_receiver.h
@@ -35,6 +35,7 @@ const char* pubsub_zmqTopicReceiver_scope(pubsub_zmq_topic_receiver_t *receiver)
const char* pubsub_zmqTopicReceiver_topic(pubsub_zmq_topic_receiver_t *receiver);
long pubsub_zmqTopicReceiver_serializerSvcId(pubsub_zmq_topic_receiver_t *receiver);
+void pubsub_zmqTopicReceiver_listConnections(pubsub_zmq_topic_receiver_t *receiver, celix_array_list_t *connectedUrls, celix_array_list_t *unconnectedUrls);
void pubsub_zmqTopicReceiver_connectTo(pubsub_zmq_topic_receiver_t *receiver, const char *url);
void pubsub_zmqTopicReceiver_disconnectFrom(pubsub_zmq_topic_receiver_t *receiver, const char *url);
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_sender.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_sender.c b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_sender.c
index 9d44c94..68c0e93 100644
--- a/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_sender.c
+++ b/bundles/pubsub/pubsub_admin_zmq/src/pubsub_zmq_topic_sender.c
@@ -245,6 +245,8 @@ void pubsub_zmqTopicSender_destroy(pubsub_zmq_topic_sender_t *sender) {
if (sender != NULL) {
celix_bundleContext_unregisterService(sender->ctx, sender->publisher.svcId);
+ zsock_destroy(&sender->zmq.socket);
+
celixThreadMutex_lock(&sender->boundedServices.mutex);
hash_map_iterator_t iter = hashMapIterator_construct(sender->boundedServices.map);
while (hashMapIterator_hasNext(&iter)) {
@@ -290,7 +292,6 @@ const char* pubsub_zmqTopicSender_url(pubsub_zmq_topic_sender_t *sender) {
return sender->url;
}
-
void pubsub_zmqTopicSender_connectTo(pubsub_zmq_topic_sender_t *sender, const celix_properties_t *endpoint) {
//TODO subscriber count -> topic info
}
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/bundles/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c b/bundles/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c
index 116dd50..b6ba67a 100644
--- a/bundles/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c
+++ b/bundles/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c
@@ -93,6 +93,7 @@ pubsub_discovery_t* pubsub_discovery_create(bundle_context_pt context, log_helpe
celix_status_t pubsub_discovery_destroy(pubsub_discovery_t *ps_discovery) {
celix_status_t status = CELIX_SUCCESS;
+ //note cleanup done in stop
celixThreadMutex_lock(&ps_discovery->discoveredEndpointsMutex);
hashMap_destroy(ps_discovery->discoveredEndpoints, false, false);
celixThreadMutex_unlock(&ps_discovery->discoveredEndpointsMutex);
@@ -103,6 +104,7 @@ celix_status_t pubsub_discovery_destroy(pubsub_discovery_t *ps_discovery) {
celixThreadMutex_unlock(&ps_discovery->discoveredEndpointsListenersMutex);
celixThreadMutex_destroy(&ps_discovery->discoveredEndpointsListenersMutex);
+ //note cleanup done in stop
celixThreadMutex_lock(&ps_discovery->announcedEndpointsMutex);
hashMap_destroy(ps_discovery->announcedEndpoints, false, false);
celixThreadMutex_unlock(&ps_discovery->announcedEndpointsMutex);
@@ -278,6 +280,7 @@ void* psd_refresh(void *data) {
L_ERROR("[PSD] Warning: error setting endpoint in etcd for key %s\n", entry->key);
entry->errorCount += 1;
}
+ free(str);
}
}
celixThreadMutex_unlock(&disc->announcedEndpointsMutex);
@@ -451,9 +454,8 @@ static void pubsub_discovery_addDiscoveredEndpoint(pubsub_discovery_t *disc, cel
assert(fwUUID != NULL);
if (fwUUID != NULL && strncmp(disc->fwUUID, fwUUID, strlen(disc->fwUUID)) == 0) {
-// if (disc->verbose) {
-// printf("[PSD] Ignoring endpoint %s from own framework\n", uuid);
-// }
+ //own endpoint -> ignore
+ celix_properties_destroy(endpoint);
return;
}
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/bundles/pubsub/pubsub_topology_manager/src/pstm_activator.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_topology_manager/src/pstm_activator.c b/bundles/pubsub/pubsub_topology_manager/src/pstm_activator.c
index b34b18e..11a1cc7 100644
--- a/bundles/pubsub/pubsub_topology_manager/src/pstm_activator.c
+++ b/bundles/pubsub/pubsub_topology_manager/src/pstm_activator.c
@@ -129,6 +129,10 @@ static int pstm_start(pstm_activator_t *act, celix_bundle_context_t *ctx) {
act->shellCmdSvcId = celix_bundleContext_registerService(ctx, &act->shellCmdSvc, OSGI_SHELL_COMMAND_SERVICE_NAME, props);
}
+ //TODO add tracker for pubsub_serializer and
+ //1) on remove reset sender/receivers entries
+ //2) on add indicate that topic/senders should be reevaluated.
+
/* NOTE: Enable those line in order to remotely expose the topic_info service
properties_pt props = properties_create();
properties_set(props, (char *) OSGI_RSA_SERVICE_EXPORTED_INTERFACES, (char *) PUBSUB_TOPIC_INFO_SERVICE);
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
----------------------------------------------------------------------
diff --git a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
index a8e31ef..84d0a0d 100644
--- a/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
+++ b/bundles/pubsub/pubsub_topology_manager/src/pubsub_topology_manager.c
@@ -150,6 +150,9 @@ celix_status_t pubsub_topologyManager_destroy(pubsub_topology_manager_t *manager
celixThreadMutex_unlock(&manager->topicSenders.mutex);
celixThreadMutex_destroy(&manager->topicSenders.mutex);
+ celixThreadMutex_destroy(&manager->announceEndpointListeners.mutex);
+ celix_arrayList_destroy(manager->announceEndpointListeners.list);
+
free(manager);
return status;
@@ -306,7 +309,7 @@ void pubsub_topologyManager_subscriberRemoved(void * handle, void *svc __attribu
char *scopeAndTopicKey = pubsubEndpoint_createScopeTopicKey(scope, topic);
celixThreadMutex_lock(&manager->topicReceivers.mutex);
- pstm_topic_receiver_or_sender_entry_t *entry = hashMap_remove(manager->topicReceivers.map, scopeAndTopicKey);
+ pstm_topic_receiver_or_sender_entry_t *entry = hashMap_get(manager->topicReceivers.map, scopeAndTopicKey);
if (entry != NULL) {
entry->usageCount -= 0;
}
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/libs/etcdlib/src/etcd.c
----------------------------------------------------------------------
diff --git a/libs/etcdlib/src/etcd.c b/libs/etcdlib/src/etcd.c
index 1f52dcb..7db128e 100644
--- a/libs/etcdlib/src/etcd.c
+++ b/libs/etcdlib/src/etcd.c
@@ -373,6 +373,7 @@ int etcd_refresh(const char* key, int ttl) {
} else {
retVal = ETCDLIB_RC_ERROR;
}
+ json_decref(root);
} else {
retVal = ETCDLIB_RC_ERROR;
fprintf(stderr, "[ETCDLIB] Error: %s is not json", reply.memory);
http://git-wip-us.apache.org/repos/asf/celix/blob/dac11022/libs/utils/src/celix_threads.c
----------------------------------------------------------------------
diff --git a/libs/utils/src/celix_threads.c b/libs/utils/src/celix_threads.c
index 7bfc37d..a6499a8 100644
--- a/libs/utils/src/celix_threads.c
+++ b/libs/utils/src/celix_threads.c
@@ -25,6 +25,7 @@
*/
#include <stdlib.h>
#include <sys/time.h>
+#include <time.h>
#include "signal.h"
#include "celix_threads.h"
@@ -42,7 +43,7 @@ celix_status_t celixThread_create(celix_thread_t *new_thread, celix_thread_attr_
return status;
}
-#ifdef _GNU_SOURCE
+#if defined(_GNU_SOURCE) && defined(__linux__)
void celixThread_setName(celix_thread_t *thread, const char *threadName) {
pthread_setname_np(thread->thread, threadName);
}