celix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pnol...@apache.org
Subject [38/54] [abbrv] celix git commit: CELIX-417: Refactors cmake usage of pubsub and rsa. Started with installing exported targets
Date Tue, 30 Jan 2018 19:30:22 GMT
http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_admin_zmq/src/topic_subscription.c
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_admin_zmq/src/topic_subscription.c b/pubsub/pubsub_admin_zmq/src/topic_subscription.c
new file mode 100644
index 0000000..0e7a794
--- /dev/null
+++ b/pubsub/pubsub_admin_zmq/src/topic_subscription.c
@@ -0,0 +1,732 @@
+/**
+ *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.
+ */
+/*
+ * topic_subscription.c
+ *
+ *  \date       Oct 2, 2015
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#include "topic_subscription.h"
+#include <czmq.h>
+/* The following undefs prevent the collision between:
+ * - sys/syslog.h (which is included within czmq)
+ * - celix/dfi/dfi_log_util.h
+ */
+#undef LOG_DEBUG
+#undef LOG_WARNING
+#undef LOG_INFO
+#undef LOG_WARNING
+
+#include <string.h>
+#include <stdlib.h>
+#include <signal.h>
+
+#include "utils.h"
+#include "celix_errno.h"
+#include "constants.h"
+#include "version.h"
+
+#include "subscriber.h"
+#include "publisher.h"
+#include "pubsub_utils.h"
+
+#ifdef BUILD_WITH_ZMQ_SECURITY
+#include "zmq_crypto.h"
+
+#define MAX_CERT_PATH_LENGTH 512
+#endif
+
+#define POLL_TIMEOUT  	250
+#define ZMQ_POLL_TIMEOUT_MS_ENV 	"ZMQ_POLL_TIMEOUT_MS"
+
+struct topic_subscription{
+
+	zsock_t* zmq_socket;
+	zcert_t * zmq_cert;
+	zcert_t * zmq_pub_cert;
+	pthread_mutex_t socket_lock;
+	service_tracker_pt tracker;
+	array_list_pt sub_ep_list;
+	celix_thread_t recv_thread;
+	bool running;
+	celix_thread_mutex_t ts_lock;
+	bundle_context_pt context;
+
+	pubsub_serializer_service_t *serializer;
+
+	hash_map_pt servicesMap; // key = service, value = msg types map
+
+	celix_thread_mutex_t pendingConnections_lock;
+	array_list_pt pendingConnections;
+
+	array_list_pt pendingDisconnections;
+	celix_thread_mutex_t pendingDisconnections_lock;
+
+	unsigned int nrSubscribers;
+};
+
+typedef struct complete_zmq_msg{
+	zframe_t* header;
+	zframe_t* payload;
+}* complete_zmq_msg_pt;
+
+typedef struct mp_handle{
+	hash_map_pt svc_msg_db;
+	hash_map_pt rcv_msg_map;
+}* mp_handle_pt;
+
+typedef struct msg_map_entry{
+	bool retain;
+	void* msgInst;
+}* msg_map_entry_pt;
+
+static celix_status_t topicsub_subscriberTracked(void * handle, service_reference_pt reference, void * service);
+static celix_status_t topicsub_subscriberUntracked(void * handle, service_reference_pt reference, void * service);
+static void* zmq_recv_thread_func(void* arg);
+static bool checkVersion(version_pt msgVersion,pubsub_msg_header_pt hdr);
+static void sigusr1_sighandler(int signo);
+static int pubsub_localMsgTypeIdForMsgType(void* handle, const char* msgType, unsigned int* msgTypeId);
+static int pubsub_getMultipart(void *handle, unsigned int msgTypeId, bool retain, void **part);
+static mp_handle_pt create_mp_handle(hash_map_pt svc_msg_db,array_list_pt rcv_msg_list);
+static void destroy_mp_handle(mp_handle_pt mp_handle);
+static void connectPendingPublishers(topic_subscription_pt sub);
+static void disconnectPendingPublishers(topic_subscription_pt sub);
+
+celix_status_t pubsub_topicSubscriptionCreate(bundle_context_pt bundle_context, char* scope, char* topic, pubsub_serializer_service_t *best_serializer, topic_subscription_pt* out){
+	celix_status_t status = CELIX_SUCCESS;
+
+#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
+
+	zsock_t* zmq_s = zsock_new (ZMQ_SUB);
+	if(zmq_s==NULL){
+#ifdef BUILD_WITH_ZMQ_SECURITY
+		zcert_destroy(&sub_cert);
+		zcert_destroy(&pub_cert);
+#endif
+
+		return CELIX_SERVICE_EXCEPTION;
+	}
+
+#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
+
+	if(strcmp(topic,PUBSUB_ANY_SUB_TOPIC)==0){
+		zsock_set_subscribe (zmq_s, "");
+	}
+	else{
+		zsock_set_subscribe (zmq_s, topic);
+	}
+
+	topic_subscription_pt ts = (topic_subscription_pt) calloc(1,sizeof(*ts));
+	ts->context = bundle_context;
+	ts->zmq_socket = zmq_s;
+	ts->running = false;
+	ts->nrSubscribers = 0;
+	ts->serializer = best_serializer;
+
+#ifdef BUILD_WITH_ZMQ_SECURITY
+	ts->zmq_cert = sub_cert;
+	ts->zmq_pub_cert = pub_cert;
+#endif
+
+	celixThreadMutex_create(&ts->socket_lock, NULL);
+	celixThreadMutex_create(&ts->ts_lock,NULL);
+	arrayList_create(&ts->sub_ep_list);
+	ts->servicesMap = hashMap_create(NULL, NULL, NULL, NULL);
+
+	arrayList_create(&ts->pendingConnections);
+	arrayList_create(&ts->pendingDisconnections);
+	celixThreadMutex_create(&ts->pendingConnections_lock, NULL);
+	celixThreadMutex_create(&ts->pendingDisconnections_lock, NULL);
+
+	char filter[128];
+	memset(filter,0,128);
+	if(strncmp(PUBSUB_SUBSCRIBER_SCOPE_DEFAULT,scope,strlen(PUBSUB_SUBSCRIBER_SCOPE_DEFAULT)) == 0) {
+		// default scope, means that subscriber has not defined a scope property
+		snprintf(filter, 128, "(&(%s=%s)(%s=%s))",
+				(char*) OSGI_FRAMEWORK_OBJECTCLASS, PUBSUB_SUBSCRIBER_SERVICE_NAME,
+				PUBSUB_SUBSCRIBER_TOPIC,topic);
+
+	} else {
+		snprintf(filter, 128, "(&(%s=%s)(%s=%s)(%s=%s))",
+				(char*) OSGI_FRAMEWORK_OBJECTCLASS, PUBSUB_SUBSCRIBER_SERVICE_NAME,
+				PUBSUB_SUBSCRIBER_TOPIC,topic,
+				PUBSUB_SUBSCRIBER_SCOPE,scope);
+	}
+	service_tracker_customizer_pt customizer = NULL;
+	status += serviceTrackerCustomizer_create(ts,NULL,topicsub_subscriberTracked,NULL,topicsub_subscriberUntracked,&customizer);
+	status += serviceTracker_createWithFilter(bundle_context, filter, customizer, &ts->tracker);
+
+	struct sigaction actions;
+	memset(&actions, 0, sizeof(actions));
+	sigemptyset(&actions.sa_mask);
+	actions.sa_flags = 0;
+	actions.sa_handler = sigusr1_sighandler;
+
+	sigaction(SIGUSR1,&actions,NULL);
+
+	*out=ts;
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionDestroy(topic_subscription_pt ts){
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+
+	ts->running = false;
+	serviceTracker_destroy(ts->tracker);
+	arrayList_clear(ts->sub_ep_list);
+	arrayList_destroy(ts->sub_ep_list);
+	/* TODO: Destroy all the serializer maps? */
+	hashMap_destroy(ts->servicesMap,false,false);
+
+	celixThreadMutex_lock(&ts->pendingConnections_lock);
+	arrayList_destroy(ts->pendingConnections);
+	celixThreadMutex_unlock(&ts->pendingConnections_lock);
+	celixThreadMutex_destroy(&ts->pendingConnections_lock);
+
+	celixThreadMutex_lock(&ts->pendingDisconnections_lock);
+	arrayList_destroy(ts->pendingDisconnections);
+	celixThreadMutex_unlock(&ts->pendingDisconnections_lock);
+	celixThreadMutex_destroy(&ts->pendingDisconnections_lock);
+
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	celixThreadMutex_lock(&ts->socket_lock);
+	zsock_destroy(&(ts->zmq_socket));
+#ifdef BUILD_WITH_ZMQ_SECURITY
+	zcert_destroy(&(ts->zmq_cert));
+	zcert_destroy(&(ts->zmq_pub_cert));
+#endif
+	celixThreadMutex_unlock(&ts->socket_lock);
+	celixThreadMutex_destroy(&ts->socket_lock);
+
+	celixThreadMutex_destroy(&ts->ts_lock);
+
+	free(ts);
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionStart(topic_subscription_pt ts){
+	celix_status_t status = CELIX_SUCCESS;
+
+	status = serviceTracker_open(ts->tracker);
+
+	ts->running = true;
+
+	if(status==CELIX_SUCCESS){
+		status=celixThread_create(&ts->recv_thread,NULL,zmq_recv_thread_func,ts);
+	}
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionStop(topic_subscription_pt ts){
+	celix_status_t status = CELIX_SUCCESS;
+
+	ts->running = false;
+
+	pthread_kill(ts->recv_thread.thread,SIGUSR1);
+
+	celixThread_join(ts->recv_thread,NULL);
+
+	status = serviceTracker_close(ts->tracker);
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionConnectPublisher(topic_subscription_pt ts, char* pubURL){
+	celix_status_t status = CELIX_SUCCESS;
+	celixThreadMutex_lock(&ts->socket_lock);
+	if(!zsock_is(ts->zmq_socket) || zsock_connect(ts->zmq_socket,"%s",pubURL) != 0){
+		status = CELIX_SERVICE_EXCEPTION;
+	}
+	celixThreadMutex_unlock(&ts->socket_lock);
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionAddConnectPublisherToPendingList(topic_subscription_pt ts, char* pubURL) {
+	celix_status_t status = CELIX_SUCCESS;
+	char *url = strdup(pubURL);
+	celixThreadMutex_lock(&ts->pendingConnections_lock);
+	arrayList_add(ts->pendingConnections, url);
+	celixThreadMutex_unlock(&ts->pendingConnections_lock);
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionAddDisconnectPublisherToPendingList(topic_subscription_pt ts, char* pubURL) {
+	celix_status_t status = CELIX_SUCCESS;
+	char *url = strdup(pubURL);
+	celixThreadMutex_lock(&ts->pendingDisconnections_lock);
+	arrayList_add(ts->pendingDisconnections, url);
+	celixThreadMutex_unlock(&ts->pendingDisconnections_lock);
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionDisconnectPublisher(topic_subscription_pt ts, char* pubURL){
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->socket_lock);
+	if(!zsock_is(ts->zmq_socket) || zsock_disconnect(ts->zmq_socket,"%s",pubURL) != 0){
+		status = CELIX_SERVICE_EXCEPTION;
+	}
+	celixThreadMutex_unlock(&ts->socket_lock);
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionAddSubscriber(topic_subscription_pt ts, pubsub_endpoint_pt subEP){
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	arrayList_add(ts->sub_ep_list,subEP);
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+
+}
+
+celix_status_t pubsub_topicIncreaseNrSubscribers(topic_subscription_pt ts) {
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	ts->nrSubscribers++;
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+}
+
+celix_status_t pubsub_topicSubscriptionRemoveSubscriber(topic_subscription_pt ts, pubsub_endpoint_pt subEP){
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	arrayList_removeElement(ts->sub_ep_list,subEP);
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+}
+
+celix_status_t pubsub_topicDecreaseNrSubscribers(topic_subscription_pt ts) {
+	celix_status_t status = CELIX_SUCCESS;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	ts->nrSubscribers--;
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+}
+
+unsigned int pubsub_topicGetNrSubscribers(topic_subscription_pt ts) {
+	return ts->nrSubscribers;
+}
+
+array_list_pt pubsub_topicSubscriptionGetSubscribersList(topic_subscription_pt sub){
+	return sub->sub_ep_list;
+}
+
+static celix_status_t topicsub_subscriberTracked(void * handle, service_reference_pt reference, void * service){
+	celix_status_t status = CELIX_SUCCESS;
+	topic_subscription_pt ts = handle;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	if (!hashMap_containsKey(ts->servicesMap, service)) {
+		bundle_pt bundle = NULL;
+		hash_map_pt msgTypes = NULL;
+
+		serviceReference_getBundle(reference, &bundle);
+
+		if(ts->serializer != NULL && bundle!=NULL){
+			ts->serializer->createSerializerMap(ts->serializer->handle,bundle,&msgTypes);
+			if(msgTypes != NULL){
+				hashMap_put(ts->servicesMap, service, msgTypes);
+				printf("PSA_ZMQ_TS: New subscriber registered.\n");
+			}
+		}
+		else{
+			printf("PSA_ZMQ_TS: Cannot register new subscriber.\n");
+			status = CELIX_SERVICE_EXCEPTION;
+		}
+	}
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+}
+
+static celix_status_t topicsub_subscriberUntracked(void * handle, service_reference_pt reference, void * service){
+	celix_status_t status = CELIX_SUCCESS;
+	topic_subscription_pt ts = handle;
+
+	celixThreadMutex_lock(&ts->ts_lock);
+	if (hashMap_containsKey(ts->servicesMap, service)) {
+		hash_map_pt msgTypes = hashMap_remove(ts->servicesMap, service);
+		if(msgTypes!=NULL && ts->serializer!=NULL){
+			ts->serializer->destroySerializerMap(ts->serializer->handle,msgTypes);
+			printf("PSA_ZMQ_TS: Subscriber unregistered.\n");
+		}
+		else{
+			printf("PSA_ZMQ_TS: Cannot unregister subscriber.\n");
+			status = CELIX_SERVICE_EXCEPTION;
+		}
+	}
+	celixThreadMutex_unlock(&ts->ts_lock);
+
+	return status;
+}
+
+
+static void process_msg(topic_subscription_pt sub,array_list_pt msg_list){
+
+	pubsub_msg_header_pt first_msg_hdr = (pubsub_msg_header_pt)zframe_data(((complete_zmq_msg_pt)arrayList_get(msg_list,0))->header);
+
+	hash_map_iterator_pt iter = hashMapIterator_create(sub->servicesMap);
+	while (hashMapIterator_hasNext(iter)) {
+		hash_map_entry_pt entry = hashMapIterator_nextEntry(iter);
+		pubsub_subscriber_pt subsvc = hashMapEntry_getKey(entry);
+		hash_map_pt msgTypes = hashMapEntry_getValue(entry);
+
+		pubsub_msg_serializer_t *msgSer = hashMap_get(msgTypes,(void*)(uintptr_t )first_msg_hdr->type);
+		if (msgSer == NULL) {
+			printf("PSA_ZMQ_TS: Primary message %d not supported. NOT sending any part of the whole message.\n",first_msg_hdr->type);
+		}
+		else{
+			void *msgInst = NULL;
+			bool validVersion = checkVersion(msgSer->msgVersion,first_msg_hdr);
+
+			if(validVersion){
+
+				celix_status_t status = msgSer->deserialize(msgSer, (const void *) zframe_data(((complete_zmq_msg_pt)arrayList_get(msg_list,0))->payload), 0, &msgInst);
+
+				if (status == CELIX_SUCCESS) {
+					bool release = true;
+					mp_handle_pt mp_handle = create_mp_handle(msgTypes,msg_list);
+					pubsub_multipart_callbacks_t mp_callbacks;
+					mp_callbacks.handle = mp_handle;
+					mp_callbacks.localMsgTypeIdForMsgType = pubsub_localMsgTypeIdForMsgType;
+					mp_callbacks.getMultipart = pubsub_getMultipart;
+					subsvc->receive(subsvc->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){
+						destroy_mp_handle(mp_handle);
+					}
+				}
+				else{
+					printf("PSA_ZMQ_TS: Cannot deserialize msgType %s.\n",msgSer->msgName);
+				}
+
+			}
+			else{
+				int major=0,minor=0;
+				version_getMajor(msgSer->msgVersion,&major);
+				version_getMinor(msgSer->msgVersion,&minor);
+				printf("PSA_ZMQ_TS: 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);
+			}
+
+		}
+	}
+	hashMapIterator_destroy(iter);
+
+	int i = 0;
+	for(;i<arrayList_size(msg_list);i++){
+		complete_zmq_msg_pt c_msg = arrayList_get(msg_list,i);
+		zframe_destroy(&(c_msg->header));
+		zframe_destroy(&(c_msg->payload));
+		free(c_msg);
+	}
+
+	arrayList_destroy(msg_list);
+
+}
+
+static void* zmq_recv_thread_func(void * arg) {
+	topic_subscription_pt sub = (topic_subscription_pt) arg;
+
+	while (sub->running) {
+
+		celixThreadMutex_lock(&sub->socket_lock);
+
+		zframe_t* headerMsg = zframe_recv(sub->zmq_socket);
+		if (headerMsg == NULL) {
+			if (errno == EINTR) {
+				//It means we got a signal and we have to exit...
+				printf("PSA_ZMQ_TS: header_recv thread for topic got a signal and will exit.\n");
+			} else {
+				perror("PSA_ZMQ_TS: header_recv thread");
+			}
+		}
+		else {
+
+			pubsub_msg_header_pt hdr = (pubsub_msg_header_pt) zframe_data(headerMsg);
+
+			if (zframe_more(headerMsg)) {
+
+				zframe_t* payloadMsg = zframe_recv(sub->zmq_socket);
+				if (payloadMsg == NULL) {
+					if (errno == EINTR) {
+						//It means we got a signal and we have to exit...
+						printf("PSA_ZMQ_TS: payload_recv thread for topic got a signal and will exit.\n");
+					} else {
+						perror("PSA_ZMQ_TS: payload_recv");
+					}
+					zframe_destroy(&headerMsg);
+				} else {
+
+					//Let's fetch all the messages from the socket
+					array_list_pt msg_list = NULL;
+					arrayList_create(&msg_list);
+					complete_zmq_msg_pt firstMsg = calloc(1, sizeof(struct complete_zmq_msg));
+					firstMsg->header = headerMsg;
+					firstMsg->payload = payloadMsg;
+					arrayList_add(msg_list, firstMsg);
+
+					bool more = zframe_more(payloadMsg);
+					while (more) {
+
+						zframe_t* h_msg = zframe_recv(sub->zmq_socket);
+						if (h_msg == NULL) {
+							if (errno == EINTR) {
+								//It means we got a signal and we have to exit...
+								printf("PSA_ZMQ_TS: h_recv thread for topic got a signal and will exit.\n");
+							} else {
+								perror("PSA_ZMQ_TS: h_recv");
+							}
+							break;
+						}
+
+						zframe_t* p_msg = zframe_recv(sub->zmq_socket);
+						if (p_msg == NULL) {
+							if (errno == EINTR) {
+								//It means we got a signal and we have to exit...
+								printf("PSA_ZMQ_TS: p_recv thread for topic got a signal and will exit.\n");
+							} else {
+								perror("PSA_ZMQ_TS: p_recv");
+							}
+							zframe_destroy(&h_msg);
+							break;
+						}
+
+						complete_zmq_msg_pt 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;
+						}
+					}
+
+					celixThreadMutex_lock(&sub->ts_lock);
+					process_msg(sub, msg_list);
+					celixThreadMutex_unlock(&sub->ts_lock);
+
+				}
+
+			} //zframe_more(headerMsg)
+			else {
+				free(headerMsg);
+				printf("PSA_ZMQ_TS: received message %u for topic %s without payload!\n", hdr->type, hdr->topic);
+			}
+
+		} // headerMsg != NULL
+		celixThreadMutex_unlock(&sub->socket_lock);
+		connectPendingPublishers(sub);
+		disconnectPendingPublishers(sub);
+	} // while
+
+	return NULL;
+}
+
+static void connectPendingPublishers(topic_subscription_pt sub) {
+	celixThreadMutex_lock(&sub->pendingConnections_lock);
+	while(!arrayList_isEmpty(sub->pendingConnections)) {
+		char * pubEP = arrayList_remove(sub->pendingConnections, 0);
+		pubsub_topicSubscriptionConnectPublisher(sub, pubEP);
+		free(pubEP);
+	}
+	celixThreadMutex_unlock(&sub->pendingConnections_lock);
+}
+
+static void disconnectPendingPublishers(topic_subscription_pt sub) {
+	celixThreadMutex_lock(&sub->pendingDisconnections_lock);
+	while(!arrayList_isEmpty(sub->pendingDisconnections)) {
+		char * pubEP = arrayList_remove(sub->pendingDisconnections, 0);
+		pubsub_topicSubscriptionDisconnectPublisher(sub, pubEP);
+		free(pubEP);
+	}
+	celixThreadMutex_unlock(&sub->pendingDisconnections_lock);
+}
+
+static void sigusr1_sighandler(int signo){
+	printf("PSA_ZMQ_TS: Topic subscription being shut down...\n");
+	return;
+}
+
+static bool checkVersion(version_pt msgVersion,pubsub_msg_header_pt 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;
+}
+
+static int pubsub_localMsgTypeIdForMsgType(void* handle, const char* msgType, unsigned int* msgTypeId){
+	*msgTypeId = utils_stringHash(msgType);
+	return 0;
+}
+
+static int pubsub_getMultipart(void *handle, unsigned int msgTypeId, bool retain, void **part){
+
+	if(handle==NULL){
+		*part = NULL;
+		return -1;
+	}
+
+	mp_handle_pt mp_handle = (mp_handle_pt)handle;
+	msg_map_entry_pt 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 mp_handle_pt create_mp_handle(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_pt 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_pt c_msg = (complete_zmq_msg_pt)arrayList_get(rcv_msg_list,i);
+		pubsub_msg_header_pt header = (pubsub_msg_header_pt)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 = 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_pt 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 destroy_mp_handle(mp_handle_pt 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_pt 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);
+}

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_admin_zmq/src/topic_subscription.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_admin_zmq/src/topic_subscription.h b/pubsub/pubsub_admin_zmq/src/topic_subscription.h
new file mode 100644
index 0000000..7267103
--- /dev/null
+++ b/pubsub/pubsub_admin_zmq/src/topic_subscription.h
@@ -0,0 +1,60 @@
+/**
+ *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.
+ */
+/*
+ * topic_subscription.h
+ *
+ *  \date       Sep 22, 2015
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#ifndef TOPIC_SUBSCRIPTION_H_
+#define TOPIC_SUBSCRIPTION_H_
+
+#include "celix_threads.h"
+#include "array_list.h"
+#include "celixbool.h"
+#include "service_tracker.h"
+
+#include "pubsub_endpoint.h"
+#include "pubsub_common.h"
+#include "pubsub_serializer.h"
+
+typedef struct topic_subscription* topic_subscription_pt;
+
+celix_status_t pubsub_topicSubscriptionCreate(bundle_context_pt bundle_context,char* scope, char* topic, pubsub_serializer_service_t *best_serializer, topic_subscription_pt* out);
+celix_status_t pubsub_topicSubscriptionDestroy(topic_subscription_pt ts);
+celix_status_t pubsub_topicSubscriptionStart(topic_subscription_pt ts);
+celix_status_t pubsub_topicSubscriptionStop(topic_subscription_pt ts);
+
+celix_status_t pubsub_topicSubscriptionAddConnectPublisherToPendingList(topic_subscription_pt ts, char* pubURL);
+celix_status_t pubsub_topicSubscriptionAddDisconnectPublisherToPendingList(topic_subscription_pt ts, char* pubURL);
+
+celix_status_t pubsub_topicSubscriptionConnectPublisher(topic_subscription_pt ts, char* pubURL);
+celix_status_t pubsub_topicSubscriptionDisconnectPublisher(topic_subscription_pt ts, char* pubURL);
+
+celix_status_t pubsub_topicSubscriptionAddSubscriber(topic_subscription_pt ts, pubsub_endpoint_pt subEP);
+celix_status_t pubsub_topicSubscriptionRemoveSubscriber(topic_subscription_pt ts, pubsub_endpoint_pt subEP);
+
+array_list_pt pubsub_topicSubscriptionGetSubscribersList(topic_subscription_pt sub);
+celix_status_t pubsub_topicIncreaseNrSubscribers(topic_subscription_pt subscription);
+celix_status_t pubsub_topicDecreaseNrSubscribers(topic_subscription_pt subscription);
+unsigned int pubsub_topicGetNrSubscribers(topic_subscription_pt subscription);
+
+#endif /*TOPIC_SUBSCRIPTION_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_admin_zmq/src/zmq_crypto.c
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_admin_zmq/src/zmq_crypto.c b/pubsub/pubsub_admin_zmq/src/zmq_crypto.c
new file mode 100644
index 0000000..fe444bd
--- /dev/null
+++ b/pubsub/pubsub_admin_zmq/src/zmq_crypto.c
@@ -0,0 +1,281 @@
+/**
+ *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.
+ */
+/*
+ * zmq_crypto.c
+ *
+ *  \date       Dec 2, 2016
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#include "zmq_crypto.h"
+
+#include <zmq.h>
+#include <openssl/conf.h>
+#include <openssl/evp.h>
+#include <openssl/err.h>
+
+#include <string.h>
+
+#define MAX_FILE_PATH_LENGTH 512
+#define ZMQ_KEY_LENGTH 40
+#define AES_KEY_LENGTH 32
+#define AES_IV_LENGTH 16
+
+#define KEY_TO_GET "aes_key"
+#define IV_TO_GET "aes_iv"
+
+static char* read_file_content(char* filePath, char* fileName);
+static void parse_key_lines(char *keysBuffer, char **key, char **iv);
+static void parse_key_line(char *line, char **key, char **iv);
+static void extract_keys_from_buffer(unsigned char *input, int inputlen, char **publicKey, char **secretKey);
+
+/**
+ * Return a valid zcert_t from an encoded file
+ * Caller is responsible for freeing by calling zcert_destroy(zcert** cert);
+ */
+zcert_t* get_zcert_from_encoded_file(char* keysFilePath, char* keysFileName, char* file_path)
+{
+
+	if (keysFilePath == NULL){
+		keysFilePath = DEFAULT_KEYS_FILE_PATH;
+	}
+
+	if (keysFileName == NULL){
+		keysFileName = DEFAULT_KEYS_FILE_NAME;
+	}
+
+	char* keys_data = read_file_content(keysFilePath, keysFileName);
+	if (keys_data == NULL){
+		return NULL;
+	}
+
+	char *key = NULL;
+	char *iv = NULL;
+	parse_key_lines(keys_data, &key, &iv);
+	free(keys_data);
+
+	if (key == NULL || iv == NULL){
+		free(key);
+		free(iv);
+
+		printf("CRYPTO: Loading AES key and/or AES iv failed!\n");
+		return NULL;
+	}
+
+	//At this point, we know an aes key and iv are stored and loaded
+
+	// generate sha256 hashes
+	unsigned char key_digest[EVP_MAX_MD_SIZE];
+	unsigned char iv_digest[EVP_MAX_MD_SIZE];
+	generate_sha256_hash((char*) key, key_digest);
+	generate_sha256_hash((char*) iv, iv_digest);
+
+	zchunk_t* encoded_secret = zchunk_slurp (file_path, 0);
+	if (encoded_secret == NULL){
+		free(key);
+		free(iv);
+
+		return NULL;
+	}
+
+	int encoded_secret_size = (int) zchunk_size (encoded_secret);
+	char* encoded_secret_data = zchunk_strdup(encoded_secret);
+	zchunk_destroy (&encoded_secret);
+
+	// Decryption of data
+	int decryptedtext_len;
+	unsigned char decryptedtext[encoded_secret_size];
+	decryptedtext_len = decrypt((unsigned char *) encoded_secret_data, encoded_secret_size, key_digest, iv_digest, decryptedtext);
+	decryptedtext[decryptedtext_len] = '\0';
+
+	EVP_cleanup();
+
+	free(encoded_secret_data);
+	free(key);
+	free(iv);
+
+	// The public and private keys are retrieved
+	char* public_text = NULL;
+	char* secret_text = NULL;
+
+	extract_keys_from_buffer(decryptedtext, decryptedtext_len, &public_text, &secret_text);
+
+	byte public_key [32] = { 0 };
+	byte secret_key [32] = { 0 };
+
+	zmq_z85_decode (public_key, public_text);
+	zmq_z85_decode (secret_key, secret_text);
+
+	zcert_t* cert_loaded = zcert_new_from(public_key, secret_key);
+
+	free(public_text);
+	free(secret_text);
+
+	return cert_loaded;
+}
+
+int generate_sha256_hash(char* text, unsigned char* digest)
+{
+	unsigned int digest_len;
+
+	EVP_MD_CTX * mdctx = EVP_MD_CTX_new();
+	EVP_DigestInit_ex(mdctx, EVP_sha256(), NULL);
+	EVP_DigestUpdate(mdctx, text, strlen(text));
+	EVP_DigestFinal_ex(mdctx, digest, &digest_len);
+	EVP_MD_CTX_free(mdctx);
+
+	return digest_len;
+}
+
+int decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char *key, unsigned char *iv, unsigned char *plaintext)
+{
+	int len;
+	int plaintext_len;
+
+	EVP_CIPHER_CTX* ctx = EVP_CIPHER_CTX_new();
+
+	EVP_DecryptInit_ex(ctx, EVP_aes_256_cbc(), NULL, key, iv);
+	EVP_DecryptUpdate(ctx, plaintext, &len, ciphertext, ciphertext_len);
+	plaintext_len = len;
+	EVP_DecryptFinal_ex(ctx, plaintext + len, &len);
+	plaintext_len += len;
+
+	EVP_CIPHER_CTX_free(ctx);
+
+	return plaintext_len;
+}
+
+/**
+ * Caller is responsible for freeing the returned value
+ */
+static char* read_file_content(char* filePath, char* fileName){
+
+	char fileNameWithPath[MAX_FILE_PATH_LENGTH];
+	snprintf(fileNameWithPath, MAX_FILE_PATH_LENGTH, "%s/%s", filePath, fileName);
+	int rc = 0;
+
+	if (!zsys_file_exists(fileNameWithPath)){
+		printf("CRYPTO: Keys file '%s' doesn't exist!\n", fileNameWithPath);
+		return NULL;
+	}
+
+	zfile_t* keys_file = zfile_new (filePath, fileName);
+	rc = zfile_input (keys_file);
+	if (rc != 0){
+		zfile_destroy(&keys_file);
+		printf("CRYPTO: Keys file '%s' not readable!\n", fileNameWithPath);
+		return NULL;
+	}
+
+	ssize_t keys_file_size = zsys_file_size (fileNameWithPath);
+	zchunk_t* keys_chunk = zfile_read (keys_file, keys_file_size, 0);
+	if (keys_chunk == NULL){
+		zfile_close(keys_file);
+		zfile_destroy(&keys_file);
+		printf("CRYPTO: Can't read file '%s'!\n", fileNameWithPath);
+		return NULL;
+	}
+
+	char* keys_data = zchunk_strdup(keys_chunk);
+	zchunk_destroy(&keys_chunk);
+	zfile_close(keys_file);
+	zfile_destroy (&keys_file);
+
+	return keys_data;
+}
+
+static void parse_key_lines(char *keysBuffer, char **key, char **iv){
+	char *line = NULL, *saveLinePointer = NULL;
+
+	bool firstTime = true;
+	do {
+		if (firstTime){
+			line = strtok_r(keysBuffer, "\n", &saveLinePointer);
+			firstTime = false;
+		}else {
+			line = strtok_r(NULL, "\n", &saveLinePointer);
+		}
+
+		if (line == NULL){
+			break;
+		}
+
+		parse_key_line(line, key, iv);
+
+	} while((*key == NULL || *iv == NULL) && line != NULL);
+
+}
+
+static void parse_key_line(char *line, char **key, char **iv){
+	char *detectedKey = NULL, *detectedValue= NULL;
+
+	char* sep_at = strchr(line, ':');
+	if (sep_at == NULL){
+		return;
+	}
+
+	*sep_at = '\0'; // overwrite first separator, creating two strings.
+	detectedKey = line;
+	detectedValue = sep_at + 1;
+
+	if (detectedKey == NULL || detectedValue == NULL){
+		return;
+	}
+	if (detectedKey[0] == '\0' || detectedValue[0] == '\0'){
+		return;
+	}
+
+	if (*key == NULL && strcmp(detectedKey, KEY_TO_GET) == 0){
+		*key = strndup(detectedValue, AES_KEY_LENGTH);
+	} else if (*iv == NULL && strcmp(detectedKey, IV_TO_GET) == 0){
+		*iv = strndup(detectedValue, AES_IV_LENGTH);
+	}
+}
+
+static void extract_keys_from_buffer(unsigned char *input, int inputlen, char **publicKey, char **secretKey) {
+	// Load decrypted text buffer
+	zchunk_t* secret_decrypted = zchunk_new(input, inputlen);
+	if (secret_decrypted == NULL){
+		printf("CRYPTO: Failed to create zchunk\n");
+		return;
+	}
+
+	zconfig_t* secret_config = zconfig_chunk_load (secret_decrypted);
+	zchunk_destroy (&secret_decrypted);
+	if (secret_config == NULL){
+		printf("CRYPTO: Failed to create zconfig\n");
+		return;
+	}
+
+	// Extract public and secret key from text buffer
+	char* public_text = zconfig_get (secret_config, "/curve/public-key", NULL);
+	char* secret_text = zconfig_get (secret_config, "/curve/secret-key", NULL);
+
+	if (public_text == NULL || secret_text == NULL){
+		zconfig_destroy(&secret_config);
+		printf("CRYPTO: Loading public / secret key from text-buffer failed!\n");
+		return;
+	}
+
+	*publicKey = strndup(public_text, ZMQ_KEY_LENGTH + 1);
+	*secretKey = strndup(secret_text, ZMQ_KEY_LENGTH + 1);
+
+	zconfig_destroy(&secret_config);
+}

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_admin_zmq/src/zmq_crypto.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_admin_zmq/src/zmq_crypto.h b/pubsub/pubsub_admin_zmq/src/zmq_crypto.h
new file mode 100644
index 0000000..f1a990f
--- /dev/null
+++ b/pubsub/pubsub_admin_zmq/src/zmq_crypto.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.
+ */
+/*
+ * zmq_crypto.h
+ *
+ *  \date       Dec 2, 2016
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#ifndef ZMQ_CRYPTO_H_
+#define ZMQ_CRYPTO_H_
+
+#include <czmq.h>
+
+#define PROPERTY_KEYS_FILE_PATH "keys.file.path"
+#define PROPERTY_KEYS_FILE_NAME "keys.file.name"
+#define DEFAULT_KEYS_FILE_PATH "/etc/"
+#define DEFAULT_KEYS_FILE_NAME "pubsub.keys"
+
+zcert_t* get_zcert_from_encoded_file(char* keysFilePath, char* keysFileName, char* file_path);
+int generate_sha256_hash(char* text, unsigned char* digest);
+int decrypt(unsigned char *ciphertext, int ciphertext_len, unsigned char *key, unsigned char *iv, unsigned char *plaintext);
+
+#endif

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_api/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_api/CMakeLists.txt b/pubsub/pubsub_api/CMakeLists.txt
new file mode 100644
index 0000000..5ceb291
--- /dev/null
+++ b/pubsub/pubsub_api/CMakeLists.txt
@@ -0,0 +1,32 @@
+# 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.
+
+#api target
+add_library(pubsub_api INTERFACE)
+
+target_include_directories(framework INTERFACE
+    $<BUILD_INTERFACE:${CMAKE_CURRENT_LIST_DIR}/include>
+    $<INSTALL_INTERFACE:include/celix/pubsub>
+)
+
+#install api
+install(TARGETS pubsub_api EXPORT celix DESTINATION ${CMAKE_INSTALL_LIBDIR} COMPONENT pubsub)
+install(DIRECTORY include/ DESTINATION include/celix/pubsub COMPONENT pubsub)
+
+#Setup target aliases to match external usage
+add_library(Celix::pubsub_api ALIAS pubsub_api)
+

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_api/include/pubsub/publisher.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_api/include/pubsub/publisher.h b/pubsub/pubsub_api/include/pubsub/publisher.h
new file mode 100644
index 0000000..3eec149
--- /dev/null
+++ b/pubsub/pubsub_api/include/pubsub/publisher.h
@@ -0,0 +1,88 @@
+/**
+ *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.
+ */
+/*
+ * publisher.h
+ *
+ *  \date       Jan 7, 2016
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#ifndef __PUBSUB_PUBLISHER_H_
+#define __PUBSUB_PUBLISHER_H_
+
+#include <stdlib.h>
+
+#define PUBSUB_PUBLISHER_SERVICE_NAME           "pubsub.publisher"
+#define PUBSUB_PUBLISHER_SERVICE_VERSION	      "2.0.0"
+ 
+//properties
+#define PUBSUB_PUBLISHER_TOPIC                  "pubsub.topic"
+#define PUBSUB_PUBLISHER_SCOPE                  "pubsub.scope"
+#define PUBSUB_PUBLISHER_STRATEGY               "pubsub.strategy"
+#define PUBSUB_PUBLISHER_CONFIG                 "pubsub.config"
+ 
+#define PUBSUB_PUBLISHER_SCOPE_DEFAULT			"default"
+//flags
+#define PUBSUB_PUBLISHER_FIRST_MSG  01
+#define PUBSUB_PUBLISHER_PART_MSG   02
+#define PUBSUB_PUBLISHER_LAST_MSG   04
+
+struct pubsub_release_callback_struct {
+    void *handle;
+    void (*release)(char *buf, void *handle);
+};
+typedef struct pubsub_release_callback_struct pubsub_release_callback_t;
+typedef struct pubsub_release_callback_struct* pubsub_release_callback_pt;
+ 
+ 
+struct pubsub_publisher {
+    void *handle;
+ 
+    /**
+     * Every msg is identifiable by msg type string. Because masg type string are performance wise not preferable (string compares),
+     * a "local" (int / platform dependent) unique id will be generated runtime
+     * with use of a distributed key/value store or communication between  participation parties.
+     * this is called the local message type id. This local message type id can be requested with the localMsgIdForMsgType method.
+     * When return is successful the msgTypeId is always greater than 0. (Note this can be used to specify/detect uninitialized msg type ids in the consumer code).
+     *
+     * Returns 0 on success.
+     */
+    int (*localMsgTypeIdForMsgType)(void *handle, const char *msgType, unsigned int *msgTypeId);
+  
+    /**
+     * send is a async function, but the msg can be safely deleted after send returns.
+     * Returns 0 on success.
+     */
+    int (*send)(void *handle, unsigned int msgTypeId, const void *msg);
+ 
+  
+    /**
+     * sendMultipart is a async function, but the msg can be safely deleted after send returns.
+     * The first (primary) message of a multipart message must have the flag PUBLISHER_PRIMARY_MSG
+     * The last message of a multipart message must have the flag PUBLISHER_LAST_MSG
+     * Returns 0 on success.
+     */
+    int (*sendMultipart)(void *handle, unsigned int msgTypeId, const void *msg, int flags);
+ 
+};
+typedef struct pubsub_publisher pubsub_publisher_t;
+typedef struct pubsub_publisher* pubsub_publisher_pt;
+
+#endif // __PUBSUB_PUBLISHER_H_

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_api/include/pubsub/subscriber.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_api/include/pubsub/subscriber.h b/pubsub/pubsub_api/include/pubsub/subscriber.h
new file mode 100644
index 0000000..5d87b8a
--- /dev/null
+++ b/pubsub/pubsub_api/include/pubsub/subscriber.h
@@ -0,0 +1,75 @@
+/**
+ *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.
+ */
+/*
+ * subscriber.h
+ *
+ *  \date       Jan 7, 2016
+ *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ *  \copyright	Apache License, Version 2.0
+ */
+
+#ifndef __PUBSUB_SUBSCRIBER_H_
+#define __PUBSUB_SUBSCRIBER_H_
+
+#include <stdbool.h>
+
+#define PUBSUB_SUBSCRIBER_SERVICE_NAME          "pubsub.subscriber"
+#define PUBSUB_SUBSCRIBER_SERVICE_VERSION       "2.0.0"
+ 
+//properties
+#define PUBSUB_SUBSCRIBER_TOPIC                "pubsub.topic"
+#define PUBSUB_SUBSCRIBER_SCOPE                "pubsub.scope"
+#define PUBSUB_SUBSCRIBER_STRATEGY             "pubsub.strategy"
+#define PUBSUB_SUBSCRIBER_CONFIG               "pubsub.config"
+
+#define PUBSUB_SUBSCRIBER_SCOPE_DEFAULT        "default"
+ 
+struct pubsub_multipart_callbacks_struct {
+    void *handle;
+    int (*localMsgTypeIdForMsgType)(void *handle, const char *msgType, unsigned int *msgId);
+    int (*getMultipart)(void *handle, unsigned int msgTypeId, bool retain, void **part);
+};
+typedef struct pubsub_multipart_callbacks_struct pubsub_multipart_callbacks_t;
+typedef struct pubsub_multipart_callbacks_struct* pubsub_multipart_callbacks_pt;
+ 
+struct pubsub_subscriber_struct {
+    void *handle;
+     
+    /**
+     * When a new message for a topic is available the receive will be called.
+     * 
+     * msgType contains fully qualified name of the type and msgTypeId is a local id which presents the type for performance reasons.
+     * Release can be used to instruct the pubsubadmin to release (free) the message when receive function returns. Set it to false to take
+     * over ownership of the msg (e.g. take the responsibility to free it).
+     *
+     * The callbacks argument is only valid inside the receive function, use the getMultipart callback, with retain=true, to keep multipart messages in memory.
+     * results of the localMsgTypeIdForMsgType callback are valid during the complete lifecycle of the component, not just a single receive call.
+     *
+     * Return 0 implies a successful handling. If return is not 0, the msg will always be released by the pubsubadmin.
+     *
+     * this method can be  NULL.
+     */
+    int (*receive)(void *handle, const char *msgType, unsigned int msgTypeId, void *msg, pubsub_multipart_callbacks_t *callbacks, bool *release);
+
+};
+typedef struct pubsub_subscriber_struct pubsub_subscriber_t;
+typedef struct pubsub_subscriber_struct* pubsub_subscriber_pt;
+
+
+#endif //  __PUBSUB_SUBSCRIBER_H_

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/publisher_endpoint_announce.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/publisher_endpoint_announce.h b/pubsub/pubsub_common/public/include/publisher_endpoint_announce.h
deleted file mode 100644
index bd39fc0..0000000
--- a/pubsub/pubsub_common/public/include/publisher_endpoint_announce.h
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- *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 PUBLISHER_ENDPOINT_ANNOUNCE_H_
-#define PUBLISHER_ENDPOINT_ANNOUNCE_H_
-
-#include "pubsub_endpoint.h"
-
-struct publisher_endpoint_announce {
-	void *handle;
-	celix_status_t (*announcePublisher)(void *handle, pubsub_endpoint_pt pubEP);
-	celix_status_t (*removePublisher)(void *handle, pubsub_endpoint_pt pubEP);
-	celix_status_t (*interestedInTopic)(void* handle, const char *scope, const char *topic);
-	celix_status_t (*uninterestedInTopic)(void* handle, const char *scope, const char *topic);
-};
-
-typedef struct publisher_endpoint_announce *publisher_endpoint_announce_pt;
-
-
-#endif /* PUBLISHER_ENDPOINT_ANNOUNCE_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_admin.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_admin.h b/pubsub/pubsub_common/public/include/pubsub_admin.h
deleted file mode 100644
index f24d825..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_admin.h
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_admin.h
- *
- *  \date       Sep 30, 2011
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_ADMIN_H_
-#define PUBSUB_ADMIN_H_
-
-#include "service_reference.h"
-
-#include "pubsub_common.h"
-#include "pubsub_endpoint.h"
-
-#define PSA_IP 	"PSA_IP"
-#define PSA_ITF	"PSA_INTERFACE"
-#define PSA_MULTICAST_IP_PREFIX "PSA_MC_PREFIX"
-
-#define PUBSUB_ADMIN_TYPE_KEY	"pubsub_admin.type"
-
-typedef struct pubsub_admin *pubsub_admin_pt;
-
-struct pubsub_admin_service {
-	pubsub_admin_pt admin;
-
-	celix_status_t (*addSubscription)(pubsub_admin_pt admin,pubsub_endpoint_pt subEP);
-	celix_status_t (*removeSubscription)(pubsub_admin_pt admin,pubsub_endpoint_pt subEP);
-
-	celix_status_t (*addPublication)(pubsub_admin_pt admin,pubsub_endpoint_pt subEP);
-	celix_status_t (*removePublication)(pubsub_admin_pt admin,pubsub_endpoint_pt subEP);
-
-	celix_status_t (*closeAllPublications)(pubsub_admin_pt admin,char* scope, char* topic);
-	celix_status_t (*closeAllSubscriptions)(pubsub_admin_pt admin,char* scope, char* topic);
-
-	/* Match principle:
-	 * - A full matching pubsub_admin gives 200 points
-	 * - A full matching serializer gives 100 points
-	 * - If QoS = sample
-	 * 		- fallback pubsub_admin order of selection is: udp_mc, zmq. Points allocation is 100,75.
-	 * 		- fallback serializers order of selection is: json, void. Points allocation is 30,20.
-	 * - If QoS = control
-	 * 		- fallback pubsub_admin order of selection is: zmq,udp_mc. Points allocation is 100,75.
-	 * 		- fallback serializers order of selection is: json, void. Points allocation is 30,20.
-	 * - If nothing is specified, QoS = sample is assumed, so the same score applies, just divided by two.
-	 *
-	 */
-	celix_status_t (*matchEndpoint)(pubsub_admin_pt admin, pubsub_endpoint_pt endpoint, double* score);
-};
-
-typedef struct pubsub_admin_service *pubsub_admin_service_pt;
-
-#endif /* PUBSUB_ADMIN_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_admin_match.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_admin_match.h b/pubsub/pubsub_common/public/include/pubsub_admin_match.h
deleted file mode 100644
index e95ca7d..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_admin_match.h
+++ /dev/null
@@ -1,40 +0,0 @@
-/**
- *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 PUBSUB_ADMIN_MATCH_H_
-#define PUBSUB_ADMIN_MATCH_H_
-
-#include "celix_errno.h"
-#include "properties.h"
-#include "array_list.h"
-
-#include "pubsub_serializer.h"
-
-#define QOS_ATTRIBUTE_KEY	"attribute.qos"
-#define QOS_TYPE_SAMPLE		"sample"	/* A.k.a. unreliable connection */
-#define QOS_TYPE_CONTROL	"control"	/* A.k.a. reliable connection */
-
-#define PUBSUB_ADMIN_FULL_MATCH_SCORE	200.0F
-#define SERIALIZER_FULL_MATCH_SCORE		100.0F
-
-celix_status_t pubsub_admin_match(properties_pt endpoint_props, const char *pubsub_admin_type, array_list_pt serializerList, double *score);
-celix_status_t pubsub_admin_get_best_serializer(properties_pt endpoint_props, array_list_pt serializerList, pubsub_serializer_service_t **serSvc);
-
-#endif /* PUBSUB_ADMIN_MATCH_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_common.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_common.h b/pubsub/pubsub_common/public/include/pubsub_common.h
deleted file mode 100644
index 5dfd8fd..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_common.h
+++ /dev/null
@@ -1,52 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_common.h
- *
- *  \date       Sep 17, 2015
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_COMMON_H_
-#define PUBSUB_COMMON_H_
-
-#define PUBSUB_SERIALIZER_SERVICE 		"pubsub_serializer"
-#define PUBSUB_ADMIN_SERVICE 			"pubsub_admin"
-#define PUBSUB_DISCOVERY_SERVICE		"pubsub_discovery"
-#define PUBSUB_TM_ANNOUNCE_PUBLISHER_SERVICE    "pubsub_tm_announce_publisher"
-
-#define PUBSUB_ANY_SUB_TOPIC		        "any"
-
-#define	PUBSUB_BUNDLE_ID			"bundle.id"
-
-#define MAX_SCOPE_LEN                           1024
-#define MAX_TOPIC_LEN				1024
-
-struct pubsub_msg_header{
-	char topic[MAX_TOPIC_LEN];
-	unsigned int type;
-	unsigned char major;
-	unsigned char minor;
-};
-
-typedef struct pubsub_msg_header* pubsub_msg_header_pt;
-
-
-#endif /* PUBSUB_COMMON_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_endpoint.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_endpoint.h b/pubsub/pubsub_common/public/include/pubsub_endpoint.h
deleted file mode 100644
index 8a979eb..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_endpoint.h
+++ /dev/null
@@ -1,58 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_endpoint.h
- *
- *  \date       Sep 21, 2015
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_ENDPOINT_H_
-#define PUBSUB_ENDPOINT_H_
-
-#include "service_reference.h"
-#include "listener_hook_service.h"
-#include "properties.h"
-
-#include "publisher.h"
-#include "subscriber.h"
-
-struct pubsub_endpoint {
-    char *frameworkUUID;
-    char *scope;
-    char *topic;
-    long serviceID;
-    char* endpoint;
-    bool is_secure;
-    properties_pt topic_props;
-};
-
-typedef struct pubsub_endpoint *pubsub_endpoint_pt;
-
-celix_status_t pubsubEndpoint_create(const char* fwUUID, const char* scope, const char* topic, long serviceId,const char* endpoint,properties_pt topic_props,pubsub_endpoint_pt* psEp);
-celix_status_t pubsubEndpoint_createFromServiceReference(service_reference_pt reference,pubsub_endpoint_pt* psEp, bool isPublisher);
-celix_status_t pubsubEndpoint_createFromListenerHookInfo(listener_hook_info_pt info,pubsub_endpoint_pt* psEp, bool isPublisher);
-celix_status_t pubsubEndpoint_clone(pubsub_endpoint_pt in, pubsub_endpoint_pt *out);
-celix_status_t pubsubEndpoint_destroy(pubsub_endpoint_pt psEp);
-bool pubsubEndpoint_equals(pubsub_endpoint_pt psEp1,pubsub_endpoint_pt psEp2);
-
-char *createScopeTopicKey(const char* scope, const char* topic);
-
-#endif /* PUBSUB_ENDPOINT_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_serializer.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_serializer.h b/pubsub/pubsub_common/public/include/pubsub_serializer.h
deleted file mode 100644
index 4489fa4..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_serializer.h
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_serializer.h
- *
- *  \date       Mar 24, 2017
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_SERIALIZER_SERVICE_H_
-#define PUBSUB_SERIALIZER_SERVICE_H_
-
-#include "service_reference.h"
-#include "hash_map.h"
-
-#include "pubsub_common.h"
-
-#define PUBSUB_SERIALIZER_TYPE_KEY	"pubsub_serializer.type"
-
-/**
- * There should be a pubsub_serializer_t
- * per msg type (msg id) per bundle
- *
- * The pubsub_serializer_service can create
- * a serializer_map per bundle. Potentially using
- * the extender pattern.
- */
-
-typedef struct pubsub_msg_serializer {
-	void* handle;
-	unsigned int msgId;
-	const char* msgName;
-	version_pt msgVersion;
-
-	celix_status_t (*serialize)(void* handle, const void* input, void** out, size_t* outLen);
-	celix_status_t (*deserialize)(void* handle, const void* input, size_t inputLen, void** out); //note inputLen can be 0 if predefined size is not needed
-	void (*freeMsg)(void* handle, void* msg);
-
-} pubsub_msg_serializer_t;
-
-typedef struct pubsub_serializer_service {
-	void* handle;
-
-	celix_status_t (*createSerializerMap)(void* handle, bundle_pt bundle, hash_map_pt* serializerMap);
-	celix_status_t (*destroySerializerMap)(void* handle, hash_map_pt serializerMap);
-
-} pubsub_serializer_service_t;
-
-#endif /* PUBSUB_SERIALIZER_SERVICE_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_topic_info.descriptor
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_topic_info.descriptor b/pubsub/pubsub_common/public/include/pubsub_topic_info.descriptor
deleted file mode 100644
index c01a2fd..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_topic_info.descriptor
+++ /dev/null
@@ -1,10 +0,0 @@
-:header
-type=interface
-name=pubsub_topic_info
-version=1.0.0
-:annotations
-:types
-:methods
-getParticipantsNumber(t)i=getParticipantsNumber(#am=handle;Pt#am=pre;*i)N
-getSubscribersNumber(t)i=getSubscribersNumber(#am=handle;Pt#am=pre;*i)N
-getPublishersNumber(t)i=getPublishersNumber(#am=handle;Pt#am=pre;*i)N

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/include/pubsub_utils.h
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/include/pubsub_utils.h b/pubsub/pubsub_common/public/include/pubsub_utils.h
deleted file mode 100644
index aff5c72..0000000
--- a/pubsub/pubsub_common/public/include/pubsub_utils.h
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_utils.h
- *
- *  \date       Sep 24, 2015
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_UTILS_H_
-#define PUBSUB_UTILS_H_
-
-#include "bundle_context.h"
-#include "array_list.h"
-
-char* pubsub_getScopeFromFilter(char* bundle_filter);
-char* pubsub_getTopicFromFilter(char* bundle_filter);
-char* pubsub_getKeysBundleDir(bundle_context_pt ctx);
-array_list_pt pubsub_getTopicsFromString(char* string);
-
-
-#endif /* PUBSUB_UTILS_H_ */

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/src/pubsub_admin_match.c
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/src/pubsub_admin_match.c b/pubsub/pubsub_common/public/src/pubsub_admin_match.c
deleted file mode 100644
index 2a695c1..0000000
--- a/pubsub/pubsub_common/public/src/pubsub_admin_match.c
+++ /dev/null
@@ -1,320 +0,0 @@
-/**
- *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 <string.h>
-#include "service_reference.h"
-
-#include "pubsub_admin.h"
-
-#include "pubsub_admin_match.h"
-
-#define KNOWN_PUBSUB_ADMIN_NUM	2
-#define KNOWN_SERIALIZER_NUM	2
-
-static char* qos_sample_pubsub_admin_prio_list[KNOWN_PUBSUB_ADMIN_NUM] = {"udp_mc","zmq"};
-static char* qos_sample_serializer_prio_list[KNOWN_SERIALIZER_NUM] = {"json","void"};
-
-static char* qos_control_pubsub_admin_prio_list[KNOWN_PUBSUB_ADMIN_NUM] = {"zmq","udp_mc"};
-static char* qos_control_serializer_prio_list[KNOWN_SERIALIZER_NUM] = {"json","void"};
-
-static double qos_pubsub_admin_score[KNOWN_PUBSUB_ADMIN_NUM] = {100.0F,75.0F};
-static double qos_serializer_score[KNOWN_SERIALIZER_NUM] = {30.0F,20.0F};
-
-static void get_serializer_type(service_reference_pt svcRef, char **serializerType);
-static void manage_service_from_reference(service_reference_pt svcRef, void **svc, bool getService);
-
-celix_status_t pubsub_admin_match(properties_pt endpoint_props, const char *pubsub_admin_type, array_list_pt serializerList, double *score){
-
-	celix_status_t status = CELIX_SUCCESS;
-	double final_score = 0;
-	int i = 0, j = 0;
-
-	const char *requested_admin_type 		= NULL;
-	const char *requested_serializer_type 	= NULL;
-	const char *requested_qos_type			= NULL;
-
-	if(endpoint_props!=NULL){
-		requested_admin_type 		= properties_get(endpoint_props,PUBSUB_ADMIN_TYPE_KEY);
-		requested_serializer_type 	= properties_get(endpoint_props,PUBSUB_SERIALIZER_TYPE_KEY);
-		requested_qos_type			= properties_get(endpoint_props,QOS_ATTRIBUTE_KEY);
-	}
-
-	/* Analyze the pubsub_admin */
-	if(requested_admin_type != NULL){ /* We got precise specification on the pubsub_admin we want */
-		if(strncmp(requested_admin_type,pubsub_admin_type,strlen(pubsub_admin_type))==0){ //Full match
-			final_score += PUBSUB_ADMIN_FULL_MATCH_SCORE;
-		}
-	}
-	else if(requested_qos_type != NULL){ /* We got QoS specification that will determine the selected PSA */
-		if(strncmp(requested_qos_type,QOS_TYPE_SAMPLE,strlen(QOS_TYPE_SAMPLE))==0){
-			for(i=0;i<KNOWN_PUBSUB_ADMIN_NUM;i++){
-				if(strncmp(qos_sample_pubsub_admin_prio_list[i],pubsub_admin_type,strlen(pubsub_admin_type))==0){
-					final_score += qos_pubsub_admin_score[i];
-					break;
-				}
-			}
-		}
-		else if(strncmp(requested_qos_type,QOS_TYPE_CONTROL,strlen(QOS_TYPE_CONTROL))==0){
-			for(i=0;i<KNOWN_PUBSUB_ADMIN_NUM;i++){
-				if(strncmp(qos_control_pubsub_admin_prio_list[i],pubsub_admin_type,strlen(pubsub_admin_type))==0){
-					final_score += qos_pubsub_admin_score[i];
-					break;
-				}
-			}
-		}
-		else{
-			printf("Unknown QoS type '%s'\n",requested_qos_type);
-			status = CELIX_ILLEGAL_ARGUMENT;
-		}
-	}
-	else{ /* We got no specification: fallback to Qos=Sample, but count half the score */
-		for(i=0;i<KNOWN_PUBSUB_ADMIN_NUM;i++){
-			if(strncmp(qos_sample_pubsub_admin_prio_list[i],pubsub_admin_type,strlen(pubsub_admin_type))==0){
-				final_score += (qos_pubsub_admin_score[i]/2);
-				break;
-			}
-		}
-	}
-
-	char *serializer_type = NULL;
-	/* Analyze the serializers */
-	if(requested_serializer_type != NULL){ /* We got precise specification on the serializer we want */
-		for(i=0;i<arrayList_size(serializerList);i++){
-			service_reference_pt svcRef = (service_reference_pt)arrayList_get(serializerList,i);
-			get_serializer_type(svcRef, &serializer_type);
-			if(serializer_type != NULL){
-				if(strncmp(requested_serializer_type,serializer_type,strlen(serializer_type))==0){
-					final_score += SERIALIZER_FULL_MATCH_SCORE;
-					break;
-				}
-			}
-		}
-	}
-	else if(requested_qos_type != NULL){ /* We got QoS specification that will determine the selected serializer */
-		if(strncmp(requested_qos_type,QOS_TYPE_SAMPLE,strlen(QOS_TYPE_SAMPLE))==0){
-			bool ser_found = false;
-			for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-				for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-					service_reference_pt svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-					get_serializer_type(svcRef, &serializer_type);
-					if(serializer_type != NULL){
-						if(strncmp(qos_sample_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-							ser_found = true;
-						}
-					}
-				}
-				if(ser_found){
-					final_score += qos_serializer_score[i];
-				}
-			}
-		}
-		else if(strncmp(requested_qos_type,QOS_TYPE_CONTROL,strlen(QOS_TYPE_CONTROL))==0){
-			bool ser_found = false;
-			for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-				for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-					service_reference_pt svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-					get_serializer_type(svcRef, &serializer_type);
-					if(serializer_type != NULL){
-						if(strncmp(qos_control_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-							ser_found = true;
-						}
-					}
-				}
-				if(ser_found){
-					final_score += qos_serializer_score[i];
-				}
-			}
-		}
-		else{
-			printf("Unknown QoS type '%s'\n",requested_qos_type);
-			status = CELIX_ILLEGAL_ARGUMENT;
-		}
-	}
-	else{ /* We got no specification: fallback to Qos=Sample, but count half the score */
-		bool ser_found = false;
-		for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-			for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-				service_reference_pt svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-				get_serializer_type(svcRef, &serializer_type);
-				if(serializer_type != NULL){
-					if(strncmp(qos_sample_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-						ser_found = true;
-					}
-				}
-			}
-			if(ser_found){
-				final_score += (qos_serializer_score[i]/2);
-			}
-		}
-	}
-
-	*score = final_score;
-
-	printf("Score for pair <%s,%s> = %f\n",pubsub_admin_type,serializer_type,final_score);
-
-	return status;
-}
-
-celix_status_t pubsub_admin_get_best_serializer(properties_pt endpoint_props, array_list_pt serializerList, pubsub_serializer_service_t **serSvc){
-	celix_status_t status = CELIX_SUCCESS;
-
-	int i = 0, j = 0;
-
-	const char *requested_serializer_type = NULL;
-	const char *requested_qos_type = NULL;
-
-	if (endpoint_props != NULL){
-		requested_serializer_type = properties_get(endpoint_props,PUBSUB_SERIALIZER_TYPE_KEY);
-		requested_qos_type = properties_get(endpoint_props,QOS_ATTRIBUTE_KEY);
-	}
-
-	service_reference_pt svcRef = NULL;
-	void *svc = NULL;
-
-	/* Analyze the serializers */
-	if(requested_serializer_type != NULL){ /* We got precise specification on the serializer we want */
-		for(i=0;i<arrayList_size(serializerList);i++){
-			svcRef = (service_reference_pt)arrayList_get(serializerList,i);
-			char *serializer_type = NULL;
-			get_serializer_type(svcRef, &serializer_type);
-			if(serializer_type != NULL){
-				if(strncmp(requested_serializer_type,serializer_type,strlen(serializer_type))==0){
-					manage_service_from_reference(svcRef, &svc,true);
-					if(svc==NULL){
-						printf("Cannot get pubsub_serializer_service from serviceReference %p\n",svcRef);
-						status = CELIX_SERVICE_EXCEPTION;
-					}
-					*serSvc = svc;
-					break;
-				}
-			}
-		}
-	}
-	else if(requested_qos_type != NULL){ /* We got QoS specification that will determine the selected serializer */
-		if(strncmp(requested_qos_type,QOS_TYPE_SAMPLE,strlen(QOS_TYPE_SAMPLE))==0){
-			bool ser_found = false;
-			for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-				for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-					svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-					char *serializer_type = NULL;
-					get_serializer_type(svcRef, &serializer_type);
-					if(serializer_type != NULL){
-						if(strncmp(qos_sample_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-							manage_service_from_reference(svcRef, &svc,true);
-							if(svc==NULL){
-								printf("Cannot get pubsub_serializer_service from serviceReference %p\n",svcRef);
-								status = CELIX_SERVICE_EXCEPTION;
-							}
-							else{
-								*serSvc = svc;
-								ser_found = true;
-								printf("Selected %s serializer as best for QoS=%s\n",qos_sample_serializer_prio_list[i],QOS_TYPE_SAMPLE);
-							}
-						}
-					}
-				}
-			}
-		}
-		else if(strncmp(requested_qos_type,QOS_TYPE_CONTROL,strlen(QOS_TYPE_CONTROL))==0){
-			bool ser_found = false;
-			for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-				for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-					svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-					char *serializer_type = NULL;
-					get_serializer_type(svcRef, &serializer_type);
-					if(serializer_type != NULL){
-						if(strncmp(qos_control_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-							manage_service_from_reference(svcRef, &svc,true);
-							if(svc==NULL){
-								printf("Cannot get pubsub_serializer_service from serviceReference %p\n",svcRef);
-								status = CELIX_SERVICE_EXCEPTION;
-							}
-							else{
-								*serSvc = svc;
-								ser_found = true;
-								printf("Selected %s serializer as best for QoS=%s\n",qos_control_serializer_prio_list[i],QOS_TYPE_CONTROL);
-							}
-						}
-					}
-				}
-			}
-		}
-		else{
-			printf("Unknown QoS type '%s'\n",requested_qos_type);
-			status = CELIX_ILLEGAL_ARGUMENT;
-		}
-	}
-	else{ /* We got no specification: fallback to Qos=Sample, but count half the score */
-		bool ser_found = false;
-		for(i=0;i<KNOWN_SERIALIZER_NUM && !ser_found;i++){
-			for(j=0;j<arrayList_size(serializerList) && !ser_found;j++){
-				svcRef = (service_reference_pt)arrayList_get(serializerList,j);
-				char *serializer_type = NULL;
-				get_serializer_type(svcRef, &serializer_type);
-				if(serializer_type != NULL){
-					if(strncmp(qos_sample_serializer_prio_list[i],serializer_type,strlen(serializer_type))==0){
-						manage_service_from_reference(svcRef, &svc,true);
-						if(svc==NULL){
-							printf("Cannot get pubsub_serializer_service from serviceReference %p\n",svcRef);
-							status = CELIX_SERVICE_EXCEPTION;
-						}
-						else{
-							*serSvc = svc;
-							ser_found = true;
-							printf("Selected %s serializer as best without any specification\n",qos_sample_serializer_prio_list[i]);
-						}
-					}
-				}
-			}
-		}
-	}
-
-	if(svc!=NULL && svcRef!=NULL){
-		manage_service_from_reference(svcRef, svc, false);
-	}
-
-	return status;
-}
-
-static void get_serializer_type(service_reference_pt svcRef, char **serializerType){
-
-	const char *serType = NULL;
-	serviceReference_getProperty(svcRef, PUBSUB_SERIALIZER_TYPE_KEY,&serType);
-	if(serType != NULL){
-		*serializerType = (char*)serType;
-	}
-	else{
-		printf("Serializer serviceReference %p has no pubsub_serializer.type property specified\n",svcRef);
-		*serializerType = NULL;
-	}
-}
-
-static void manage_service_from_reference(service_reference_pt svcRef, void **svc, bool getService){
-	bundle_context_pt context = NULL;
-	bundle_pt bundle = NULL;
-	serviceReference_getBundle(svcRef, &bundle);
-	bundle_getContext(bundle, &context);
-	if(getService){
-		bundleContext_getService(context, svcRef, svc);
-	}
-	else{
-		bundleContext_ungetService(context, svcRef, NULL);
-	}
-}

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/src/pubsub_endpoint.c
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/src/pubsub_endpoint.c b/pubsub/pubsub_common/public/src/pubsub_endpoint.c
deleted file mode 100644
index c3fd293..0000000
--- a/pubsub/pubsub_common/public/src/pubsub_endpoint.c
+++ /dev/null
@@ -1,254 +0,0 @@
-/**
- *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.
- */
-/*
- * endpoint_description.c
- *
- *  \date       25 Jul 2014
- *  \author     <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright  Apache License, Version 2.0
- */
-
-#include <string.h>
-#include <stdlib.h>
-
-#include "celix_errno.h"
-#include "celix_log.h"
-
-#include "pubsub_common.h"
-#include "pubsub_endpoint.h"
-#include "constants.h"
-
-#include "pubsub_utils.h"
-
-
-static void pubsubEndpoint_setFields(pubsub_endpoint_pt psEp, const char* fwUUID, const char* scope, const char* topic, long serviceId,const char* endpoint,properties_pt topic_props, bool cloneProps);
-static properties_pt pubsubEndpoint_getTopicProperties(bundle_pt bundle, const char *topic, bool isPublisher);
-
-static void pubsubEndpoint_setFields(pubsub_endpoint_pt psEp, const char* fwUUID, const char* scope, const char* topic, long serviceId,const char* endpoint,properties_pt topic_props, bool cloneProps){
-
-	if (fwUUID != NULL) {
-		psEp->frameworkUUID = strdup(fwUUID);
-	}
-
-	if (scope != NULL) {
-		psEp->scope = strdup(scope);
-	}
-
-	if (topic != NULL) {
-		psEp->topic = strdup(topic);
-	}
-
-	psEp->serviceID = serviceId;
-
-	if(endpoint != NULL) {
-		psEp->endpoint = strdup(endpoint);
-	}
-
-	if(topic_props != NULL){
-		if(cloneProps){
-			properties_copy(topic_props, &(psEp->topic_props));
-		}
-		else{
-			psEp->topic_props = topic_props;
-		}
-	}
-}
-
-static properties_pt pubsubEndpoint_getTopicProperties(bundle_pt bundle, const char *topic, bool isPublisher){
-
-	properties_pt topic_props = NULL;
-
-	bool isSystemBundle = false;
-	bundle_isSystemBundle(bundle, &isSystemBundle);
-	long bundleId = -1;
-	bundle_isSystemBundle(bundle, &isSystemBundle);
-	bundle_getBundleId(bundle,&bundleId);
-
-	if(isSystemBundle == false) {
-
-		char *bundleRoot = NULL;
-		char* topicPropertiesPath = NULL;
-		bundle_getEntry(bundle, ".", &bundleRoot);
-
-		if(bundleRoot != NULL){
-
-			asprintf(&topicPropertiesPath, "%s/META-INF/topics/%s/%s.properties", bundleRoot, isPublisher?"pub":"sub", topic);
-			topic_props = properties_load(topicPropertiesPath);
-			if(topic_props==NULL){
-				printf("PSEP: Could not load properties for %s on topic %s, bundleId=%ld\n", isPublisher?"publication":"subscription", topic,bundleId);
-			}
-
-			free(topicPropertiesPath);
-			free(bundleRoot);
-		}
-	}
-
-	return topic_props;
-}
-
-celix_status_t pubsubEndpoint_create(const char* fwUUID, const char* scope, const char* topic, long serviceId,const char* endpoint,properties_pt topic_props,pubsub_endpoint_pt* psEp){
-	celix_status_t status = CELIX_SUCCESS;
-
-	*psEp = calloc(1, sizeof(**psEp));
-
-	pubsubEndpoint_setFields(*psEp, fwUUID, scope, topic, serviceId, endpoint, topic_props, true);
-
-	return status;
-
-}
-
-celix_status_t pubsubEndpoint_clone(pubsub_endpoint_pt in, pubsub_endpoint_pt *out){
-	celix_status_t status = CELIX_SUCCESS;
-
-	*out = calloc(1,sizeof(**out));
-
-	pubsubEndpoint_setFields(*out, in->frameworkUUID, in->scope, in->topic, in->serviceID, in->endpoint, in->topic_props, true);
-
-	return status;
-
-}
-
-celix_status_t pubsubEndpoint_createFromServiceReference(service_reference_pt reference, pubsub_endpoint_pt* psEp, bool isPublisher){
-	celix_status_t status = CELIX_SUCCESS;
-
-	pubsub_endpoint_pt ep = calloc(1,sizeof(*ep));
-
-	bundle_pt bundle = NULL;
-	bundle_context_pt ctxt = NULL;
-	const char* fwUUID = NULL;
-	serviceReference_getBundle(reference,&bundle);
-	bundle_getContext(bundle,&ctxt);
-	bundleContext_getProperty(ctxt,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
-
-	const char* scope = NULL;
-	serviceReference_getProperty(reference, PUBSUB_SUBSCRIBER_SCOPE,&scope);
-
-	const char* topic = NULL;
-	serviceReference_getProperty(reference, PUBSUB_SUBSCRIBER_TOPIC,&topic);
-
-	const char* serviceId = NULL;
-	serviceReference_getProperty(reference,(char*)OSGI_FRAMEWORK_SERVICE_ID,&serviceId);
-
-	/* TODO: is topic_props==NULL a fatal error such that EP cannot be created? */
-	properties_pt topic_props = pubsubEndpoint_getTopicProperties(bundle, topic, isPublisher);
-
-	pubsubEndpoint_setFields(ep, fwUUID, scope!=NULL?scope:PUBSUB_SUBSCRIBER_SCOPE_DEFAULT, topic, strtol(serviceId,NULL,10), NULL, topic_props, false);
-
-	if (!ep->frameworkUUID || !ep->serviceID || !ep->scope || !ep->topic) {
-		fw_log(logger, OSGI_FRAMEWORK_LOG_ERROR, "PUBSUB_ENDPOINT: incomplete description!.");
-		status = CELIX_BUNDLE_EXCEPTION;
-		pubsubEndpoint_destroy(ep);
-		*psEp = NULL;
-	}
-	else{
-		*psEp = ep;
-	}
-
-	return status;
-
-}
-
-celix_status_t pubsubEndpoint_createFromListenerHookInfo(listener_hook_info_pt info,pubsub_endpoint_pt* psEp, bool isPublisher){
-	celix_status_t status = CELIX_SUCCESS;
-
-	const char* fwUUID=NULL;
-	bundleContext_getProperty(info->context,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
-
-	if(fwUUID==NULL){
-		return CELIX_BUNDLE_EXCEPTION;
-	}
-
-	char* topic = pubsub_getTopicFromFilter(info->filter);
-	if(topic==NULL){
-		return CELIX_BUNDLE_EXCEPTION;
-	}
-
-	*psEp = calloc(1, sizeof(**psEp));
-
-	char* scope = pubsub_getScopeFromFilter(info->filter);
-	if(scope == NULL) {
-		scope = strdup(PUBSUB_PUBLISHER_SCOPE_DEFAULT);
-	}
-
-	bundle_pt bundle = NULL;
-	long bundleId = -1;
-	bundleContext_getBundle(info->context,&bundle);
-
-	bundle_getBundleId(bundle,&bundleId);
-
-	properties_pt topic_props = pubsubEndpoint_getTopicProperties(bundle, topic, isPublisher);
-
-	/* TODO: is topic_props==NULL a fatal error such that EP cannot be created? */
-	pubsubEndpoint_setFields(*psEp, fwUUID, scope!=NULL?scope:PUBSUB_SUBSCRIBER_SCOPE_DEFAULT, topic, bundleId, NULL, topic_props, false);
-
-	free(topic);
-	free(scope);
-
-
-	return status;
-}
-
-celix_status_t pubsubEndpoint_destroy(pubsub_endpoint_pt psEp){
-
-	if(psEp->frameworkUUID!=NULL){
-		free(psEp->frameworkUUID);
-		psEp->frameworkUUID = NULL;
-	}
-
-	if(psEp->scope!=NULL){
-		free(psEp->scope);
-		psEp->scope = NULL;
-	}
-
-	if(psEp->topic!=NULL){
-		free(psEp->topic);
-		psEp->topic = NULL;
-	}
-
-	if(psEp->endpoint!=NULL){
-		free(psEp->endpoint);
-		psEp->endpoint = NULL;
-	}
-
-	if(psEp->topic_props != NULL){
-		properties_destroy(psEp->topic_props);
-	}
-
-	free(psEp);
-
-	return CELIX_SUCCESS;
-
-}
-
-bool pubsubEndpoint_equals(pubsub_endpoint_pt psEp1,pubsub_endpoint_pt psEp2){
-
-	return ((strcmp(psEp1->frameworkUUID,psEp2->frameworkUUID)==0) &&
-			(strcmp(psEp1->scope,psEp2->scope)==0) &&
-			(strcmp(psEp1->topic,psEp2->topic)==0) &&
-			(psEp1->serviceID == psEp2->serviceID) /*&&
-			((psEp1->endpoint==NULL && psEp2->endpoint==NULL)||(strcmp(psEp1->endpoint,psEp2->endpoint)==0))*/
-	);
-}
-
-char *createScopeTopicKey(const char* scope, const char* topic) {
-	char *result = NULL;
-	asprintf(&result, "%s:%s", scope, topic);
-
-	return result;
-}

http://git-wip-us.apache.org/repos/asf/celix/blob/81804e00/pubsub/pubsub_common/public/src/pubsub_utils.c
----------------------------------------------------------------------
diff --git a/pubsub/pubsub_common/public/src/pubsub_utils.c b/pubsub/pubsub_common/public/src/pubsub_utils.c
deleted file mode 100644
index abc5ae6..0000000
--- a/pubsub/pubsub_common/public/src/pubsub_utils.c
+++ /dev/null
@@ -1,170 +0,0 @@
-/**
- *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.
- */
-/*
- * pubsub_utils.c
- *
- *  \date       Sep 24, 2015
- *  \author    	<a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- *  \copyright	Apache License, Version 2.0
- */
-
-#include <string.h>
-#include <stdlib.h>
-
-#include "constants.h"
-
-#include "pubsub_common.h"
-#include "publisher.h"
-#include "pubsub_utils.h"
-
-#include "array_list.h"
-#include "bundle.h"
-
-#include <unistd.h>
-#include <sys/types.h>
-#include <sys/stat.h>
-
-#define MAX_KEYBUNDLE_LENGTH 256
-
-char* pubsub_getScopeFromFilter(char* bundle_filter){
-
-	char* scope = NULL;
-
-	char* filter = strdup(bundle_filter);
-
-	char* oc = strstr(filter,OSGI_FRAMEWORK_OBJECTCLASS);
-	if(oc!=NULL){
-		oc+=strlen(OSGI_FRAMEWORK_OBJECTCLASS)+1;
-		if(strncmp(oc,PUBSUB_PUBLISHER_SERVICE_NAME,strlen(PUBSUB_PUBLISHER_SERVICE_NAME))==0){
-
-			char* scopes = strstr(filter,PUBSUB_PUBLISHER_SCOPE);
-			if(scopes!=NULL){
-
-				scopes+=strlen(PUBSUB_PUBLISHER_SCOPE)+1;
-				char* bottom=strchr(scopes,')');
-				*bottom='\0';
-
-				scope=strdup(scopes);
-			} else {
-			    scope=strdup(PUBSUB_PUBLISHER_SCOPE_DEFAULT);
-			}
-		}
-	}
-
-	free(filter);
-
-	return scope;
-}
-
-char* pubsub_getTopicFromFilter(char* bundle_filter){
-
-	char* topic = NULL;
-
-	char* filter = strdup(bundle_filter);
-
-	char* oc = strstr(filter,OSGI_FRAMEWORK_OBJECTCLASS);
-	if(oc!=NULL){
-		oc+=strlen(OSGI_FRAMEWORK_OBJECTCLASS)+1;
-		if(strncmp(oc,PUBSUB_PUBLISHER_SERVICE_NAME,strlen(PUBSUB_PUBLISHER_SERVICE_NAME))==0){
-
-			char* topics = strstr(filter,PUBSUB_PUBLISHER_TOPIC);
-			if(topics!=NULL){
-
-				topics+=strlen(PUBSUB_PUBLISHER_TOPIC)+1;
-				char* bottom=strchr(topics,')');
-				*bottom='\0';
-
-				topic=strdup(topics);
-
-			}
-		}
-	}
-
-	free(filter);
-
-	return topic;
-
-}
-
-array_list_pt pubsub_getTopicsFromString(char* string){
-
-	array_list_pt topic_list = NULL;
-	arrayList_create(&topic_list);
-
-	char* topics = strdup(string);
-
-	char* topic = strtok(topics,",;|# ");
-	arrayList_add(topic_list,strdup(topic));
-
-	while( (topic = strtok(NULL,",;|# ")) !=NULL){
-		arrayList_add(topic_list,strdup(topic));
-	}
-
-	free(topics);
-
-	return topic_list;
-
-}
-
-/**
- * Loop through all bundles and look for the bundle with the keys inside.
- * If no key bundle found, return NULL
- *
- * Caller is responsible for freeing the object
- */
-char* pubsub_getKeysBundleDir(bundle_context_pt ctx)
-{
-	array_list_pt bundles = NULL;
-	bundleContext_getBundles(ctx, &bundles);
-	int nrOfBundles = arrayList_size(bundles);
-	long bundle_id = -1;
-	char* result = NULL;
-
-	for (int i = 0; i < nrOfBundles; i++){
-		bundle_pt b = arrayList_get(bundles, i);
-
-		/* Skip bundle 0 (framework bundle) since it has no path nor revisions */
-		bundle_getBundleId(b, &bundle_id);
-		if(bundle_id==0){
-			continue;
-		}
-
-		char* dir = NULL;
-		bundle_getEntry(b, ".", &dir);
-
-		char cert_dir[MAX_KEYBUNDLE_LENGTH];
-		snprintf(cert_dir, MAX_KEYBUNDLE_LENGTH, "%s/META-INF/keys", dir);
-
-		struct stat s;
-		int err = stat(cert_dir, &s);
-		if (err != -1){
-			if (S_ISDIR(s.st_mode)){
-				result = dir;
-				break;
-			}
-		}
-
-		free(dir);
-	}
-
-	arrayList_destroy(bundles);
-
-	return result;
-}
-


Mime
View raw message