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 2017/02/06 18:34:26 UTC
[14/19] celix git commit: CELIX-389: Refactors pubsub.
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/celix-pubsub/pubsub/pubsub_topology_manager/private/include/pubsub_topology_manager.h
----------------------------------------------------------------------
diff --git a/celix-pubsub/pubsub/pubsub_topology_manager/private/include/pubsub_topology_manager.h b/celix-pubsub/pubsub/pubsub_topology_manager/private/include/pubsub_topology_manager.h
deleted file mode 100644
index 2e940aa..0000000
--- a/celix-pubsub/pubsub/pubsub_topology_manager/private/include/pubsub_topology_manager.h
+++ /dev/null
@@ -1,86 +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_topology_manager.h
- *
- * \date Sep 29, 2011
- * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- * \copyright Apache License, Version 2.0
- */
-
-#ifndef PUBSUB_TOPOLOGY_MANAGER_H_
-#define PUBSUB_TOPOLOGY_MANAGER_H_
-
-#include "endpoint_listener.h"
-#include "service_reference.h"
-#include "bundle_context.h"
-#include "log_helper.h"
-
-#include "pubsub_common.h"
-#include "pubsub_endpoint.h"
-#include "publisher.h"
-#include "subscriber.h"
-
-
-struct pubsub_topology_manager {
- bundle_context_pt context;
-
- celix_thread_mutex_t psaListLock;
- array_list_pt psaList;
-
- celix_thread_mutex_t discoveryListLock;
- hash_map_pt discoveryList; //<serviceReference,NULL>
-
- celix_thread_mutex_t publicationsLock;
- hash_map_pt publications; //<topic(string),list<pubsub_ep>>
-
- celix_thread_mutex_t subscriptionsLock;
- hash_map_pt subscriptions; //<topic(string),list<pubsub_ep>>
-
- log_helper_pt loghelper;
-};
-
-typedef struct pubsub_topology_manager *pubsub_topology_manager_pt;
-
-celix_status_t pubsub_topologyManager_create(bundle_context_pt context, log_helper_pt logHelper, pubsub_topology_manager_pt *manager);
-celix_status_t pubsub_topologyManager_destroy(pubsub_topology_manager_pt manager);
-celix_status_t pubsub_topologyManager_closeImports(pubsub_topology_manager_pt manager);
-
-celix_status_t pubsub_topologyManager_psaAdding(void *handle, service_reference_pt reference, void **service);
-celix_status_t pubsub_topologyManager_psaAdded(void *handle, service_reference_pt reference, void *service);
-celix_status_t pubsub_topologyManager_psaModified(void *handle, service_reference_pt reference, void *service);
-celix_status_t pubsub_topologyManager_psaRemoved(void *handle, service_reference_pt reference, void *service);
-
-celix_status_t pubsub_topologyManager_pubsubDiscoveryAdding(void* handle, service_reference_pt reference, void** service);
-celix_status_t pubsub_topologyManager_pubsubDiscoveryAdded(void* handle, service_reference_pt reference, void* service);
-celix_status_t pubsub_topologyManager_pubsubDiscoveryModified(void * handle, service_reference_pt reference, void* service);
-celix_status_t pubsub_topologyManager_pubsubDiscoveryRemoved(void * handle, service_reference_pt reference, void* service);
-
-celix_status_t pubsub_topologyManager_subscriberAdding(void * handle, service_reference_pt reference, void **service);
-celix_status_t pubsub_topologyManager_subscriberAdded(void * handle, service_reference_pt reference, void * service);
-celix_status_t pubsub_topologyManager_subscriberModified(void * handle, service_reference_pt reference, void * service);
-celix_status_t pubsub_topologyManager_subscriberRemoved(void * handle, service_reference_pt reference, void * service);
-
-celix_status_t pubsub_topologyManager_publisherTrackerAdded(void *handle, array_list_pt listeners);
-celix_status_t pubsub_topologyManager_publisherTrackerRemoved(void *handle, array_list_pt listeners);
-
-celix_status_t pubsub_topologyManager_announcePublisher(void *handle, pubsub_endpoint_pt pubEP);
-celix_status_t pubsub_topologyManager_removePublisher(void *handle, pubsub_endpoint_pt pubEP);
-
-#endif /* PUBSUB_TOPOLOGY_MANAGER_H_ */
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pstm_activator.c
----------------------------------------------------------------------
diff --git a/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pstm_activator.c b/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pstm_activator.c
deleted file mode 100644
index a35d161..0000000
--- a/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pstm_activator.c
+++ /dev/null
@@ -1,248 +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.
- */
-/*
- * pstm_activator.c
- *
- * \date Sep 29, 2011
- * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- * \copyright Apache License, Version 2.0
- */
-
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-
-#include "constants.h"
-#include "bundle_activator.h"
-#include "service_tracker.h"
-#include "service_registration.h"
-
-#include "endpoint_listener.h"
-#include "remote_constants.h"
-#include "listener_hook_service.h"
-#include "log_service.h"
-#include "log_helper.h"
-
-
-#include "pubsub_topology_manager.h"
-#include "publisher_endpoint_announce.h"
-
-struct activator {
- bundle_context_pt context;
-
- pubsub_topology_manager_pt manager;
-
- service_tracker_pt pubsubDiscoveryTracker;
- service_tracker_pt pubsubAdminTracker;
- service_tracker_pt pubsubSubscribersTracker;
-
- listener_hook_service_pt hookService;
- service_registration_pt hook;
-
- publisher_endpoint_announce_pt publisherEPDiscover;
- service_registration_pt publisherEPDiscoverService;
-
- log_helper_pt loghelper;
-};
-
-
-static celix_status_t bundleActivator_createPSDTracker(struct activator *activator, service_tracker_pt *tracker);
-static celix_status_t bundleActivator_createPSATracker(struct activator *activator, service_tracker_pt *tracker);
-static celix_status_t bundleActivator_createPSSubTracker(struct activator *activator, service_tracker_pt *tracker);
-
-
-static celix_status_t bundleActivator_createPSDTracker(struct activator *activator, service_tracker_pt *tracker) {
- celix_status_t status;
-
- service_tracker_customizer_pt customizer = NULL;
-
- status = serviceTrackerCustomizer_create(activator->manager,
- pubsub_topologyManager_pubsubDiscoveryAdding,
- pubsub_topologyManager_pubsubDiscoveryAdded,
- pubsub_topologyManager_pubsubDiscoveryModified,
- pubsub_topologyManager_pubsubDiscoveryRemoved,
- &customizer);
-
- if (status == CELIX_SUCCESS) {
- status = serviceTracker_create(activator->context, (char *) PUBSUB_DISCOVERY_SERVICE, customizer, tracker);
- }
-
- return status;
-}
-
-static celix_status_t bundleActivator_createPSATracker(struct activator *activator, service_tracker_pt *tracker) {
- celix_status_t status = CELIX_SUCCESS;
-
- service_tracker_customizer_pt customizer = NULL;
-
- status = serviceTrackerCustomizer_create(activator->manager,
- pubsub_topologyManager_psaAdding,
- pubsub_topologyManager_psaAdded,
- pubsub_topologyManager_psaModified,
- pubsub_topologyManager_psaRemoved,
- &customizer);
-
- if (status == CELIX_SUCCESS) {
- status = serviceTracker_create(activator->context, PUBSUB_ADMIN_SERVICE, customizer, tracker);
- }
-
- return status;
-}
-
-static celix_status_t bundleActivator_createPSSubTracker(struct activator *activator, service_tracker_pt *tracker) {
- celix_status_t status = CELIX_SUCCESS;
-
- service_tracker_customizer_pt customizer = NULL;
-
- status = serviceTrackerCustomizer_create(activator->manager,
- pubsub_topologyManager_subscriberAdding,
- pubsub_topologyManager_subscriberAdded,
- pubsub_topologyManager_subscriberModified,
- pubsub_topologyManager_subscriberRemoved,
- &customizer);
-
- if (status == CELIX_SUCCESS) {
- status = serviceTracker_create(activator->context, PUBSUB_SUBSCRIBER_SERVICE_NAME, customizer, tracker);
- }
-
- return status;
-}
-
-celix_status_t bundleActivator_create(bundle_context_pt context, void **userData) {
- celix_status_t status = CELIX_SUCCESS;
- struct activator *activator = NULL;
-
- activator = calloc(1,sizeof(struct activator));
-
- if (!activator) {
- return CELIX_ENOMEM;
- }
-
- activator->context = context;
-
- logHelper_create(context, &activator->loghelper);
- logHelper_start(activator->loghelper);
-
- status = pubsub_topologyManager_create(context, activator->loghelper, &activator->manager);
- if (status == CELIX_SUCCESS) {
- status = bundleActivator_createPSDTracker(activator, &activator->pubsubDiscoveryTracker);
- if (status == CELIX_SUCCESS) {
- status = bundleActivator_createPSATracker(activator, &activator->pubsubAdminTracker);
- if (status == CELIX_SUCCESS) {
- status = bundleActivator_createPSSubTracker(activator, &activator->pubsubSubscribersTracker);
- if (status == CELIX_SUCCESS) {
- *userData = activator;
- }
- if (status != CELIX_SUCCESS){
- serviceTracker_destroy(activator->pubsubAdminTracker);
- }
- }
- if (status != CELIX_SUCCESS){
- serviceTracker_destroy(activator->pubsubDiscoveryTracker);
- }
- }
- if (status != CELIX_SUCCESS){
- pubsub_topologyManager_destroy(activator->manager);
- }
- }
- if (status != CELIX_SUCCESS){ // an exception occurred so free allocated memory
- logHelper_stop(activator->loghelper);
- logHelper_destroy(&activator->loghelper);
- free(activator);
-
- }
-
- return status;
-}
-
-
-celix_status_t bundleActivator_start(void * userData, bundle_context_pt context) {
- celix_status_t status = CELIX_SUCCESS;
- struct activator *activator = userData;
-
- publisher_endpoint_announce_pt pubEPDiscover = calloc(1, sizeof(*pubEPDiscover));
- pubEPDiscover->handle = activator->manager;
- pubEPDiscover->announcePublisher = pubsub_topologyManager_announcePublisher;
- pubEPDiscover->removePublisher = pubsub_topologyManager_removePublisher;
- activator->publisherEPDiscover = pubEPDiscover;
-
- status += bundleContext_registerService(context, (char *) PUBSUB_TM_ANNOUNCE_PUBLISHER_SERVICE, pubEPDiscover, NULL, &activator->publisherEPDiscoverService);
-
-
- listener_hook_service_pt hookService = calloc(1,sizeof(*hookService));
- hookService->handle = activator->manager;
- hookService->added = pubsub_topologyManager_publisherTrackerAdded;
- hookService->removed = pubsub_topologyManager_publisherTrackerRemoved;
- activator->hookService = hookService;
-
- status += bundleContext_registerService(context, (char *) OSGI_FRAMEWORK_LISTENER_HOOK_SERVICE_NAME, hookService, NULL, &activator->hook);
-
- /* 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);
- status += bundleContext_registerService(context, (char *) PUBSUB_TOPIC_INFO_SERVICE, activator->topicInfo, props, &activator->topicInfoService);
- */
- status += serviceTracker_open(activator->pubsubAdminTracker);
-
- status += serviceTracker_open(activator->pubsubDiscoveryTracker);
-
- status += serviceTracker_open(activator->pubsubSubscribersTracker);
-
-
- return status;
-}
-
-celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context) {
- celix_status_t status = CELIX_SUCCESS;
- struct activator *activator = userData;
-
- serviceTracker_close(activator->pubsubSubscribersTracker);
- serviceTracker_close(activator->pubsubDiscoveryTracker);
- serviceTracker_close(activator->pubsubAdminTracker);
-
- serviceRegistration_unregister(activator->publisherEPDiscoverService);
- free(activator->publisherEPDiscover);
-
- serviceRegistration_unregister(activator->hook);
- free(activator->hookService);
-
- return status;
-}
-
-celix_status_t bundleActivator_destroy(void * userData, bundle_context_pt context) {
- celix_status_t status = CELIX_SUCCESS;
-
- struct activator *activator = userData;
- if (!activator || !activator->manager) {
- status = CELIX_BUNDLE_EXCEPTION;
- } else {
-
- serviceTracker_destroy(activator->pubsubSubscribersTracker);
- serviceTracker_destroy(activator->pubsubDiscoveryTracker);
- serviceTracker_destroy(activator->pubsubAdminTracker);
-
- logHelper_stop(activator->loghelper);
- logHelper_destroy(&activator->loghelper);
-
- status = pubsub_topologyManager_destroy(activator->manager);
- free(activator);
- }
-
- return status;
-}
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pubsub_topology_manager.c
----------------------------------------------------------------------
diff --git a/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pubsub_topology_manager.c b/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pubsub_topology_manager.c
deleted file mode 100644
index a6541b9..0000000
--- a/celix-pubsub/pubsub/pubsub_topology_manager/private/src/pubsub_topology_manager.c
+++ /dev/null
@@ -1,862 +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_topology_manager.c
- *
- * \date Sep 29, 2011
- * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
- * \copyright Apache License, Version 2.0
- */
-
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <stdbool.h>
-
-#include "hash_map.h"
-#include "array_list.h"
-#include "bundle_context.h"
-#include "constants.h"
-#include "module.h"
-#include "bundle.h"
-#include "remote_service_admin.h"
-#include "remote_constants.h"
-#include "filter.h"
-#include "listener_hook_service.h"
-#include "utils.h"
-#include "service_reference.h"
-#include "service_registration.h"
-#include "log_service.h"
-#include "log_helper.h"
-
-#include "publisher_endpoint_announce.h"
-#include "pubsub_topology_manager.h"
-#include "pubsub_endpoint.h"
-#include "pubsub_admin.h"
-#include "pubsub_utils.h"
-
-
-celix_status_t pubsub_topologyManager_create(bundle_context_pt context, log_helper_pt logHelper, pubsub_topology_manager_pt *manager) {
- celix_status_t status = CELIX_SUCCESS;
-
- *manager = calloc(1, sizeof(**manager));
- if (!*manager) {
- return CELIX_ENOMEM;
- }
-
- (*manager)->context = context;
-
- celix_thread_mutexattr_t psaAttr;
- celixThreadMutexAttr_create(&psaAttr);
- celixThreadMutexAttr_settype(&psaAttr, CELIX_THREAD_MUTEX_RECURSIVE);
- status = celixThreadMutex_create(&(*manager)->psaListLock, &psaAttr);
- celixThreadMutexAttr_destroy(&psaAttr);
-
- status = celixThreadMutex_create(&(*manager)->publicationsLock, NULL);
- status = celixThreadMutex_create(&(*manager)->subscriptionsLock, NULL);
- status = celixThreadMutex_create(&(*manager)->discoveryListLock, NULL);
-
- arrayList_create(&(*manager)->psaList);
-
- (*manager)->discoveryList = hashMap_create(serviceReference_hashCode, NULL, serviceReference_equals2, NULL);
- (*manager)->publications = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
- (*manager)->subscriptions = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL);
-
- (*manager)->loghelper = logHelper;
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_destroy(pubsub_topology_manager_pt manager) {
- celix_status_t status = CELIX_SUCCESS;
-
- celixThreadMutex_lock(&manager->discoveryListLock);
- hashMap_destroy(manager->discoveryList, false, false);
- celixThreadMutex_unlock(&manager->discoveryListLock);
- celixThreadMutex_destroy(&manager->discoveryListLock);
-
- celixThreadMutex_lock(&manager->psaListLock);
- arrayList_destroy(manager->psaList);
- celixThreadMutex_unlock(&manager->psaListLock);
- celixThreadMutex_destroy(&manager->psaListLock);
-
- celixThreadMutex_lock(&manager->publicationsLock);
- hash_map_iterator_pt pubit = hashMapIterator_create(manager->publications);
- while(hashMapIterator_hasNext(pubit)){
- array_list_pt l = (array_list_pt)hashMapIterator_nextValue(pubit);
- int i;
- for(i=0;i<arrayList_size(l);i++){
- pubsubEndpoint_destroy((pubsub_endpoint_pt)arrayList_get(l,i));
- }
- arrayList_destroy(l);
- }
- hashMapIterator_destroy(pubit);
- hashMap_destroy(manager->publications, true, false);
- celixThreadMutex_unlock(&manager->publicationsLock);
- celixThreadMutex_destroy(&manager->publicationsLock);
-
- celixThreadMutex_lock(&manager->subscriptionsLock);
- hash_map_iterator_pt subit = hashMapIterator_create(manager->subscriptions);
- while(hashMapIterator_hasNext(subit)){
- array_list_pt l = (array_list_pt)hashMapIterator_nextValue(subit);
- int i;
- for(i=0;i<arrayList_size(l);i++){
- pubsubEndpoint_destroy((pubsub_endpoint_pt)arrayList_get(l,i));
- }
- arrayList_destroy(l);
- }
- hashMapIterator_destroy(subit);
- hashMap_destroy(manager->subscriptions, true, false);
- celixThreadMutex_unlock(&manager->subscriptionsLock);
- celixThreadMutex_destroy(&manager->subscriptionsLock);
-
- free(manager);
-
- return status;
-}
-
-
-celix_status_t pubsub_topologyManager_psaAdding(void * handle, service_reference_pt reference, void **service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- status = bundleContext_getService(manager->context, reference, service);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_psaAdded(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
- int i, j;
-
- pubsub_admin_service_pt new_psa = (pubsub_admin_service_pt) service;
- logHelper_log(manager->loghelper, OSGI_LOGSERVICE_INFO, "PSTM: Added PSA");
-
- // Add already detected subscriptions to new PSA
- celixThreadMutex_lock(&manager->subscriptionsLock);
- hash_map_iterator_pt subscriptionsIterator = hashMapIterator_create(manager->subscriptions);
-
- while (hashMapIterator_hasNext(subscriptionsIterator)) {
- array_list_pt sub_ep_list = hashMapIterator_nextValue(subscriptionsIterator);
- for(i=0;i<arrayList_size(sub_ep_list);i++){
- pubsub_endpoint_pt sub = (pubsub_endpoint_pt)arrayList_get(sub_ep_list,i);
- double new_psa_score;
- new_psa->matchSubscriber(new_psa->admin, sub, &new_psa_score);
- pubsub_admin_service_pt best_psa = NULL;
- double highest_score = 0;
-
- for(j=0;j<arrayList_size(manager->psaList);j++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,j);
- double score;
- psa->matchSubscriber(psa->admin, sub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL && (new_psa_score > highest_score)){
- best_psa->removeSubscription(best_psa->admin, sub);
- }
- if (new_psa_score > highest_score){
- status += new_psa->addSubscription(new_psa->admin, sub);
- }
- }
- }
-
- hashMapIterator_destroy(subscriptionsIterator);
-
- celixThreadMutex_unlock(&manager->subscriptionsLock);
-
- // Add already detected publications to new PSA
- status = celixThreadMutex_lock(&manager->publicationsLock);
- hash_map_iterator_pt publicationsIterator = hashMapIterator_create(manager->publications);
-
- while (hashMapIterator_hasNext(publicationsIterator)) {
- array_list_pt pub_ep_list = hashMapIterator_nextValue(publicationsIterator);
- for(i=0;i<arrayList_size(pub_ep_list);i++){
- pubsub_endpoint_pt pub = (pubsub_endpoint_pt)arrayList_get(pub_ep_list,i);
- double new_psa_score;
- new_psa->matchPublisher(new_psa->admin, pub, &new_psa_score);
- pubsub_admin_service_pt best_psa = NULL;
- double highest_score = 0;
-
- for(j=0;j<arrayList_size(manager->psaList);j++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,j);
- double score;
- psa->matchPublisher(psa->admin, pub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL && (new_psa_score > highest_score)){
- best_psa->removePublication(best_psa->admin, pub);
- }
- if (new_psa_score > highest_score){
- status += new_psa->addPublication(new_psa->admin, pub);
- }
- }
- }
-
- hashMapIterator_destroy(publicationsIterator);
-
- celixThreadMutex_unlock(&manager->publicationsLock);
-
- celixThreadMutex_lock(&manager->psaListLock);
- arrayList_add(manager->psaList, new_psa);
- celixThreadMutex_unlock(&manager->psaListLock);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_psaModified(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
-
- // Nop...
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_psaRemoved(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt) service;
-
- celixThreadMutex_lock(&manager->psaListLock);
-
- /* Deactivate all publications */
- celixThreadMutex_lock(&manager->publicationsLock);
-
- hash_map_iterator_pt pubit = hashMapIterator_create(manager->publications);
- while(hashMapIterator_hasNext(pubit)){
- hash_map_entry_pt pub_entry = hashMapIterator_nextEntry(pubit);
- char* scope_topic_key = (char*)hashMapEntry_getKey(pub_entry);
- // Extract scope/topic name from key
- char scope[MAX_SCOPE_LEN];
- char topic[MAX_TOPIC_LEN];
- sscanf(scope_topic_key, "%[^:]:%s", scope, topic );
- array_list_pt pubEP_list = (array_list_pt)hashMapEntry_getValue(pub_entry);
-
- status = psa->closeAllPublications(psa->admin,scope,topic);
-
- if(status==CELIX_SUCCESS){
- celixThreadMutex_lock(&manager->discoveryListLock);
- hash_map_iterator_pt iter = hashMapIterator_create(manager->discoveryList);
- while(hashMapIterator_hasNext(iter)){
- service_reference_pt disc_sr = (service_reference_pt)hashMapIterator_nextKey(iter);
- publisher_endpoint_announce_pt disc = NULL;
- bundleContext_getService(manager->context, disc_sr, (void**) &disc);
- const char* fwUUID = NULL;
- bundleContext_getProperty(manager->context,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
- int i;
- for(i=0;i<arrayList_size(pubEP_list);i++){
- pubsub_endpoint_pt pubEP = (pubsub_endpoint_pt)arrayList_get(pubEP_list,i);
- if(strcmp(pubEP->frameworkUUID,fwUUID)==0){
- disc->removePublisher(disc->handle,pubEP);
- }
- }
- bundleContext_ungetService(manager->context, disc_sr, NULL);
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->discoveryListLock);
- }
- }
- hashMapIterator_destroy(pubit);
-
- celixThreadMutex_unlock(&manager->publicationsLock);
-
- /* Deactivate all subscriptions */
- celixThreadMutex_lock(&manager->subscriptionsLock);
- hash_map_iterator_pt subit = hashMapIterator_create(manager->subscriptions);
- while(hashMapIterator_hasNext(subit)){
- // TODO do some error checking
- char* scope_topic = (char*)hashMapIterator_nextKey(subit);
- char scope[MAX_TOPIC_LEN];
- char topic[MAX_TOPIC_LEN];
- memset(scope, 0 , MAX_TOPIC_LEN*sizeof(char));
- memset(topic, 0 , MAX_TOPIC_LEN*sizeof(char));
- sscanf(scope_topic, "%[^:]:%s", scope, topic );
- status += psa->closeAllSubscriptions(psa->admin,scope, topic);
- }
- hashMapIterator_destroy(subit);
- celixThreadMutex_unlock(&manager->subscriptionsLock);
-
- arrayList_removeElement(manager->psaList, psa);
-
- celixThreadMutex_unlock(&manager->psaListLock);
-
- logHelper_log(manager->loghelper, OSGI_LOGSERVICE_INFO, "PSTM: Removed PSA");
-
- return status;
-}
-
-
-celix_status_t pubsub_topologyManager_subscriberAdding(void * handle, service_reference_pt reference, void **service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- status = bundleContext_getService(manager->context, reference, service);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_subscriberAdded(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
- //subscriber_service_pt subscriber = (subscriber_service_pt)service;
-
- pubsub_endpoint_pt sub = NULL;
- if(pubsubEndpoint_createFromServiceReference(reference,&sub) == CELIX_SUCCESS){
- celixThreadMutex_lock(&manager->subscriptionsLock);
- char *sub_key = createScopeTopicKey(sub->scope, sub->topic);
-
- array_list_pt sub_list_by_topic = hashMap_get(manager->subscriptions,sub_key);
- if(sub_list_by_topic==NULL){
- arrayList_create(&sub_list_by_topic);
- hashMap_put(manager->subscriptions,strdup(sub_key),sub_list_by_topic);
- }
- free(sub_key);
- arrayList_add(sub_list_by_topic,sub);
-
- celixThreadMutex_unlock(&manager->subscriptionsLock);
-
- int j;
- celixThreadMutex_lock(&manager->psaListLock);
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
-
- for(j=0;j<arrayList_size(manager->psaList);j++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,j);
- double score;
- psa->matchSubscriber(psa->admin, sub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- best_psa->addSubscription(best_psa->admin,sub);
- }
-
- // Inform discoveries for interest in the topic
- celixThreadMutex_lock(&manager->discoveryListLock);
- hash_map_iterator_pt iter = hashMapIterator_create(manager->discoveryList);
- while(hashMapIterator_hasNext(iter)){
- service_reference_pt disc_sr = (service_reference_pt)hashMapIterator_nextKey(iter);
- publisher_endpoint_announce_pt disc = NULL;
- bundleContext_getService(manager->context, disc_sr, (void**) &disc);
- disc->interestedInTopic(disc->handle, sub->scope, sub->topic);
- bundleContext_ungetService(manager->context, disc_sr, NULL);
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->discoveryListLock);
-
- celixThreadMutex_unlock(&manager->psaListLock);
- }
- else{
- status = CELIX_INVALID_BUNDLE_CONTEXT;
- }
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_subscriberModified(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
-
- // Nop...
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_subscriberRemoved(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- pubsub_endpoint_pt subcmp = NULL;
- if(pubsubEndpoint_createFromServiceReference(reference,&subcmp) == CELIX_SUCCESS){
-
- int j,k;
- celixThreadMutex_lock(&manager->subscriptionsLock);
-
- // Inform discoveries that we not interested in the topic any more
- celixThreadMutex_lock(&manager->discoveryListLock);
- hash_map_iterator_pt iter = hashMapIterator_create(manager->discoveryList);
- while(hashMapIterator_hasNext(iter)){
- service_reference_pt disc_sr = (service_reference_pt)hashMapIterator_nextKey(iter);
- publisher_endpoint_announce_pt disc = NULL;
- bundleContext_getService(manager->context, disc_sr, (void**) &disc);
- disc->uninterestedInTopic(disc->handle, subcmp->scope, subcmp->topic);
- bundleContext_ungetService(manager->context, disc_sr, NULL);
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->discoveryListLock);
-
- char *sub_key = createScopeTopicKey(subcmp->scope,subcmp->topic);
- array_list_pt sub_list_by_topic = hashMap_get(manager->subscriptions,sub_key);
- free(sub_key);
- if(sub_list_by_topic!=NULL){
- for(j=0;j<arrayList_size(sub_list_by_topic);j++){
- pubsub_endpoint_pt sub = arrayList_get(sub_list_by_topic,j);
- if(pubsubEndpoint_equals(sub,subcmp)){
- celixThreadMutex_lock(&manager->psaListLock);
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
- for(k=0;k<arrayList_size(manager->psaList);k++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,k);
- double score;
- psa->matchSubscriber(psa->admin, sub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- best_psa->removeSubscription(best_psa->admin,sub);
- }
- celixThreadMutex_unlock(&manager->psaListLock);
-
- }
- arrayList_remove(sub_list_by_topic,j);
-
- /* If it was the last subscriber for this topic, tell PSA to close the ZMQ socket */
- if(arrayList_size(sub_list_by_topic)==0){
- celixThreadMutex_lock(&manager->psaListLock);
- for(k=0;k<arrayList_size(manager->psaList);k++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,k);
- psa->closeAllSubscriptions(psa->admin,sub->scope, sub->topic);
- }
- celixThreadMutex_unlock(&manager->psaListLock);
- }
-
- pubsubEndpoint_destroy(sub);
-
- }
- }
-
- celixThreadMutex_unlock(&manager->subscriptionsLock);
-
- pubsubEndpoint_destroy(subcmp);
-
- }
- else{
- status = CELIX_INVALID_BUNDLE_CONTEXT;
- }
-
- return status;
-
-}
-
-celix_status_t pubsub_topologyManager_pubsubDiscoveryAdding(void* handle, service_reference_pt reference, void** service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- bundleContext_getService(manager->context, reference, service);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_pubsubDiscoveryAdded(void* handle, service_reference_pt reference, void* service) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = (pubsub_topology_manager_pt)handle;
- publisher_endpoint_announce_pt disc = (publisher_endpoint_announce_pt)service;
-
- const char* fwUUID = NULL;
-
- bundleContext_getProperty(manager->context,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
- if(fwUUID==NULL){
- printf("PSD: ERRROR: Cannot retrieve fwUUID.\n");
- return CELIX_INVALID_BUNDLE_CONTEXT;
- }
-
- celixThreadMutex_lock(&manager->publicationsLock);
-
- celixThreadMutex_lock(&manager->discoveryListLock);
- hashMap_put(manager->discoveryList, reference, NULL);
- celixThreadMutex_unlock(&manager->discoveryListLock);
-
- hash_map_iterator_pt iter = hashMapIterator_create(manager->publications);
- while(hashMapIterator_hasNext(iter)){
- array_list_pt pubEP_list = (array_list_pt)hashMapIterator_nextValue(iter);
- for(int i = 0; i < arrayList_size(pubEP_list); i++) {
- pubsub_endpoint_pt pubEP = (pubsub_endpoint_pt)arrayList_get(pubEP_list,i);
- if( (strcmp(pubEP->frameworkUUID,fwUUID)==0) && (pubEP->endpoint!=NULL)){
- status += disc->announcePublisher(disc->handle,pubEP);
- }
- }
- }
- hashMapIterator_destroy(iter);
-
- celixThreadMutex_unlock(&manager->publicationsLock);
-
- celixThreadMutex_lock(&manager->subscriptionsLock);
- iter = hashMapIterator_create(manager->subscriptions);
-
- while(hashMapIterator_hasNext(iter)) {
- array_list_pt l = (array_list_pt)hashMapIterator_nextValue(iter);
- int i;
- for(i=0;i<arrayList_size(l);i++){
- pubsub_endpoint_pt subEp = (pubsub_endpoint_pt)arrayList_get(l,i);
-
- disc->interestedInTopic(disc->handle, subEp->scope, subEp->topic);
- }
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->subscriptionsLock);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_pubsubDiscoveryModified(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
-
- status = pubsub_topologyManager_pubsubDiscoveryRemoved(handle, reference, service);
- if (status == CELIX_SUCCESS) {
- status = pubsub_topologyManager_pubsubDiscoveryAdded(handle, reference, service);
- }
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_pubsubDiscoveryRemoved(void * handle, service_reference_pt reference, void * service) {
- celix_status_t status = CELIX_SUCCESS;
-
- pubsub_topology_manager_pt manager = handle;
-
- celixThreadMutex_lock(&manager->discoveryListLock);
-
-
- if (hashMap_remove(manager->discoveryList, reference)) {
- logHelper_log(manager->loghelper, OSGI_LOGSERVICE_INFO, "EndpointListener Removed");
- }
-
- celixThreadMutex_unlock(&manager->discoveryListLock);
-
- return status;
-}
-
-
-celix_status_t pubsub_topologyManager_publisherTrackerAdded(void *handle, array_list_pt listeners) {
-
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- int l_index;
-
- for (l_index = 0; l_index < arrayList_size(listeners); l_index++) {
-
- listener_hook_info_pt info = arrayList_get(listeners, l_index);
-
- const char* fwUUID=NULL;
- bundleContext_getProperty(info->context,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
-
- char* scope = pubsub_getScopeFromFilter(info->filter);
- char* topic = pubsub_getTopicFromFilter(info->filter);
- if(scope == NULL) {
- scope = strdup(PUBSUB_PUBLISHER_SCOPE_DEFAULT);
- }
-
- //TODO: Can we use a better serviceID??
- bundle_pt bundle = NULL;
- long bundleId = -1;
- bundleContext_getBundle(info->context,&bundle);
- bundle_getBundleId(bundle,&bundleId);
-
- if(fwUUID !=NULL && topic !=NULL){
-
- pubsub_endpoint_pt pub = NULL;
- if(pubsubEndpoint_create(fwUUID, scope, topic,bundleId,NULL,&pub) == CELIX_SUCCESS){
-
- celixThreadMutex_lock(&manager->publicationsLock);
- char *pub_key = createScopeTopicKey(scope, topic);
- array_list_pt pub_list_by_topic = hashMap_get(manager->publications, pub_key);
- if(pub_list_by_topic==NULL){
- arrayList_create(&pub_list_by_topic);
- hashMap_put(manager->publications,strdup(pub_key),pub_list_by_topic);
- }
- free(pub_key);
- arrayList_add(pub_list_by_topic,pub);
-
- celixThreadMutex_unlock(&manager->publicationsLock);
-
- int j;
- celixThreadMutex_lock(&manager->psaListLock);
-
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
-
- for(j=0;j<arrayList_size(manager->psaList);j++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,j);
- double score;
- psa->matchPublisher(psa->admin, pub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- status = best_psa->addPublication(best_psa->admin,pub);
- if(status==CELIX_SUCCESS){
- celixThreadMutex_lock(&manager->discoveryListLock);
- hash_map_iterator_pt iter = hashMapIterator_create(manager->discoveryList);
- while(hashMapIterator_hasNext(iter)){
- service_reference_pt disc_sr = (service_reference_pt)hashMapIterator_nextKey(iter);
- publisher_endpoint_announce_pt disc = NULL;
- bundleContext_getService(manager->context, disc_sr, (void**) &disc);
- disc->announcePublisher(disc->handle,pub);
- bundleContext_ungetService(manager->context, disc_sr, NULL);
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->discoveryListLock);
- }
- }
-
- celixThreadMutex_unlock(&manager->psaListLock);
-
- }
-
- }
- else{
- status=CELIX_INVALID_BUNDLE_CONTEXT;
- }
-
- free(topic);
- free(scope);
- }
-
- return status;
-
-}
-
-
-celix_status_t pubsub_topologyManager_publisherTrackerRemoved(void *handle, array_list_pt listeners) {
- celix_status_t status = CELIX_SUCCESS;
- pubsub_topology_manager_pt manager = handle;
-
- int l_index;
-
- for (l_index = 0; l_index < arrayList_size(listeners); l_index++) {
-
- listener_hook_info_pt info = arrayList_get(listeners, l_index);
-
- char* pub_scope = pubsub_getScopeFromFilter(info->filter);
- char* pub_topic = pubsub_getTopicFromFilter(info->filter);
-
- const char* fwUUID=NULL;
- bundleContext_getProperty(info->context,OSGI_FRAMEWORK_FRAMEWORK_UUID,&fwUUID);
-
- //TODO: Can we use a better serviceID??
- bundle_pt bundle = NULL;
- long bundleId = -1;
- bundleContext_getBundle(info->context,&bundle);
- bundle_getBundleId(bundle,&bundleId);
-
- if(bundle !=NULL && pub_topic !=NULL && bundleId>0){
-
- pubsub_endpoint_pt pubcmp = NULL;
- if(pubsubEndpoint_create(fwUUID, pub_scope, pub_topic,bundleId,NULL,&pubcmp) == CELIX_SUCCESS){
-
- int j,k;
- celixThreadMutex_lock(&manager->psaListLock);
- celixThreadMutex_lock(&manager->publicationsLock);
-
- char *pub_key = createScopeTopicKey(pub_scope, pub_topic);
- array_list_pt pub_list_by_topic = hashMap_get(manager->publications,pub_key);
- if(pub_list_by_topic!=NULL){
- for(j=0;j<arrayList_size(pub_list_by_topic);j++){
- pubsub_endpoint_pt pub = arrayList_get(pub_list_by_topic,j);
- if(pubsubEndpoint_equals(pub,pubcmp)){
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
-
- for(k=0;k<arrayList_size(manager->psaList);k++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,k);
- double score;
- psa->matchPublisher(psa->admin, pub, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- status = best_psa->removePublication(best_psa->admin,pub);
- if(status==CELIX_SUCCESS){
- celixThreadMutex_lock(&manager->discoveryListLock);
- hash_map_iterator_pt iter = hashMapIterator_create(manager->discoveryList);
- while(hashMapIterator_hasNext(iter)){
- service_reference_pt disc_sr = (service_reference_pt)hashMapIterator_nextKey(iter);
- publisher_endpoint_announce_pt disc = NULL;
- bundleContext_getService(manager->context, disc_sr, (void**) &disc);
- disc->removePublisher(disc->handle,pub);
- bundleContext_ungetService(manager->context, disc_sr, NULL);
- }
- hashMapIterator_destroy(iter);
- celixThreadMutex_unlock(&manager->discoveryListLock);
- }
- }
- }
- arrayList_remove(pub_list_by_topic,j);
-
- /* If it was the last publisher for this topic, tell PSA to close the ZMQ socket and then inform the discovery */
- if(arrayList_size(pub_list_by_topic)==0){
- for(k=0;k<arrayList_size(manager->psaList);k++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,k);
- psa->closeAllPublications(psa->admin,pub->scope, pub->topic);
- }
- }
-
- pubsubEndpoint_destroy(pub);
-
- }
- }
-
- celixThreadMutex_unlock(&manager->publicationsLock);
- celixThreadMutex_unlock(&manager->psaListLock);
-
- pubsubEndpoint_destroy(pubcmp);
-
- free(pub_key);
-
- }
-
- }
- else{
- status=CELIX_INVALID_BUNDLE_CONTEXT;
- }
-
- free(pub_scope);
- free(pub_topic);
- }
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_announcePublisher(void *handle, pubsub_endpoint_pt pubEP){
- celix_status_t status = CELIX_SUCCESS;
- printf("PSTM: New publisher discovered for topic %s [fwUUID=%s, ep=%s]\n",pubEP->topic,pubEP->frameworkUUID,pubEP->endpoint);
-
- pubsub_topology_manager_pt manager = handle;
- celixThreadMutex_lock(&manager->publicationsLock);
- celixThreadMutex_lock(&manager->psaListLock);
- int i;
-
- char *pub_key = createScopeTopicKey(pubEP->scope, pubEP->topic);
-
- array_list_pt pub_list_by_topic = hashMap_get(manager->publications,pub_key);
- if(pub_list_by_topic==NULL){
- arrayList_create(&pub_list_by_topic);
- hashMap_put(manager->publications,strdup(pub_key),pub_list_by_topic);
- }
- free(pub_key);
-
- /* Shouldn't be any other duplicate, since it's filtered out by the discovery */
- pubsub_endpoint_pt p = NULL;
- pubsubEndpoint_create(pubEP->frameworkUUID,pubEP->scope,pubEP->topic,pubEP->serviceID,pubEP->endpoint,&p);
- arrayList_add(pub_list_by_topic,p);
-
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
-
- for(i=0;i<arrayList_size(manager->psaList);i++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,i);
- double score;
- psa->matchPublisher(psa->admin, p, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- status += best_psa->addPublication(best_psa->admin,p);
- }
-
- celixThreadMutex_unlock(&manager->psaListLock);
- celixThreadMutex_unlock(&manager->publicationsLock);
-
- return status;
-}
-
-celix_status_t pubsub_topologyManager_removePublisher(void *handle, pubsub_endpoint_pt pubEP){
- celix_status_t status = CELIX_SUCCESS;
- printf("PSTM: Publisher removed for topic %s [fwUUID=%s, ep=%s]\n",pubEP->topic,pubEP->frameworkUUID,pubEP->endpoint);
-
- pubsub_topology_manager_pt manager = handle;
- celixThreadMutex_lock(&manager->psaListLock);
- celixThreadMutex_lock(&manager->publicationsLock);
- int i;
-
- char *pub_key = createScopeTopicKey(pubEP->scope, pubEP->topic);
- array_list_pt pub_list_by_topic = hashMap_get(manager->publications,pub_key);
- if(pub_list_by_topic==NULL){
- printf("PSTM: ERROR: Cannot find topic for known endpoint [%s,%s,%s]. Something is inconsistent.\n",pub_key,pubEP->frameworkUUID,pubEP->endpoint);
- status = CELIX_ILLEGAL_STATE;
- }
- else{
-
- pubsub_endpoint_pt p = NULL;
- bool found = false;
-
- for(i=0;!found && i<arrayList_size(pub_list_by_topic);i++){
- p = (pubsub_endpoint_pt)arrayList_get(pub_list_by_topic,i);
- found = pubsubEndpoint_equals(p,pubEP);
- }
-
- if(found && p !=NULL){
-
- double highest_score = -1;
- pubsub_admin_service_pt best_psa = NULL;
-
- for(i=0;i<arrayList_size(manager->psaList);i++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,i);
- double score;
- psa->matchPublisher(psa->admin, p, &score);
- if (score > highest_score){
- highest_score = score;
- best_psa = psa;
- }
- }
- if (best_psa != NULL){
- status += best_psa->removePublication(best_psa->admin,p);
- }
-
- arrayList_removeElement(pub_list_by_topic,p);
-
- /* If it was the last publisher for this topic, tell PSA to close the ZMQ socket */
- if(arrayList_size(pub_list_by_topic)==0){
-
- for(i=0;i<arrayList_size(manager->psaList);i++){
- pubsub_admin_service_pt psa = (pubsub_admin_service_pt)arrayList_get(manager->psaList,i);
- status += psa->closeAllPublications(psa->admin,p->scope, p->topic);
- }
- }
-
- pubsubEndpoint_destroy(p);
- }
-
-
- }
- free(pub_key);
- celixThreadMutex_unlock(&manager->publicationsLock);
- celixThreadMutex_unlock(&manager->psaListLock);
-
-
- return status;
-}
-
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/cmake/FindCZMQ.cmake
----------------------------------------------------------------------
diff --git a/cmake/FindCZMQ.cmake b/cmake/FindCZMQ.cmake
new file mode 100644
index 0000000..4f4891c
--- /dev/null
+++ b/cmake/FindCZMQ.cmake
@@ -0,0 +1,42 @@
+# 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.
+
+
+# - Try to find CZMQ
+# Once done this will define
+# CZMQ_FOUND - System has Zmq
+# CZMQ_INCLUDE_DIRS - The Zmq include directories
+# CZMQ_LIBRARIES - The libraries needed to use Zmq
+# CZMQ_DEFINITIONS - Compiler switches required for using Zmq
+
+find_path(CZMQ_INCLUDE_DIR czmq.h
+ /usr/include
+ /usr/local/include )
+
+find_library(CZMQ_LIBRARY NAMES czmq
+ PATHS /usr/lib /usr/local/lib /usr/lib64 /usr/local/lib64 )
+
+set(CZMQ_LIBRARIES ${CZMQ_LIBRARY} )
+set(CZMQ_INCLUDE_DIRS ${CZMQ_INCLUDE_DIR} )
+
+include(FindPackageHandleStandardArgs)
+# handle the QUIETLY and REQUIRED arguments and set CZMQ_FOUND to TRUE
+# if all listed variables are TRUE
+find_package_handle_standard_args(Czmq DEFAULT_MSG
+ CZMQ_LIBRARY CZMQ_INCLUDE_DIR)
+
+mark_as_advanced(CZMQ_INCLUDE_DIR CZMQ_LIBRARY )
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/cmake/FindZMQ.cmake
----------------------------------------------------------------------
diff --git a/cmake/FindZMQ.cmake b/cmake/FindZMQ.cmake
new file mode 100644
index 0000000..b2c2663
--- /dev/null
+++ b/cmake/FindZMQ.cmake
@@ -0,0 +1,42 @@
+# 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.
+
+
+# - Try to find ZMQ
+# Once done this will define
+# ZMQ_FOUND - System has Zmq
+# ZMQ_INCLUDE_DIRS - The Zmq include directories
+# ZMQ_LIBRARIES - The libraries needed to use Zmq
+# ZMQ_DEFINITIONS - Compiler switches required for using Zmq
+
+find_path(ZMQ_INCLUDE_DIR zmq.h zmq_utils.h
+ /usr/include
+ /usr/local/include )
+
+find_library(ZMQ_LIBRARY NAMES zmq
+ PATHS /usr/lib /usr/local/lib /usr/lib64 /usr/local/lib64 )
+
+set(ZMQ_LIBRARIES ${ZMQ_LIBRARY} )
+set(ZMQ_INCLUDE_DIRS ${ZMQ_INCLUDE_DIR} )
+
+include(FindPackageHandleStandardArgs)
+# handle the QUIETLY and REQUIRED arguments and set ZMQ_FOUND to TRUE
+# if all listed variables are TRUE
+find_package_handle_standard_args(Zmq DEFAULT_MSG
+ ZMQ_LIBRARY ZMQ_INCLUDE_DIR)
+
+mark_as_advanced(ZMQ_INCLUDE_DIR ZMQ_LIBRARY )
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/CMakeLists.txt b/pubsub/CMakeLists.txt
new file mode 100644
index 0000000..6b33a57
--- /dev/null
+++ b/pubsub/CMakeLists.txt
@@ -0,0 +1,52 @@
+# 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.
+
+celix_subproject(PUBSUB "Option to build the pubsub bundles" OFF DEPS UTILS)
+if (PUBSUB)
+
+ option(BUILD_PUBSUB_PSA_ZMQ "Build ZeroMQ PubSub Admin (LGPL License)" OFF)
+ if (BUILD_PUBSUB_PSA_ZMQ)
+ message(WARNING "Celix will now contain a dependency with a LGPL License (ZeroMQ). For more information about this, consult the pubsub/README.md file.")
+ option(BUILD_ZMQ_SECURITY "Build with security for ZeroMQ." OFF)
+ endif (BUILD_PUBSUB_PSA_ZMQ)
+
+ include_directories("${PROJECT_SOURCE_DIR}/utils/public/include")
+ include_directories("${PROJECT_SOURCE_DIR}/framework/public/include")
+
+ set (PUBSUB_SERIALIZER_SRC "${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/src/pubsub_serializer.c")
+ set (SERIALIZER_PATH "" CACHE FILEPATH "Path to the directory which will contain the serializer (include / src).")
+ set (SERIALIZER_LIB_INCLUDE_DIR "" CACHE FILEPATH "Path to the include dir of the addiotional libs.")
+ set (SERIALIZER_LIB_PATH "" CACHE FILEPATH "Path to the additional library.")
+ if (EXISTS ${SERIALIZER_PATH})
+ file (GLOB PUBSUB_SERIALIZER_SRC ${SERIALIZER_PATH}/src/*.c)
+
+ if (SERIALIZER_LIB_PATH)
+ get_filename_component(SERIALIZER_LIB_DIR ${SERIALIZER_LIB_PATH} DIRECTORY)
+ get_filename_component(SERIALIZER_LIB_FULLNAME ${SERIALIZER_LIB_PATH} NAME_WE)
+ string (REPLACE "lib" "" SERIALIZER_LIBRARY ${SERIALIZER_LIB_FULLNAME})
+ endif()
+ endif()
+
+ add_subdirectory(pubsub_topology_manager)
+ add_subdirectory(pubsub_discovery)
+ add_subdirectory(pubsub_admin_zmq)
+ add_subdirectory(pubsub_admin_udp_mc)
+ add_subdirectory(examples)
+ add_subdirectory(deploy)
+ add_subdirectory(keygen)
+
+endif(PUBSUB)
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/README.md
----------------------------------------------------------------------
diff --git a/pubsub/README.md b/pubsub/README.md
new file mode 100644
index 0000000..1eaa49c
--- /dev/null
+++ b/pubsub/README.md
@@ -0,0 +1,68 @@
+# Publisher / subscriber implementation
+
+This subdirectory contains an implementation for a publish-subscribe remote services system, that use dfi library for message serialization.
+For low-level communication, UDP and ZMQ is used.
+
+# Description
+
+This publisher / subscriber implementation is based on the concepts of the remote service admin (i.e. rsa / topology / discovery pattern).
+
+Publishers are senders of data, subscribers can receive data. Publishers can publish/send data to certain channels (called 'topics' further on), subscribers can subscribe to these topics. For every topic a publisher service is created by the pubsub admin. This publisher is announced through etcd. So etcd is used for discovery of the publishers. Subscribers are also registered as a service by the pubsub admin and will watch etcd for changes and when a new publisher is announced, the subscriber will check if the topic matches its interests. If the subscriber is interested in/subscribed to a certain topic, a connection between publisher and subscriber will be instantiated by the pubsub admin.
+
+The dfi library is used for message serialization. The publisher / subscriber implementation will arrange that every message which will be send gets an unique id.
+
+For communication between publishers and subscribers UDP and ZeroMQ can be used. When using ZeroMQ it's also possible to setup a secure connection to encrypt the traffic being send between publishers and subscribers. This connection can be secured with ZeroMQ by using a curve25519 key pair per topic.
+
+The publisher/subscriber implementation supports sending of a single message and sending of multipart messages.
+
+## Getting started
+
+The publisher/subscriber implementation contains 2 different PubSubAdmins for managing connections:
+ * PubsubAdminUDP: This pubsub admin is using linux sockets to setup a connection.
+ * PubsubAdminZMQ (LGPL License): This pubsub admin is using ZeroMQ and is disabled as default. This is a because the pubsub admin is using ZeroMQ which is licensed as LGPL ([View ZeroMQ License](https://github.com/zeromq/libzmq#license)).
+
+ The ZeroMQ pubsub admin can be enabled by specifying the build flag `BUILD_PUBSUB_PSA_ZMQ=ON`. To get the ZeroMQ pubsub admin running, [ZeroMQ](https://github.com/zeromq/libzmq) and [CZMQ](https://github.com/zeromq/czmq) need to be installed. Also, to make use of encrypted traffic, [OpenSSL](https://github.com/openssl/openssl) is required.
+
+## Running instructions
+
+### Running PSA UDP-Multicast
+
+1. Open a terminal
+1. Run `etcd`
+1. Open second terminal on project build location
+1. Run `cd deploy/pubsub/pubsub_publisher_udp_mc`
+1. Run `sh run.sh`
+1. Open third terminal on project build location
+1. Run `cd deploy/pubsub/pubsub_subscriber_udp_mc`
+1. Run `sh run.sh`
+
+Design information can be found at pubsub\_admin\_udp\_mc/README.md
+
+### Running PSA ZMQ
+
+For ZeroMQ without encryption, skip the steps 1-12 below
+
+1. Run `touch ~/pubsub.keys`
+1. Run `echo "aes_key:{AES_KEY here}" >> ~/pubsub.keys`
+1. Run `echo "aes_iv:{AES_IV here}" >> ~/pubsub.keys`
+1. Run `touch ~/pubsub.conf`
+1. Run `echo "keys.file.path=$HOME" >> ~/pubsub.conf`
+1. Run `echo "keys.file.name=pubsub.keys" >> ~/pubsub.conf`
+1. To generate ZMQ keypairs
+1. Run `pubsub/keygen/makecert cert_topic1.pub cert_topic1.key`
+1. To encrypt files
+1. Run `pubsub/keygen/ed_file ~/pubsub.keys cert_topic1.key cert_topic1.key.enc`
+1. Store the keys in the pubsub/examples/keys/ directory
+1. Build project to include these keys
+
+For ZeroMQ without encryption, start here
+
+1. Run `etcd`
+1. Open second terminal on pubsub root
+1. Run `cd deploy/pubsub/pubsub_publisher_zmq`
+1. Run `cat ~/pubsub.conf >> config.properties` (only for ZeroMQ with encryption)
+1. Run `sh run.sh`
+1. Open third terminal on pubsub root
+1. Run `cd deploy/pubsub/pubsub_subscriber_zmq`
+1. Run `cat ~/pubsub.conf >> config.properties` (only for ZeroMQ with encryption)
+1. Run `sh run.sh`
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/api/pubsub/publisher.h
----------------------------------------------------------------------
diff --git a/pubsub/api/pubsub/publisher.h b/pubsub/api/pubsub/publisher.h
new file mode 100644
index 0000000..58ac589
--- /dev/null
+++ b/pubsub/api/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 "1.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, 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, 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/2d0923ea/pubsub/api/pubsub/subscriber.h
----------------------------------------------------------------------
diff --git a/pubsub/api/pubsub/subscriber.h b/pubsub/api/pubsub/subscriber.h
new file mode 100644
index 0000000..cbbe96c
--- /dev/null
+++ b/pubsub/api/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 "1.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/2d0923ea/pubsub/deploy/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/deploy/CMakeLists.txt b/pubsub/deploy/CMakeLists.txt
new file mode 100644
index 0000000..3365166
--- /dev/null
+++ b/pubsub/deploy/CMakeLists.txt
@@ -0,0 +1,160 @@
+# 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.
+
+
+# UDP Multicast
+add_deploy("pubsub_publisher_udp_mc"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminUdpMc
+ org.apache.celix.pubsub_publisher.PoiPublisher
+ org.apache.celix.pubsub_publisher.PoiPublisher2
+)
+
+add_deploy("pubsub_subscriber_udp_mc"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminUdpMc
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+)
+
+add_deploy("pubsub_subscriber2_udp_mc"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminUdpMc
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+)
+
+if (BUILD_PUBSUB_PSA_ZMQ)
+
+ # Dynamic ZMQ / UDP admin
+ add_deploy("pubsub_publisher"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_admin.PubSubAdminUdpMc
+ org.apache.celix.pubsub_publisher.PoiPublisher
+ org.apache.celix.pubsub_publisher.PoiPublisher2
+ PROPERTIES
+ poi1.psa=zmq
+ poi2.psa=udp
+ )
+
+ add_deploy("pubsub_subscriber"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_admin.PubSubAdminUdpMc
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+ PROPERTIES
+ poi1.psa=zmq
+ poi2.psa=udp
+ )
+
+ # ZMQ
+ add_deploy("pubsub_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_publisher.PoiPublisher
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+ )
+
+ add_deploy("pubsub_publisher_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_publisher.PoiPublisher
+ org.apache.celix.pubsub_publisher.PoiPublisher2
+ PROPERTIES
+ pubsub.scope=my_small_scope
+ )
+
+ add_deploy("pubsub_subscriber_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+ )
+
+ add_deploy("pubsub_subscriber2_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_subscriber.PoiSubscriber
+ )
+
+ # ZMQ Multipart
+ add_deploy("pubsub_mp_subscriber_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_subscriber.MpSubscriber
+ )
+
+ add_deploy("pubsub_mp_publisher_zmq"
+ GROUP "pubsub"
+ BUNDLES
+ shell
+ shell_tui
+ org.apache.celix.pubsub_discovery.etcd.PubsubDiscovery
+ org.apache.celix.pubsub_topology_manager.PubSubTopologyManager
+ org.apache.celix.pubsub_admin.PubSubAdminZmq
+ org.apache.celix.pubsub_publisher.MpPublisher
+ )
+
+endif()
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/examples/CMakeLists.txt b/pubsub/examples/CMakeLists.txt
new file mode 100644
index 0000000..22ca5ca
--- /dev/null
+++ b/pubsub/examples/CMakeLists.txt
@@ -0,0 +1,19 @@
+# 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.
+
+add_subdirectory(pubsub)
+add_subdirectory(mp_pubsub)
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/keys/README.md
----------------------------------------------------------------------
diff --git a/pubsub/examples/keys/README.md b/pubsub/examples/keys/README.md
new file mode 100644
index 0000000..8517415
--- /dev/null
+++ b/pubsub/examples/keys/README.md
@@ -0,0 +1,19 @@
+
+Store the AES key for encrypting and decrypting the encoded secret keys safe in a file!
+Default file is `/etc/pubsub.keys` with the following format:
+```
+aes_key:{32 character AES key here}
+aes_iv:{16 character AES iv here}
+```
+
+Use the $PROJECT_BUILD/pubsub/keygen/makecert for generating keypairs
+Use the $PROJECT_BUILD/pubsub/keygen/ed_file for encrypting and decrypting private keys
+
+Public keys need to be stored in the 'public' folder having the following format:
+- pub_{topic}.pub
+- sub_{topic}.pub
+
+Secret keys need to be stored in the 'private' folder having the following format:
+- pub_{topic}.key.enc
+- sub_{topic}.key.enc
+These files need to be encrypted using the 'ed_file' executable.
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/keys/publisher/private/.gitkeep
----------------------------------------------------------------------
diff --git a/pubsub/examples/keys/publisher/private/.gitkeep b/pubsub/examples/keys/publisher/private/.gitkeep
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/keys/publisher/public/.gitkeep
----------------------------------------------------------------------
diff --git a/pubsub/examples/keys/publisher/public/.gitkeep b/pubsub/examples/keys/publisher/public/.gitkeep
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/keys/subscriber/private/.gitkeep
----------------------------------------------------------------------
diff --git a/pubsub/examples/keys/subscriber/private/.gitkeep b/pubsub/examples/keys/subscriber/private/.gitkeep
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/keys/subscriber/public/.gitkeep
----------------------------------------------------------------------
diff --git a/pubsub/examples/keys/subscriber/public/.gitkeep b/pubsub/examples/keys/subscriber/public/.gitkeep
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/CMakeLists.txt b/pubsub/examples/mp_pubsub/CMakeLists.txt
new file mode 100644
index 0000000..c828832
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/CMakeLists.txt
@@ -0,0 +1,23 @@
+# 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_directories("common/include")
+
+add_subdirectory(publisher)
+add_subdirectory(subscriber)
+
+
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/common/include/ew.h
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/common/include/ew.h b/pubsub/examples/mp_pubsub/common/include/ew.h
new file mode 100644
index 0000000..81ca5f3
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/common/include/ew.h
@@ -0,0 +1,53 @@
+/**
+ *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.
+ */
+/*
+ * ew.h
+ *
+ * \date Jan 15, 2016
+ * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ * \copyright Apache License, Version 2.0
+ */
+
+#ifndef EW_H_
+#define EW_H_
+
+#define MIN_AREA 50.0F
+#define MAX_AREA 15000.0F
+
+#define MSG_EW_NAME "ew" //Has to match the message name in the msg descriptor!
+
+typedef enum color{
+ GREEN,
+ BLUE,
+ RED,
+ BLACK,
+ WHITE,
+ LAST_COLOR
+} color_e;
+
+const char* color_tostring[] = {"GREEN","BLUE","RED","BLACK","WHITE"};
+
+struct ew_data{
+ double area;
+ color_e color;
+};
+
+typedef struct ew_data* ew_data_pt;
+
+#endif /* EW_H_ */
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/common/include/ide.h
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/common/include/ide.h b/pubsub/examples/mp_pubsub/common/include/ide.h
new file mode 100644
index 0000000..2b9588d
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/common/include/ide.h
@@ -0,0 +1,49 @@
+/**
+ *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.
+ */
+/*
+ * ide.h
+ *
+ * \date Jan 15, 2016
+ * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ * \copyright Apache License, Version 2.0
+ */
+
+#ifndef IDE_H_
+#define IDE_H_
+
+#define MSG_IDE_NAME "ide" //Has to match the message name in the msg descriptor!
+
+typedef enum shape{
+ SQUARE,
+ CIRCLE,
+ TRIANGLE,
+ RECTANGLE,
+ HEXAGON,
+ LAST_SHAPE
+} shape_e;
+
+const char* shape_tostring[] = {"SQUARE","CIRCLE","TRIANGLE","RECTANGLE","HEXAGON"};
+
+struct ide{
+ shape_e shape;
+};
+
+typedef struct ide* ide_pt;
+
+#endif /* IDE_H_ */
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/common/include/kinematics.h
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/common/include/kinematics.h b/pubsub/examples/mp_pubsub/common/include/kinematics.h
new file mode 100644
index 0000000..5601509
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/common/include/kinematics.h
@@ -0,0 +1,55 @@
+/**
+ *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.
+ */
+/*
+ * kinematics.h
+ *
+ * \date Nov 12, 2015
+ * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ * \copyright Apache License, Version 2.0
+ */
+
+#ifndef KINEMATICS_H_
+#define KINEMATICS_H_
+
+#define MIN_LAT -90.0F
+#define MAX_LAT 90.0F
+#define MIN_LON -180.0F
+#define MAX_LON 180.0F
+
+#define MIN_OCCUR 1
+#define MAX_OCCUR 5
+
+#define MSG_KINEMATICS_NAME "kinematics" //Has to match the message name in the msg descriptor!
+
+struct position{
+ double lat;
+ double lon;
+};
+
+typedef struct position position_t;
+
+struct kinematics{
+ position_t position;
+ int occurrences;
+};
+
+typedef struct kinematics* kinematics_pt;
+
+
+#endif /* KINEMATICS_H_ */
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/msg_descriptors/msg_ew.descriptor
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/msg_descriptors/msg_ew.descriptor b/pubsub/examples/mp_pubsub/msg_descriptors/msg_ew.descriptor
new file mode 100644
index 0000000..7eb8c29
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/msg_descriptors/msg_ew.descriptor
@@ -0,0 +1,9 @@
+:header
+type=message
+name=ew
+version=1.0.0
+:annotations
+classname=org.example.Ew
+:types
+:message
+{Di area color}
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/msg_descriptors/msg_ide.descriptor
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/msg_descriptors/msg_ide.descriptor b/pubsub/examples/mp_pubsub/msg_descriptors/msg_ide.descriptor
new file mode 100644
index 0000000..f26286d
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/msg_descriptors/msg_ide.descriptor
@@ -0,0 +1,9 @@
+:header
+type=message
+name=ide
+version=1.0.0
+:annotations
+classname=org.example.Ide
+:types
+:message
+{i shape}
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/msg_descriptors/msg_kinematics.descriptor
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/msg_descriptors/msg_kinematics.descriptor b/pubsub/examples/mp_pubsub/msg_descriptors/msg_kinematics.descriptor
new file mode 100644
index 0000000..447b645
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/msg_descriptors/msg_kinematics.descriptor
@@ -0,0 +1,10 @@
+:header
+type=message
+name=kinematics
+version=1.0.0
+:annotations
+classname=org.example.Kinematics
+:types
+position={DD lat long}
+:message
+{lposition;N position occurrences}
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/publisher/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/publisher/CMakeLists.txt b/pubsub/examples/mp_pubsub/publisher/CMakeLists.txt
new file mode 100644
index 0000000..76a01f1
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/publisher/CMakeLists.txt
@@ -0,0 +1,50 @@
+# 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_directories("private/include")
+include_directories("${PROJECT_SOURCE_DIR}/framework/public/include")
+include_directories("${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/include")
+include_directories("${PROJECT_SOURCE_DIR}/pubsub/api/pubsub")
+include_directories("${PROJECT_SOURCE_DIR}/pubsub/examples/mp_pubsub/common/include")
+
+add_bundle(org.apache.celix.pubsub_publisher.MpPublisher
+ SYMBOLIC_NAME "apache_celix_pubsub_mp_publisher"
+ VERSION "1.0.0"
+ SOURCES
+ private/src/mp_pub_activator.c
+ private/src/mp_publisher.c
+ ${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/src/pubsub_utils.c
+)
+
+bundle_files(org.apache.celix.pubsub_publisher.MpPublisher
+ ${PROJECT_SOURCE_DIR}/pubsub/examples/mp_pubsub/msg_descriptors/msg_ew.descriptor
+ ${PROJECT_SOURCE_DIR}/pubsub/examples/mp_pubsub/msg_descriptors/msg_ide.descriptor
+ ${PROJECT_SOURCE_DIR}/pubsub/examples/mp_pubsub/msg_descriptors/msg_kinematics.descriptor
+ DESTINATION "META-INF/descriptors"
+)
+
+bundle_files(org.apache.celix.pubsub_publisher.MpPublisher
+ ${PROJECT_SOURCE_DIR}/pubsub/examples/keys/publisher
+ DESTINATION "META-INF/keys"
+)
+
+bundle_files(org.apache.celix.pubsub_publisher.MpPublisher
+ ${PROJECT_SOURCE_DIR}/pubsub/examples/keys/subscriber/public
+ DESTINATION "META-INF/keys/subscriber"
+)
+
+target_link_libraries(org.apache.celix.pubsub_publisher.MpPublisher celix_framework celix_utils)
http://git-wip-us.apache.org/repos/asf/celix/blob/2d0923ea/pubsub/examples/mp_pubsub/publisher/private/include/mp_publisher_private.h
----------------------------------------------------------------------
diff --git a/pubsub/examples/mp_pubsub/publisher/private/include/mp_publisher_private.h b/pubsub/examples/mp_pubsub/publisher/private/include/mp_publisher_private.h
new file mode 100644
index 0000000..a9c070f
--- /dev/null
+++ b/pubsub/examples/mp_pubsub/publisher/private/include/mp_publisher_private.h
@@ -0,0 +1,58 @@
+/**
+ *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.
+ */
+/*
+ * mp_publisher_private.h
+ *
+ * \date Sep 21, 2010
+ * \author <a href="mailto:dev@celix.apache.org">Apache Celix Project Team</a>
+ * \copyright Apache License, Version 2.0
+ */
+
+#ifndef MP_PUBLISHER_PRIVATE_H_
+#define MP_PUBLISHER_PRIVATE_H_
+
+#include <celixbool.h>
+
+#include "publisher.h"
+
+struct pubsub_sender {
+ array_list_pt trackers;
+ const char *ident;
+ long bundleId;
+};
+
+typedef struct pubsub_sender * pubsub_sender_pt;
+
+typedef struct send_thread_struct{
+ pubsub_publisher_pt service;
+ pubsub_sender_pt publisher;
+} *send_thread_struct_pt;
+
+pubsub_sender_pt publisher_create(array_list_pt trackers,const char* ident,long bundleId);
+
+void publisher_start(pubsub_sender_pt client);
+void publisher_stop(pubsub_sender_pt client);
+
+void publisher_destroy(pubsub_sender_pt client);
+
+celix_status_t publisher_publishSvcAdded(void * handle, service_reference_pt reference, void * service);
+celix_status_t publisher_publishSvcRemoved(void * handle, service_reference_pt reference, void * service);
+
+
+#endif /* MP_PUBLISHER_PRIVATE_H_ */