http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c b/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c new file mode 100644 index 0000000..94a8e11 --- /dev/null +++ b/pubsub/pubsub_discovery/src/pubsub_discovery_impl.c @@ -0,0 +1,457 @@ +/** + *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 <stdio.h> +#include <string.h> +#include <stdlib.h> +#include <unistd.h> +#include <stdbool.h> +#include <netdb.h> +#include <netinet/in.h> + +#include "constants.h" +#include "celix_threads.h" +#include "bundle_context.h" +#include "array_list.h" +#include "utils.h" +#include "celix_errno.h" +#include "filter.h" +#include "service_reference.h" +#include "service_registration.h" + +#include "publisher_endpoint_announce.h" +#include "etcd_common.h" +#include "etcd_watcher.h" +#include "etcd_writer.h" +#include "pubsub_endpoint.h" +#include "pubsub_discovery_impl.h" + +/* Discovery activator functions */ +celix_status_t pubsub_discovery_create(bundle_context_pt context, pubsub_discovery_pt *ps_discovery) { + celix_status_t status = CELIX_SUCCESS; + + *ps_discovery = calloc(1, sizeof(**ps_discovery)); + + if (*ps_discovery == NULL) { + status = CELIX_ENOMEM; + } + else{ + (*ps_discovery)->context = context; + (*ps_discovery)->discoveredPubs = hashMap_create(utils_stringHash, NULL, utils_stringEquals, NULL); + (*ps_discovery)->listenerReferences = hashMap_create(serviceReference_hashCode, NULL, serviceReference_equals2, NULL); + (*ps_discovery)->watchers = hashMap_create(utils_stringHash,NULL,utils_stringEquals, NULL); + celixThreadMutex_create(&(*ps_discovery)->listenerReferencesMutex, NULL); + celixThreadMutex_create(&(*ps_discovery)->discoveredPubsMutex, NULL); + celixThreadMutex_create(&(*ps_discovery)->watchersMutex, NULL); + } + + return status; +} + +celix_status_t pubsub_discovery_destroy(pubsub_discovery_pt ps_discovery) { + celix_status_t status = CELIX_SUCCESS; + + celixThreadMutex_lock(&ps_discovery->discoveredPubsMutex); + + hash_map_iterator_pt iter = hashMapIterator_create(ps_discovery->discoveredPubs); + + 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++) { + pubsubEndpoint_destroy(((pubsub_endpoint_pt)arrayList_get(pubEP_list,i))); + } + arrayList_destroy(pubEP_list); + } + + hashMapIterator_destroy(iter); + + hashMap_destroy(ps_discovery->discoveredPubs, true, false); + ps_discovery->discoveredPubs = NULL; + + celixThreadMutex_unlock(&ps_discovery->discoveredPubsMutex); + + celixThreadMutex_destroy(&ps_discovery->discoveredPubsMutex); + + + celixThreadMutex_lock(&ps_discovery->listenerReferencesMutex); + + hashMap_destroy(ps_discovery->listenerReferences, false, false); + ps_discovery->listenerReferences = NULL; + + celixThreadMutex_unlock(&ps_discovery->listenerReferencesMutex); + + celixThreadMutex_destroy(&ps_discovery->listenerReferencesMutex); + + free(ps_discovery); + + return status; +} + +celix_status_t pubsub_discovery_start(pubsub_discovery_pt ps_discovery) { + celix_status_t status = CELIX_SUCCESS; + status = etcdCommon_init(ps_discovery->context); + ps_discovery->writer = etcdWriter_create(ps_discovery); + + return status; +} + +celix_status_t pubsub_discovery_stop(pubsub_discovery_pt ps_discovery) { + celix_status_t status = CELIX_SUCCESS; + + const char* fwUUID = NULL; + + bundleContext_getProperty(ps_discovery->context, OSGI_FRAMEWORK_FRAMEWORK_UUID, &fwUUID); + if (fwUUID == NULL) { + printf("PSD: Cannot retrieve fwUUID.\n"); + return CELIX_INVALID_BUNDLE_CONTEXT; + } + + celixThreadMutex_lock(&ps_discovery->watchersMutex); + + hash_map_iterator_pt iter = hashMapIterator_create(ps_discovery->watchers); + while (hashMapIterator_hasNext(iter)) { + struct watcher_info * wi = hashMapIterator_nextValue(iter); + etcdWatcher_stop(wi->watcher); + } + hashMapIterator_destroy(iter); + + celixThreadMutex_lock(&ps_discovery->discoveredPubsMutex); + + /* Unexport all publishers for the local framework, and also delete from ETCD publisher belonging to the local framework */ + + iter = hashMapIterator_create(ps_discovery->discoveredPubs); + while (hashMapIterator_hasNext(iter)) { + array_list_pt pubEP_list = (array_list_pt) hashMapIterator_nextValue(iter); + + 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) { + etcdWriter_deletePublisherEndpoint(ps_discovery->writer, pubEP); + } else { + pubsub_discovery_informPublishersListeners(ps_discovery, pubEP, false); + arrayList_remove(pubEP_list, i); + pubsubEndpoint_destroy(pubEP); + i--; + } + } + } + + hashMapIterator_destroy(iter); + + celixThreadMutex_unlock(&ps_discovery->discoveredPubsMutex); + etcdWriter_destroy(ps_discovery->writer); + + iter = hashMapIterator_create(ps_discovery->watchers); + while (hashMapIterator_hasNext(iter)) { + struct watcher_info * wi = hashMapIterator_nextValue(iter); + etcdWatcher_destroy(wi->watcher); + } + hashMapIterator_destroy(iter); + hashMap_destroy(ps_discovery->watchers, true, true); + celixThreadMutex_unlock(&ps_discovery->watchersMutex); + return status; +} + +/* Functions called by the etcd_watcher */ + +celix_status_t pubsub_discovery_addNode(pubsub_discovery_pt pubsub_discovery, pubsub_endpoint_pt pubEP) { + celix_status_t status = CELIX_SUCCESS; + bool inform=false; + celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex); + + char *pubs_key = createScopeTopicKey(pubEP->scope, pubEP->topic); + array_list_pt pubEP_list = (array_list_pt)hashMap_get(pubsub_discovery->discoveredPubs,pubs_key); + if(pubEP_list==NULL){ + arrayList_create(&pubEP_list); + arrayList_add(pubEP_list,pubEP); + hashMap_put(pubsub_discovery->discoveredPubs,strdup(pubs_key),pubEP_list); + inform=true; + } + else{ + int i; + bool found = false; + for(i=0;i<arrayList_size(pubEP_list) && !found;i++){ + found = pubsubEndpoint_equals(pubEP,(pubsub_endpoint_pt)arrayList_get(pubEP_list,i)); + } + if(found){ + pubsubEndpoint_destroy(pubEP); + } + else{ + arrayList_add(pubEP_list,pubEP); + inform=true; + } + } + free(pubs_key); + + celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex); + + if(inform){ + status = pubsub_discovery_informPublishersListeners(pubsub_discovery,pubEP,true); + } + + return status; +} + +celix_status_t pubsub_discovery_removeNode(pubsub_discovery_pt pubsub_discovery, pubsub_endpoint_pt pubEP) { + celix_status_t status = CELIX_SUCCESS; + pubsub_endpoint_pt p = NULL; + bool found = false; + + celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex); + char *pubs_key = createScopeTopicKey(pubEP->scope, pubEP->topic); + array_list_pt pubEP_list = (array_list_pt) hashMap_get(pubsub_discovery->discoveredPubs, pubs_key); + free(pubs_key); + if (pubEP_list == NULL) { + printf("PSD: Cannot find any registered publisher for topic %s. Something is not consistent.\n", pubEP->topic); + status = CELIX_ILLEGAL_STATE; + } else { + int i; + + for (i = 0; !found && i < arrayList_size(pubEP_list); i++) { + p = arrayList_get(pubEP_list, i); + found = pubsubEndpoint_equals(pubEP, p); + if (found) { + arrayList_remove(pubEP_list, i); + pubsubEndpoint_destroy(p); + } + } + } + + celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex); + if (found) { + status = pubsub_discovery_informPublishersListeners(pubsub_discovery, pubEP, false); + } + pubsubEndpoint_destroy(pubEP); + + return status; +} + +/* Callback to the pubsub_topology_manager */ +celix_status_t pubsub_discovery_informPublishersListeners(pubsub_discovery_pt pubsub_discovery, pubsub_endpoint_pt pubEP, bool epAdded) { + celix_status_t status = CELIX_SUCCESS; + + // Inform listeners of new publisher endpoint + celixThreadMutex_lock(&pubsub_discovery->listenerReferencesMutex); + + if (pubsub_discovery->listenerReferences != NULL) { + hash_map_iterator_pt iter = hashMapIterator_create(pubsub_discovery->listenerReferences); + while (hashMapIterator_hasNext(iter)) { + service_reference_pt reference = hashMapIterator_nextKey(iter); + + publisher_endpoint_announce_pt listener = NULL; + + bundleContext_getService(pubsub_discovery->context, reference, (void**) &listener); + if (epAdded) { + listener->announcePublisher(listener->handle, pubEP); + } else { + listener->removePublisher(listener->handle, pubEP); + } + bundleContext_ungetService(pubsub_discovery->context, reference, NULL); + } + hashMapIterator_destroy(iter); + } + + celixThreadMutex_unlock(&pubsub_discovery->listenerReferencesMutex); + + return status; +} + + +/* Service's functions implementation */ +celix_status_t pubsub_discovery_announcePublisher(void *handle, pubsub_endpoint_pt pubEP) { + celix_status_t status = CELIX_SUCCESS; + printf("pubsub_discovery_announcePublisher : %s / %s\n", pubEP->topic, pubEP->endpoint); + pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt) handle; + + celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex); + + char *pub_key = createScopeTopicKey(pubEP->scope,pubEP->topic); + array_list_pt pubEP_list = (array_list_pt)hashMap_get(pubsub_discovery->discoveredPubs,pub_key); + + if(pubEP_list==NULL){ + arrayList_create(&pubEP_list); + hashMap_put(pubsub_discovery->discoveredPubs,strdup(pub_key),pubEP_list); + } + free(pub_key); + pubsub_endpoint_pt p = NULL; + pubsubEndpoint_clone(pubEP, &p); + + arrayList_add(pubEP_list,p); + + status = etcdWriter_addPublisherEndpoint(pubsub_discovery->writer,p,true); + + celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex); + + return status; +} + +celix_status_t pubsub_discovery_removePublisher(void *handle, pubsub_endpoint_pt pubEP) { + celix_status_t status = CELIX_SUCCESS; + + pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt) handle; + + celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex); + + char *pub_key = createScopeTopicKey(pubEP->scope,pubEP->topic); + array_list_pt pubEP_list = (array_list_pt)hashMap_get(pubsub_discovery->discoveredPubs,pub_key); + free(pub_key); + if(pubEP_list==NULL){ + printf("PSD: Cannot find any registered publisher for topic %s. Something is not consistent.\n",pubEP->topic); + status = CELIX_ILLEGAL_STATE; + } + else{ + + int i; + bool found = false; + pubsub_endpoint_pt p = NULL; + + for(i=0;!found && i<arrayList_size(pubEP_list);i++){ + p = (pubsub_endpoint_pt)arrayList_get(pubEP_list,i); + found = pubsubEndpoint_equals(pubEP,p); + } + + if(!found){ + printf("PSD: Trying to remove a not existing endpoint. Something is not consistent.\n"); + status = CELIX_ILLEGAL_STATE; + } + else{ + + arrayList_removeElement(pubEP_list,p); + + status = etcdWriter_deletePublisherEndpoint(pubsub_discovery->writer,p); + + pubsubEndpoint_destroy(p); + } + } + + celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex); + + return status; +} + +celix_status_t pubsub_discovery_interestedInTopic(void *handle, const char* scope, const char* topic) { + pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt) handle; + + char *scope_topic_key = createScopeTopicKey(scope, topic); + celixThreadMutex_lock(&pubsub_discovery->watchersMutex); + struct watcher_info * wi = hashMap_get(pubsub_discovery->watchers, scope_topic_key); + if(wi) { + wi->nr_references++; + free(scope_topic_key); + } else { + wi = calloc(1, sizeof(*wi)); + etcdWatcher_create(pubsub_discovery, pubsub_discovery->context, scope, topic, &wi->watcher); + wi->nr_references = 1; + hashMap_put(pubsub_discovery->watchers, scope_topic_key, wi); + } + + celixThreadMutex_unlock(&pubsub_discovery->watchersMutex); + + return CELIX_SUCCESS; +} + +celix_status_t pubsub_discovery_uninterestedInTopic(void *handle, const char* scope, const char* topic) { + pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt) handle; + + char *scope_topic_key = createScopeTopicKey(scope, topic); + celixThreadMutex_lock(&pubsub_discovery->watchersMutex); + + hash_map_entry_pt entry = hashMap_getEntry(pubsub_discovery->watchers, scope_topic_key); + if(entry) { + struct watcher_info * wi = hashMapEntry_getValue(entry); + wi->nr_references--; + if(wi->nr_references == 0) { + char *key = hashMapEntry_getKey(entry); + hashMap_remove(pubsub_discovery->watchers, scope_topic_key); + free(key); + free(scope_topic_key); + etcdWatcher_stop(wi->watcher); + etcdWatcher_destroy(wi->watcher); + free(wi); + } + } else { + fprintf(stderr, "[DISC] Inconsistency error: Removing unknown topic %s\n", topic); + } + celixThreadMutex_unlock(&pubsub_discovery->watchersMutex); + return CELIX_SUCCESS; +} + +/* pubsub_topology_manager tracker callbacks */ + +celix_status_t pubsub_discovery_tmPublisherAnnounceAdded(void * handle, service_reference_pt reference, void * service) { + celix_status_t status = CELIX_SUCCESS; + + pubsub_discovery_pt pubsub_discovery = (pubsub_discovery_pt)handle; + publisher_endpoint_announce_pt listener = (publisher_endpoint_announce_pt)service; + + celixThreadMutex_lock(&pubsub_discovery->discoveredPubsMutex); + celixThreadMutex_lock(&pubsub_discovery->listenerReferencesMutex); + + /* Notify the PSTM about discovered publisher endpoints */ + hash_map_iterator_pt iter = hashMapIterator_create(pubsub_discovery->discoveredPubs); + while(hashMapIterator_hasNext(iter)){ + array_list_pt pubEP_list = (array_list_pt)hashMapIterator_nextValue(iter); + int i; + for(i=0;i<arrayList_size(pubEP_list);i++){ + pubsub_endpoint_pt pubEP = (pubsub_endpoint_pt)arrayList_get(pubEP_list,i); + status += listener->announcePublisher(listener->handle, pubEP); + } + } + + hashMapIterator_destroy(iter); + + hashMap_put(pubsub_discovery->listenerReferences, reference, NULL); + + celixThreadMutex_unlock(&pubsub_discovery->listenerReferencesMutex); + celixThreadMutex_unlock(&pubsub_discovery->discoveredPubsMutex); + + printf("PSD: pubsub_tm_announce_publisher added.\n"); + + return status; +} + +celix_status_t pubsub_discovery_tmPublisherAnnounceModified(void * handle, service_reference_pt reference, void * service) { + celix_status_t status = CELIX_SUCCESS; + + status = pubsub_discovery_tmPublisherAnnounceRemoved(handle, reference, service); + if (status == CELIX_SUCCESS) { + status = pubsub_discovery_tmPublisherAnnounceAdded(handle, reference, service); + } + + return status; +} + +celix_status_t pubsub_discovery_tmPublisherAnnounceRemoved(void * handle, service_reference_pt reference, void * service) { + celix_status_t status = CELIX_SUCCESS; + pubsub_discovery_pt pubsub_discovery = handle; + + celixThreadMutex_lock(&pubsub_discovery->listenerReferencesMutex); + + if (pubsub_discovery->listenerReferences != NULL) { + if (hashMap_remove(pubsub_discovery->listenerReferences, reference)) { + printf("PSD: pubsub_tm_announce_publisher removed.\n"); + } + } + celixThreadMutex_unlock(&pubsub_discovery->listenerReferencesMutex); + + return status; +} +
http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_discovery/src/pubsub_discovery_impl.h ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_discovery/src/pubsub_discovery_impl.h b/pubsub/pubsub_discovery/src/pubsub_discovery_impl.h new file mode 100644 index 0000000..676a6ab --- /dev/null +++ b/pubsub/pubsub_discovery/src/pubsub_discovery_impl.h @@ -0,0 +1,72 @@ +/** + *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_DISCOVERY_IMPL_H_ +#define PUBSUB_DISCOVERY_IMPL_H_ + +#include "bundle_context.h" +#include "service_reference.h" + +#include "etcd_watcher.h" +#include "etcd_writer.h" +#include "pubsub_endpoint.h" + +#define FREE_MEM(ptr) if(ptr) {free(ptr); ptr = NULL;} + +struct watcher_info { + etcd_watcher_pt watcher; + int nr_references; +}; + +struct pubsub_discovery { + bundle_context_pt context; + + celix_thread_mutex_t discoveredPubsMutex; + hash_map_pt discoveredPubs; //<topic,List<pubsub_endpoint_pt>> + + celix_thread_mutex_t listenerReferencesMutex; + hash_map_pt listenerReferences; //key=serviceReference, value=nop + + celix_thread_mutex_t watchersMutex; + hash_map_pt watchers; //key = topicname, value = struct watcher_info + + etcd_writer_pt writer; +}; + + +celix_status_t pubsub_discovery_create(bundle_context_pt context, pubsub_discovery_pt* node_discovery); +celix_status_t pubsub_discovery_destroy(pubsub_discovery_pt node_discovery); +celix_status_t pubsub_discovery_start(pubsub_discovery_pt node_discovery); +celix_status_t pubsub_discovery_stop(pubsub_discovery_pt node_discovery); + +celix_status_t pubsub_discovery_addNode(pubsub_discovery_pt node_discovery, pubsub_endpoint_pt pubEP); +celix_status_t pubsub_discovery_removeNode(pubsub_discovery_pt node_discovery, pubsub_endpoint_pt pubEP); + +celix_status_t pubsub_discovery_tmPublisherAnnounceAdded(void * handle, service_reference_pt reference, void * service); +celix_status_t pubsub_discovery_tmPublisherAnnounceModified(void * handle, service_reference_pt reference, void * service); +celix_status_t pubsub_discovery_tmPublisherAnnounceRemoved(void * handle, service_reference_pt reference, void * service); + +celix_status_t pubsub_discovery_announcePublisher(void *handle, pubsub_endpoint_pt pubEP); +celix_status_t pubsub_discovery_removePublisher(void *handle, pubsub_endpoint_pt pubEP); +celix_status_t pubsub_discovery_interestedInTopic(void *handle, const char* scope, const char* topic); +celix_status_t pubsub_discovery_uninterestedInTopic(void *handle, const char* scope, const char* topic); + +celix_status_t pubsub_discovery_informPublishersListeners(pubsub_discovery_pt discovery, pubsub_endpoint_pt endpoint, bool endpointAdded); + +#endif /* PUBSUB_DISCOVERY_IMPL_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/CMakeLists.txt b/pubsub/pubsub_serializer_json/CMakeLists.txt index 147873a..b86f30e 100644 --- a/pubsub/pubsub_serializer_json/CMakeLists.txt +++ b/pubsub/pubsub_serializer_json/CMakeLists.txt @@ -17,27 +17,23 @@ find_package(Jansson REQUIRED) -include_directories("private/include") -include_directories("public/include") -include_directories("${PROJECT_SOURCE_DIR}/utils/public/include") -include_directories("${PROJECT_SOURCE_DIR}/log_service/public/include") -include_directories("${PROJECT_SOURCE_DIR}/dfi/public/include") -include_directories("${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/include") -include_directories("${PROJECT_SOURCE_DIR}/pubsub/api/pubsub") -include_directories("${JANSSON_INCLUDE_DIR}") add_bundle(org.apache.celix.pubsub_serializer.PubSubSerializerJson BUNDLE_SYMBOLICNAME "apache_celix_pubsub_serializer_json" VERSION "1.0.0" SOURCES - private/src/ps_activator.c - private/src/pubsub_serializer_impl.c - ${PROJECT_SOURCE_DIR}/log_service/public/src/log_helper.c + src/ps_activator.c + src/pubsub_serializer_impl.c ${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/src/pubsub_utils.c ) +target_include_directories(org.apache.celix.pubsub_serializer.PubSubSerializerJson PRIVATE + src + ${JANSSON_INCLUDE_DIR} +) + set_target_properties(org.apache.celix.pubsub_serializer.PubSubSerializerJson PROPERTIES INSTALL_RPATH "$ORIGIN") -target_link_libraries(org.apache.celix.pubsub_serializer.PubSubSerializerJson celix_framework celix_utils celix_dfi ${JANSSON_LIBRARIES}) +target_link_libraries(org.apache.celix.pubsub_serializer.PubSubSerializerJson PRIVATE Celix::framework Celix::dfi ${JANSSON_LIBRARIES} Celix::log_helper) install_bundle(org.apache.celix.pubsub_serializer.PubSubSerializerJson) http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/private/include/pubsub_serializer_impl.h ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/private/include/pubsub_serializer_impl.h b/pubsub/pubsub_serializer_json/private/include/pubsub_serializer_impl.h deleted file mode 100644 index c36f20e..0000000 --- a/pubsub/pubsub_serializer_json/private/include/pubsub_serializer_impl.h +++ /dev/null @@ -1,55 +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_impl.h - * - * \date Mar 24, 2017 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef PUBSUB_SERIALIZER_JSON_H_ -#define PUBSUB_SERIALIZER_JSON_H_ - -#include "dyn_common.h" -#include "dyn_type.h" -#include "dyn_message.h" -#include "log_helper.h" - -#include "pubsub_serializer.h" - -#define PUBSUB_SERIALIZER_TYPE "json" - -typedef struct pubsub_serializer { - bundle_context_pt bundle_context; - log_helper_pt loghelper; -} pubsub_serializer_t; - -celix_status_t pubsubSerializer_create(bundle_context_pt context, pubsub_serializer_t* *serializer); -celix_status_t pubsubSerializer_destroy(pubsub_serializer_t* serializer); - -celix_status_t pubsubSerializer_createSerializerMap(pubsub_serializer_t* serializer, bundle_pt bundle, hash_map_pt* serializerMap); -celix_status_t pubsubSerializer_destroySerializerMap(pubsub_serializer_t*, hash_map_pt serializerMap); - -/* Start of serializer specific functions */ -celix_status_t pubsubMsgSerializer_serialize(pubsub_msg_serializer_t* msgSerializer, const void* msg, void** out, size_t *outLen); -celix_status_t pubsubMsgSerializer_deserialize(pubsub_msg_serializer_t* msgSerializer, const void* input, size_t inputLen, void **out); -void pubsubMsgSerializer_freeMsg(pubsub_msg_serializer_t* msgSerializer, void *msg); - -#endif /* PUBSUB_SERIALIZER_JSON_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/private/src/ps_activator.c ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/private/src/ps_activator.c b/pubsub/pubsub_serializer_json/private/src/ps_activator.c deleted file mode 100644 index fec5892..0000000 --- a/pubsub/pubsub_serializer_json/private/src/ps_activator.c +++ /dev/null @@ -1,107 +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. - */ -/* - * ps_activator.c - * - * \date Mar 24, 2017 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#include <stdlib.h> - -#include "bundle_activator.h" -#include "service_registration.h" - -#include "pubsub_serializer_impl.h" - -struct activator { - pubsub_serializer_t* serializer; - pubsub_serializer_service_t* serializerService; - service_registration_pt registration; -}; - -celix_status_t bundleActivator_create(bundle_context_pt context, void **userData) { - celix_status_t status = CELIX_SUCCESS; - struct activator *activator; - - activator = calloc(1, sizeof(*activator)); - if (!activator) { - status = CELIX_ENOMEM; - } - else{ - *userData = activator; - status = pubsubSerializer_create(context, &(activator->serializer)); - } - - return status; -} - -celix_status_t bundleActivator_start(void * userData, bundle_context_pt context) { - celix_status_t status = CELIX_SUCCESS; - struct activator *activator = userData; - pubsub_serializer_service_t* pubsubSerializerSvc = calloc(1, sizeof(*pubsubSerializerSvc)); - - if (!pubsubSerializerSvc) { - status = CELIX_ENOMEM; - } - else{ - pubsubSerializerSvc->handle = activator->serializer; - - pubsubSerializerSvc->createSerializerMap = (void*)pubsubSerializer_createSerializerMap; - pubsubSerializerSvc->destroySerializerMap = (void*)pubsubSerializer_destroySerializerMap; - activator->serializerService = pubsubSerializerSvc; - - /* Set serializer type */ - properties_pt props = properties_create(); - properties_set(props,PUBSUB_SERIALIZER_TYPE_KEY,PUBSUB_SERIALIZER_TYPE); - - status = bundleContext_registerService(context, PUBSUB_SERIALIZER_SERVICE, pubsubSerializerSvc, props, &activator->registration); - - } - - return status; -} - -celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context) { - celix_status_t status = CELIX_SUCCESS; - struct activator *activator = userData; - - serviceRegistration_unregister(activator->registration); - activator->registration = NULL; - - free(activator->serializerService); - activator->serializerService = NULL; - - return status; -} - -celix_status_t bundleActivator_destroy(void * userData, bundle_context_pt context) { - celix_status_t status = CELIX_SUCCESS; - struct activator *activator = userData; - - pubsubSerializer_destroy(activator->serializer); - activator->serializer = NULL; - - free(activator); - - return status; -} - - http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/private/src/pubsub_serializer_impl.c ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/private/src/pubsub_serializer_impl.c b/pubsub/pubsub_serializer_json/private/src/pubsub_serializer_impl.c deleted file mode 100644 index 685d499..0000000 --- a/pubsub/pubsub_serializer_json/private/src/pubsub_serializer_impl.c +++ /dev/null @@ -1,295 +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_impl.c - * - * \date Mar 24, 2017 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#include <stdio.h> -#include <stdlib.h> -#include <string.h> -#include <unistd.h> -#include <dirent.h> -#include <inttypes.h> - -#include "utils.h" -#include "hash_map.h" -#include "bundle_context.h" - -#include "log_helper.h" - -#include "json_serializer.h" - -#include "pubsub_serializer_impl.h" - -#define SYSTEM_BUNDLE_ARCHIVE_PATH "CELIX_FRAMEWORK_EXTENDER_PATH" -#define MAX_PATH_LEN 1024 - -static char* pubsubSerializer_getMsgDescriptionDir(bundle_pt bundle); -static void pubsubSerializer_addMsgSerializerFromBundle(const char *root, bundle_pt bundle, hash_map_pt msgTypesMap); -static void pubsubSerializer_fillMsgSerializerMap(hash_map_pt msgTypesMap,bundle_pt bundle); - -celix_status_t pubsubSerializer_create(bundle_context_pt context, pubsub_serializer_t** serializer) { - celix_status_t status = CELIX_SUCCESS; - - *serializer = calloc(1, sizeof(**serializer)); - - if (!*serializer) { - status = CELIX_ENOMEM; - } - else{ - - (*serializer)->bundle_context= context; - - if (logHelper_create(context, &(*serializer)->loghelper) == CELIX_SUCCESS) { - logHelper_start((*serializer)->loghelper); - } - - } - - return status; -} - -celix_status_t pubsubSerializer_destroy(pubsub_serializer_t* serializer) { - celix_status_t status = CELIX_SUCCESS; - - logHelper_stop(serializer->loghelper); - logHelper_destroy(&serializer->loghelper); - - free(serializer); - - return status; -} - -celix_status_t pubsubSerializer_createSerializerMap(pubsub_serializer_t* serializer, bundle_pt bundle, hash_map_pt* serializerMap) { - celix_status_t status = CELIX_SUCCESS; - - hash_map_pt map = hashMap_create(NULL, NULL, NULL, NULL); - - if (map != NULL) { - pubsubSerializer_fillMsgSerializerMap(map, bundle); - } else { - logHelper_log(serializer->loghelper, OSGI_LOGSERVICE_ERROR, "Cannot allocate memory for msg map"); - status = CELIX_ENOMEM; - } - - if (status == CELIX_SUCCESS) { - *serializerMap = map; - } - return status; -} - -celix_status_t pubsubSerializer_destroySerializerMap(pubsub_serializer_t* serializer, hash_map_pt serializerMap) { - celix_status_t status = CELIX_SUCCESS; - if (serializerMap == NULL) { - return CELIX_ILLEGAL_ARGUMENT; - } - - hash_map_iterator_t iter = hashMapIterator_construct(serializerMap); - while (hashMapIterator_hasNext(&iter)) { - pubsub_msg_serializer_t* msgSerializer = hashMapIterator_nextValue(&iter); - dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; - dynMessage_destroy(dynMsg); //note msgSer->name and msgSer->version owned by dynType - free(msgSerializer); //also contains the service struct. - } - - hashMap_destroy(serializerMap, false, false); - - return status; -} - - -celix_status_t pubsubMsgSerializer_serialize(pubsub_msg_serializer_t* msgSerializer, const void* msg, void** out, size_t *outLen) { - celix_status_t status = CELIX_SUCCESS; - - char *jsonOutput = NULL; - dyn_type* dynType = NULL; - dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; - dynMessage_getMessageType(dynMsg, &dynType); - - if (jsonSerializer_serialize(dynType, msg, &jsonOutput) != 0){ - status = CELIX_BUNDLE_EXCEPTION; - } - - if (status == CELIX_SUCCESS) { - *out = jsonOutput; - *outLen = strlen(jsonOutput) + 1; - } - - return status; -} - -celix_status_t pubsubMsgSerializer_deserialize(pubsub_msg_serializer_t* msgSerializer, const void* input, size_t inputLen, void **out) { - - celix_status_t status = CELIX_SUCCESS; - void *msg = NULL; - dyn_type* dynType = NULL; - dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; - dynMessage_getMessageType(dynMsg, &dynType); - - if (jsonSerializer_deserialize(dynType, (const char*)input, &msg) != 0) { - status = CELIX_BUNDLE_EXCEPTION; - } - else{ - *out = msg; - } - - return status; -} - -void pubsubMsgSerializer_freeMsg(pubsub_msg_serializer_t* msgSerializer, void *msg) { - dyn_type* dynType = NULL; - dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; - dynMessage_getMessageType(dynMsg, &dynType); - if (dynType != NULL) { - dynType_free(dynType, msg); - } -} - - -static void pubsubSerializer_fillMsgSerializerMap(hash_map_pt msgSerializers, bundle_pt bundle) { - char* root = NULL; - char* metaInfPath = NULL; - - root = pubsubSerializer_getMsgDescriptionDir(bundle); - - if(root != NULL){ - asprintf(&metaInfPath, "%s/META-INF/descriptors", root); - - pubsubSerializer_addMsgSerializerFromBundle(root, bundle, msgSerializers); - pubsubSerializer_addMsgSerializerFromBundle(metaInfPath, bundle, msgSerializers); - - free(metaInfPath); - free(root); - } -} - -static char* pubsubSerializer_getMsgDescriptionDir(bundle_pt bundle) -{ - char *root = NULL; - - bool isSystemBundle = false; - bundle_isSystemBundle(bundle, &isSystemBundle); - - if(isSystemBundle == true) { - bundle_context_pt context; - bundle_getContext(bundle, &context); - - const char *prop = NULL; - - bundleContext_getProperty(context, SYSTEM_BUNDLE_ARCHIVE_PATH, &prop); - - if(prop != NULL) { - root = strdup(prop); - } else { - root = getcwd(NULL, 0); - } - } else { - bundle_getEntry(bundle, ".", &root); - } - - return root; -} - - -static void pubsubSerializer_addMsgSerializerFromBundle(const char *root, bundle_pt bundle, hash_map_pt msgSerializers) -{ - char path[MAX_PATH_LEN]; - struct dirent *entry = NULL; - DIR *dir = opendir(root); - - if(dir) { - entry = readdir(dir); - } - - while (entry != NULL) { - - if (strstr(entry->d_name, ".descriptor") != NULL) { - - printf("DMU: Parsing entry '%s'\n", entry->d_name); - - snprintf(path, MAX_PATH_LEN, "%s/%s", root, entry->d_name); - FILE *stream = fopen(path,"r"); - - if (stream != NULL){ - dyn_message_type* msgType = NULL; - - int rc = dynMessage_parse(stream, &msgType); - if (rc == 0 && msgType != NULL) { - - char* msgName = NULL; - rc += dynMessage_getName(msgType,&msgName); - - version_pt msgVersion = NULL; - rc += dynMessage_getVersion(msgType, &msgVersion); - - if(rc == 0 && msgName != NULL && msgVersion != NULL){ - - unsigned int msgId = utils_stringHash(msgName); - - pubsub_msg_serializer_t *msgSerializer = calloc(1,sizeof(pubsub_msg_serializer_t)); - - msgSerializer->handle = msgType; - msgSerializer->msgId = msgId; - msgSerializer->msgName = msgName; - msgSerializer->msgVersion = msgVersion; - msgSerializer->serialize = (void*) pubsubMsgSerializer_serialize; - msgSerializer->deserialize = (void*) pubsubMsgSerializer_deserialize; - msgSerializer->freeMsg = (void*) pubsubMsgSerializer_freeMsg; - - bool clash = hashMap_containsKey(msgSerializers, (void*)(uintptr_t)msgId); - if (clash){ - printf("Cannot add msg %s. clash in msg id %d!!\n", msgName, msgId); - free(msgSerializer); - dynMessage_destroy(msgType); - } - else if (msgId != 0){ - printf("Adding %u : %s\n", msgId, msgName); - hashMap_put(msgSerializers, (void*)(uintptr_t)msgId, msgSerializer); - } - else{ - printf("Error creating msg serializer\n"); - free(msgSerializer); - dynMessage_destroy(msgType); - } - - } - else{ - printf("Cannot retrieve name and/or version from msg\n"); - } - - } else{ - printf("DMU: cannot parse message from descriptor %s\n.",path); - } - fclose(stream); - }else{ - printf("DMU: cannot open descriptor file %s\n.",path); - } - - } - entry = readdir(dir); - } - - if(dir) { - closedir(dir); - } -} http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/src/ps_activator.c ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/src/ps_activator.c b/pubsub/pubsub_serializer_json/src/ps_activator.c new file mode 100644 index 0000000..fec5892 --- /dev/null +++ b/pubsub/pubsub_serializer_json/src/ps_activator.c @@ -0,0 +1,107 @@ +/** + *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. + */ +/* + * ps_activator.c + * + * \date Mar 24, 2017 + * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ + +#include <stdlib.h> + +#include "bundle_activator.h" +#include "service_registration.h" + +#include "pubsub_serializer_impl.h" + +struct activator { + pubsub_serializer_t* serializer; + pubsub_serializer_service_t* serializerService; + service_registration_pt registration; +}; + +celix_status_t bundleActivator_create(bundle_context_pt context, void **userData) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator; + + activator = calloc(1, sizeof(*activator)); + if (!activator) { + status = CELIX_ENOMEM; + } + else{ + *userData = activator; + status = pubsubSerializer_create(context, &(activator->serializer)); + } + + return status; +} + +celix_status_t bundleActivator_start(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + pubsub_serializer_service_t* pubsubSerializerSvc = calloc(1, sizeof(*pubsubSerializerSvc)); + + if (!pubsubSerializerSvc) { + status = CELIX_ENOMEM; + } + else{ + pubsubSerializerSvc->handle = activator->serializer; + + pubsubSerializerSvc->createSerializerMap = (void*)pubsubSerializer_createSerializerMap; + pubsubSerializerSvc->destroySerializerMap = (void*)pubsubSerializer_destroySerializerMap; + activator->serializerService = pubsubSerializerSvc; + + /* Set serializer type */ + properties_pt props = properties_create(); + properties_set(props,PUBSUB_SERIALIZER_TYPE_KEY,PUBSUB_SERIALIZER_TYPE); + + status = bundleContext_registerService(context, PUBSUB_SERIALIZER_SERVICE, pubsubSerializerSvc, props, &activator->registration); + + } + + return status; +} + +celix_status_t bundleActivator_stop(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + + serviceRegistration_unregister(activator->registration); + activator->registration = NULL; + + free(activator->serializerService); + activator->serializerService = NULL; + + return status; +} + +celix_status_t bundleActivator_destroy(void * userData, bundle_context_pt context) { + celix_status_t status = CELIX_SUCCESS; + struct activator *activator = userData; + + pubsubSerializer_destroy(activator->serializer); + activator->serializer = NULL; + + free(activator); + + return status; +} + + http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.c ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.c b/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.c new file mode 100644 index 0000000..685d499 --- /dev/null +++ b/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.c @@ -0,0 +1,295 @@ +/** + *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_impl.c + * + * \date Mar 24, 2017 + * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ + +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> +#include <dirent.h> +#include <inttypes.h> + +#include "utils.h" +#include "hash_map.h" +#include "bundle_context.h" + +#include "log_helper.h" + +#include "json_serializer.h" + +#include "pubsub_serializer_impl.h" + +#define SYSTEM_BUNDLE_ARCHIVE_PATH "CELIX_FRAMEWORK_EXTENDER_PATH" +#define MAX_PATH_LEN 1024 + +static char* pubsubSerializer_getMsgDescriptionDir(bundle_pt bundle); +static void pubsubSerializer_addMsgSerializerFromBundle(const char *root, bundle_pt bundle, hash_map_pt msgTypesMap); +static void pubsubSerializer_fillMsgSerializerMap(hash_map_pt msgTypesMap,bundle_pt bundle); + +celix_status_t pubsubSerializer_create(bundle_context_pt context, pubsub_serializer_t** serializer) { + celix_status_t status = CELIX_SUCCESS; + + *serializer = calloc(1, sizeof(**serializer)); + + if (!*serializer) { + status = CELIX_ENOMEM; + } + else{ + + (*serializer)->bundle_context= context; + + if (logHelper_create(context, &(*serializer)->loghelper) == CELIX_SUCCESS) { + logHelper_start((*serializer)->loghelper); + } + + } + + return status; +} + +celix_status_t pubsubSerializer_destroy(pubsub_serializer_t* serializer) { + celix_status_t status = CELIX_SUCCESS; + + logHelper_stop(serializer->loghelper); + logHelper_destroy(&serializer->loghelper); + + free(serializer); + + return status; +} + +celix_status_t pubsubSerializer_createSerializerMap(pubsub_serializer_t* serializer, bundle_pt bundle, hash_map_pt* serializerMap) { + celix_status_t status = CELIX_SUCCESS; + + hash_map_pt map = hashMap_create(NULL, NULL, NULL, NULL); + + if (map != NULL) { + pubsubSerializer_fillMsgSerializerMap(map, bundle); + } else { + logHelper_log(serializer->loghelper, OSGI_LOGSERVICE_ERROR, "Cannot allocate memory for msg map"); + status = CELIX_ENOMEM; + } + + if (status == CELIX_SUCCESS) { + *serializerMap = map; + } + return status; +} + +celix_status_t pubsubSerializer_destroySerializerMap(pubsub_serializer_t* serializer, hash_map_pt serializerMap) { + celix_status_t status = CELIX_SUCCESS; + if (serializerMap == NULL) { + return CELIX_ILLEGAL_ARGUMENT; + } + + hash_map_iterator_t iter = hashMapIterator_construct(serializerMap); + while (hashMapIterator_hasNext(&iter)) { + pubsub_msg_serializer_t* msgSerializer = hashMapIterator_nextValue(&iter); + dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; + dynMessage_destroy(dynMsg); //note msgSer->name and msgSer->version owned by dynType + free(msgSerializer); //also contains the service struct. + } + + hashMap_destroy(serializerMap, false, false); + + return status; +} + + +celix_status_t pubsubMsgSerializer_serialize(pubsub_msg_serializer_t* msgSerializer, const void* msg, void** out, size_t *outLen) { + celix_status_t status = CELIX_SUCCESS; + + char *jsonOutput = NULL; + dyn_type* dynType = NULL; + dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; + dynMessage_getMessageType(dynMsg, &dynType); + + if (jsonSerializer_serialize(dynType, msg, &jsonOutput) != 0){ + status = CELIX_BUNDLE_EXCEPTION; + } + + if (status == CELIX_SUCCESS) { + *out = jsonOutput; + *outLen = strlen(jsonOutput) + 1; + } + + return status; +} + +celix_status_t pubsubMsgSerializer_deserialize(pubsub_msg_serializer_t* msgSerializer, const void* input, size_t inputLen, void **out) { + + celix_status_t status = CELIX_SUCCESS; + void *msg = NULL; + dyn_type* dynType = NULL; + dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; + dynMessage_getMessageType(dynMsg, &dynType); + + if (jsonSerializer_deserialize(dynType, (const char*)input, &msg) != 0) { + status = CELIX_BUNDLE_EXCEPTION; + } + else{ + *out = msg; + } + + return status; +} + +void pubsubMsgSerializer_freeMsg(pubsub_msg_serializer_t* msgSerializer, void *msg) { + dyn_type* dynType = NULL; + dyn_message_type *dynMsg = (dyn_message_type*)msgSerializer->handle; + dynMessage_getMessageType(dynMsg, &dynType); + if (dynType != NULL) { + dynType_free(dynType, msg); + } +} + + +static void pubsubSerializer_fillMsgSerializerMap(hash_map_pt msgSerializers, bundle_pt bundle) { + char* root = NULL; + char* metaInfPath = NULL; + + root = pubsubSerializer_getMsgDescriptionDir(bundle); + + if(root != NULL){ + asprintf(&metaInfPath, "%s/META-INF/descriptors", root); + + pubsubSerializer_addMsgSerializerFromBundle(root, bundle, msgSerializers); + pubsubSerializer_addMsgSerializerFromBundle(metaInfPath, bundle, msgSerializers); + + free(metaInfPath); + free(root); + } +} + +static char* pubsubSerializer_getMsgDescriptionDir(bundle_pt bundle) +{ + char *root = NULL; + + bool isSystemBundle = false; + bundle_isSystemBundle(bundle, &isSystemBundle); + + if(isSystemBundle == true) { + bundle_context_pt context; + bundle_getContext(bundle, &context); + + const char *prop = NULL; + + bundleContext_getProperty(context, SYSTEM_BUNDLE_ARCHIVE_PATH, &prop); + + if(prop != NULL) { + root = strdup(prop); + } else { + root = getcwd(NULL, 0); + } + } else { + bundle_getEntry(bundle, ".", &root); + } + + return root; +} + + +static void pubsubSerializer_addMsgSerializerFromBundle(const char *root, bundle_pt bundle, hash_map_pt msgSerializers) +{ + char path[MAX_PATH_LEN]; + struct dirent *entry = NULL; + DIR *dir = opendir(root); + + if(dir) { + entry = readdir(dir); + } + + while (entry != NULL) { + + if (strstr(entry->d_name, ".descriptor") != NULL) { + + printf("DMU: Parsing entry '%s'\n", entry->d_name); + + snprintf(path, MAX_PATH_LEN, "%s/%s", root, entry->d_name); + FILE *stream = fopen(path,"r"); + + if (stream != NULL){ + dyn_message_type* msgType = NULL; + + int rc = dynMessage_parse(stream, &msgType); + if (rc == 0 && msgType != NULL) { + + char* msgName = NULL; + rc += dynMessage_getName(msgType,&msgName); + + version_pt msgVersion = NULL; + rc += dynMessage_getVersion(msgType, &msgVersion); + + if(rc == 0 && msgName != NULL && msgVersion != NULL){ + + unsigned int msgId = utils_stringHash(msgName); + + pubsub_msg_serializer_t *msgSerializer = calloc(1,sizeof(pubsub_msg_serializer_t)); + + msgSerializer->handle = msgType; + msgSerializer->msgId = msgId; + msgSerializer->msgName = msgName; + msgSerializer->msgVersion = msgVersion; + msgSerializer->serialize = (void*) pubsubMsgSerializer_serialize; + msgSerializer->deserialize = (void*) pubsubMsgSerializer_deserialize; + msgSerializer->freeMsg = (void*) pubsubMsgSerializer_freeMsg; + + bool clash = hashMap_containsKey(msgSerializers, (void*)(uintptr_t)msgId); + if (clash){ + printf("Cannot add msg %s. clash in msg id %d!!\n", msgName, msgId); + free(msgSerializer); + dynMessage_destroy(msgType); + } + else if (msgId != 0){ + printf("Adding %u : %s\n", msgId, msgName); + hashMap_put(msgSerializers, (void*)(uintptr_t)msgId, msgSerializer); + } + else{ + printf("Error creating msg serializer\n"); + free(msgSerializer); + dynMessage_destroy(msgType); + } + + } + else{ + printf("Cannot retrieve name and/or version from msg\n"); + } + + } else{ + printf("DMU: cannot parse message from descriptor %s\n.",path); + } + fclose(stream); + }else{ + printf("DMU: cannot open descriptor file %s\n.",path); + } + + } + entry = readdir(dir); + } + + if(dir) { + closedir(dir); + } +} http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.h ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.h b/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.h new file mode 100644 index 0000000..c36f20e --- /dev/null +++ b/pubsub/pubsub_serializer_json/src/pubsub_serializer_impl.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. + */ +/* + * pubsub_serializer_impl.h + * + * \date Mar 24, 2017 + * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> + * \copyright Apache License, Version 2.0 + */ + +#ifndef PUBSUB_SERIALIZER_JSON_H_ +#define PUBSUB_SERIALIZER_JSON_H_ + +#include "dyn_common.h" +#include "dyn_type.h" +#include "dyn_message.h" +#include "log_helper.h" + +#include "pubsub_serializer.h" + +#define PUBSUB_SERIALIZER_TYPE "json" + +typedef struct pubsub_serializer { + bundle_context_pt bundle_context; + log_helper_pt loghelper; +} pubsub_serializer_t; + +celix_status_t pubsubSerializer_create(bundle_context_pt context, pubsub_serializer_t* *serializer); +celix_status_t pubsubSerializer_destroy(pubsub_serializer_t* serializer); + +celix_status_t pubsubSerializer_createSerializerMap(pubsub_serializer_t* serializer, bundle_pt bundle, hash_map_pt* serializerMap); +celix_status_t pubsubSerializer_destroySerializerMap(pubsub_serializer_t*, hash_map_pt serializerMap); + +/* Start of serializer specific functions */ +celix_status_t pubsubMsgSerializer_serialize(pubsub_msg_serializer_t* msgSerializer, const void* msg, void** out, size_t *outLen); +celix_status_t pubsubMsgSerializer_deserialize(pubsub_msg_serializer_t* msgSerializer, const void* input, size_t inputLen, void **out); +void pubsubMsgSerializer_freeMsg(pubsub_msg_serializer_t* msgSerializer, void *msg); + +#endif /* PUBSUB_SERIALIZER_JSON_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/pubsub_topology_manager/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/pubsub/pubsub_topology_manager/CMakeLists.txt b/pubsub/pubsub_topology_manager/CMakeLists.txt index b6eb796..784ca21 100644 --- a/pubsub/pubsub_topology_manager/CMakeLists.txt +++ b/pubsub/pubsub_topology_manager/CMakeLists.txt @@ -15,21 +15,12 @@ # specific language governing permissions and limitations # under the License. -include_directories("${PROJECT_SOURCE_DIR}/log_service/public/include") -include_directories("${PROJECT_SOURCE_DIR}/remote_services/remote_service_admin/public/include") -include_directories("${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/include") -include_directories("${PROJECT_SOURCE_DIR}/pubsub/pubsub_admin/public/include") -include_directories("${PROJECT_SOURCE_DIR}/pubsub/api/pubsub") -include_directories("private/include") -include_directories("public/include") - add_bundle(org.apache.celix.pubsub_topology_manager.PubSubTopologyManager BUNDLE_SYMBOLICNAME "apache_celix_pubsub_topology_manager" VERSION "1.0.0" SOURCES private/src/pstm_activator.c private/src/pubsub_topology_manager.c - ${PROJECT_SOURCE_DIR}/log_service/public/src/log_helper.c ${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/src/pubsub_endpoint.c ${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/src/pubsub_utils.c ) @@ -38,7 +29,8 @@ bundle_files(org.apache.celix.pubsub_topology_manager.PubSubTopologyManager ${PROJECT_SOURCE_DIR}/pubsub/pubsub_common/public/include/pubsub_topic_info.descriptor DESTINATION "META-INF/descriptors/services" ) +target_link_libraries(org.apache.celix.pubsub_topology_manager.PubSubTopologyManager PRIVATE Celix::framework Celix::log_helper) + -target_link_libraries(org.apache.celix.pubsub_topology_manager.PubSubTopologyManager celix_framework celix_utils) install_bundle(org.apache.celix.pubsub_topology_manager.PubSubTopologyManager) http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/pubsub/test/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/pubsub/test/CMakeLists.txt b/pubsub/test/CMakeLists.txt index 8279c0c..65e22e5 100644 --- a/pubsub/test/CMakeLists.txt +++ b/pubsub/test/CMakeLists.txt @@ -31,7 +31,7 @@ add_bundle(pubsub_sut test/sut_activator.c VERSION 1.0.0 ) -target_link_libraries(pubsub_sut celix_framework celix_utils) +target_link_libraries(pubsub_sut PRIVATE Celix::framework) bundle_files(pubsub_sut msg_descriptors/msg.descriptor msg_descriptors/sync.descriptor @@ -66,9 +66,9 @@ add_bundle(pubsub_tst ) if (APPLE) #Note that the launcher celix_test_runner is linked with CppuTest, not the bundle libs. Default libCppUTest.a is not compiled for relocation - target_link_libraries(pubsub_tst celix_framework celix_utils -Wl,-undefined -Wl,dynamic_lookup) + target_link_libraries(pubsub_tst PRIVATE Celix::framework -Wl,-undefined -Wl,dynamic_lookup) else () - target_link_libraries(pubsub_tst celix_framework celix_utils) + target_link_libraries(pubsub_tst PRIVATE Celix::framework) endif () bundle_files(pubsub_tst http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/remote_services/CMakeLists.txt b/remote_services/CMakeLists.txt index 767ece7..7ca1c44 100644 --- a/remote_services/CMakeLists.txt +++ b/remote_services/CMakeLists.txt @@ -21,14 +21,16 @@ if (REMOTE_SERVICE_ADMIN) add_subdirectory(examples) add_subdirectory(topology_manager) - + + add_subdirectory(discovery_common) add_subdirectory(discovery_configured) add_subdirectory(discovery_etcd) add_subdirectory(discovery_shm) #TODO refactor shm rsa to use dfi + add_subdirectory(remote_service_admin_api) + add_subdirectory(remote_service_admin_common) #add_subdirectory(remote_service_admin_shm) - add_subdirectory(remote_service_admin) add_subdirectory(remote_service_admin_dfi) http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/discovery.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/discovery.h b/remote_services/discovery/private/include/discovery.h deleted file mode 100644 index ee79caf..0000000 --- a/remote_services/discovery/private/include/discovery.h +++ /dev/null @@ -1,67 +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. - */ -/* - * discovery.h - * - * \date Sep 29, 2011 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef DISCOVERY_H_ -#define DISCOVERY_H_ - -#include "bundle_context.h" -#include "service_reference.h" - -#include "endpoint_description.h" -#include "endpoint_listener.h" - -#define DISCOVERY_SERVER_INTERFACE "DISCOVERY_CFG_SERVER_INTERFACE" -#define DISCOVERY_SERVER_IP "DISCOVERY_CFG_SERVER_IP" -#define DISCOVERY_SERVER_PORT "DISCOVERY_CFG_SERVER_PORT" -#define DISCOVERY_SERVER_PATH "DISCOVERY_CFG_SERVER_PATH" -#define DISCOVERY_POLL_ENDPOINTS "DISCOVERY_CFG_POLL_ENDPOINTS" -#define DISCOVERY_SERVER_MAX_EP "DISCOVERY_CFG_SERVER_MAX_EP" - -typedef struct discovery *discovery_pt; - - -/* those one could be put into a general discovery.h - file */ -celix_status_t discovery_create(bundle_context_pt context, discovery_pt *discovery); -celix_status_t discovery_destroy(discovery_pt discovery); - -celix_status_t discovery_start(discovery_pt discovery); -celix_status_t discovery_stop(discovery_pt discovery); - -celix_status_t discovery_endpointAdded(void *handle, endpoint_description_pt endpoint, char *machtedFilter); -celix_status_t discovery_endpointRemoved(void *handle, endpoint_description_pt endpoint, char *machtedFilter); - -celix_status_t discovery_endpointListenerAdding(void * handle, service_reference_pt reference, void **service); -celix_status_t discovery_endpointListenerAdded(void * handle, service_reference_pt reference, void * service); -celix_status_t discovery_endpointListenerModified(void * handle, service_reference_pt reference, void * service); -celix_status_t discovery_endpointListenerRemoved(void * handle, service_reference_pt reference, void * service); - -celix_status_t discovery_informEndpointListeners(discovery_pt discovery, endpoint_description_pt endpoint, bool endpointAdded); -celix_status_t discovery_updateEndpointListener(discovery_pt discovery, service_reference_pt reference, endpoint_listener_pt service); - -celix_status_t discovery_addDiscoveredEndpoint(discovery_pt discovery, endpoint_description_pt endpoint); -celix_status_t discovery_removeDiscoveredEndpoint(discovery_pt discovery, endpoint_description_pt endpoint); - -#endif /* DISCOVERY_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/endpoint_descriptor_common.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/endpoint_descriptor_common.h b/remote_services/discovery/private/include/endpoint_descriptor_common.h deleted file mode 100644 index a186a18..0000000 --- a/remote_services/discovery/private/include/endpoint_descriptor_common.h +++ /dev/null @@ -1,61 +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_descriptor_common.h - * - * \date Aug 8, 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef ENDPOINT_DESCRIPTOR_COMMON_H_ -#define ENDPOINT_DESCRIPTOR_COMMON_H_ - -/* - * Private constant & enum definitions for endpoint descriptor reader and writer, not needed for normal usage of the reader and writer. - */ - -typedef enum { - VALUE_TYPE_STRING, - VALUE_TYPE_LONG, - VALUE_TYPE_DOUBLE, - VALUE_TYPE_FLOAT, - VALUE_TYPE_INTEGER, - VALUE_TYPE_BYTE, - VALUE_TYPE_CHAR, - VALUE_TYPE_BOOLEAN, - VALUE_TYPE_SHORT, -} valueType; - -static const __attribute__((unused)) xmlChar* XML = (const xmlChar*) "xml"; -static const __attribute__((unused)) xmlChar* XMLNS = (const xmlChar*) "http://www.osgi.org/xmlns/rsa/v1.0.0";; - -static const __attribute__((unused)) xmlChar* ENDPOINT_DESCRIPTIONS = (const xmlChar*) "endpoint-descriptions"; -static const xmlChar* ENDPOINT_DESCRIPTION = (const xmlChar*) "endpoint-description"; - -static const xmlChar* ARRAY = (const xmlChar*) "array"; -static const __attribute__((unused)) xmlChar* LIST = (const xmlChar*) "list"; -static const __attribute__((unused)) xmlChar* SET = (const xmlChar*) "set"; - -static const xmlChar* PROPERTY = (const xmlChar*) "property"; -static const xmlChar* NAME = (const xmlChar*) "name"; -static const xmlChar* VALUE = (const xmlChar*) "value"; -static const xmlChar* VALUE_TYPE = (const xmlChar*) "value-type"; - -#endif /* ENDPOINT_DESCRIPTOR_COMMON_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/endpoint_descriptor_reader.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/endpoint_descriptor_reader.h b/remote_services/discovery/private/include/endpoint_descriptor_reader.h deleted file mode 100644 index 7a05d9e..0000000 --- a/remote_services/discovery/private/include/endpoint_descriptor_reader.h +++ /dev/null @@ -1,42 +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_descriptor_reader.h - * - * \date 26 Jul 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef ENDPOINT_DESCRIPTOR_READER_H_ -#define ENDPOINT_DESCRIPTOR_READER_H_ - -#include "endpoint_discovery_poller.h" -#include "celix_errno.h" -#include "array_list.h" - -typedef struct endpoint_descriptor_reader *endpoint_descriptor_reader_pt; - -celix_status_t endpointDescriptorReader_create(endpoint_discovery_poller_pt poller, endpoint_descriptor_reader_pt *reader); -celix_status_t endpointDescriptorReader_destroy(endpoint_descriptor_reader_pt reader); - -celix_status_t endpointDescriptorReader_parseDocument(endpoint_descriptor_reader_pt reader, char *document, array_list_pt *endpoints); - - -#endif /* ENDPOINT_DESCRIPTOR_READER_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/endpoint_descriptor_writer.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/endpoint_descriptor_writer.h b/remote_services/discovery/private/include/endpoint_descriptor_writer.h deleted file mode 100644 index 3c5a9be..0000000 --- a/remote_services/discovery/private/include/endpoint_descriptor_writer.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. - */ -/* - * endpoint_descriptor_writer.h - * - * \date 26 Jul 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef ENDPOINT_DESCRIPTOR_WRITER_H_ -#define ENDPOINT_DESCRIPTOR_WRITER_H_ - -#include "celix_errno.h" -#include "array_list.h" - -typedef struct endpoint_descriptor_writer *endpoint_descriptor_writer_pt; - -celix_status_t endpointDescriptorWriter_create(endpoint_descriptor_writer_pt *writer); -celix_status_t endpointDescriptorWriter_destroy(endpoint_descriptor_writer_pt writer); -celix_status_t endpointDescriptorWriter_writeDocument(endpoint_descriptor_writer_pt writer, array_list_pt endpoints, char **document); - -#endif /* ENDPOINT_DESCRIPTOR_WRITER_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/endpoint_discovery_poller.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/endpoint_discovery_poller.h b/remote_services/discovery/private/include/endpoint_discovery_poller.h deleted file mode 100644 index d344e55..0000000 --- a/remote_services/discovery/private/include/endpoint_discovery_poller.h +++ /dev/null @@ -1,56 +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_discovery_poller.h - * - * \date 3 Jul 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef ENDPOINT_DISCOVERY_POLLER_H_ -#define ENDPOINT_DISCOVERY_POLLER_H_ - -#include "celix_errno.h" -#include "discovery.h" -#include "log_helper.h" - -struct endpoint_discovery_poller { - discovery_pt discovery; - hash_map_pt entries; - log_helper_pt* loghelper; - - celix_thread_mutex_t pollerLock; - celix_thread_t pollerThread; - - unsigned int poll_interval; - volatile bool running; -}; - -typedef struct endpoint_discovery_poller *endpoint_discovery_poller_pt; - -celix_status_t endpointDiscoveryPoller_create(discovery_pt discovery, bundle_context_pt context, endpoint_discovery_poller_pt *poller); -celix_status_t endpointDiscoveryPoller_destroy(endpoint_discovery_poller_pt poller); - -celix_status_t endpointDiscoveryPoller_addDiscoveryEndpoint(endpoint_discovery_poller_pt poller, char *url); -celix_status_t endpointDiscoveryPoller_removeDiscoveryEndpoint(endpoint_discovery_poller_pt poller, char *url); - -celix_status_t endpointDiscoveryPoller_getDiscoveryEndpoints(endpoint_discovery_poller_pt poller, array_list_pt urls); - -#endif /* ENDPOINT_DISCOVERY_POLLER_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/include/endpoint_discovery_server.h ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/include/endpoint_discovery_server.h b/remote_services/discovery/private/include/endpoint_discovery_server.h deleted file mode 100644 index 51082b5..0000000 --- a/remote_services/discovery/private/include/endpoint_discovery_server.h +++ /dev/null @@ -1,81 +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_discovery_server.h - * - * \date Aug 12, 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ - -#ifndef ENDPOINT_DISCOVERY_SERVER_H_ -#define ENDPOINT_DISCOVERY_SERVER_H_ - -#include "celix_errno.h" -#include "discovery.h" - -typedef struct endpoint_discovery_server *endpoint_discovery_server_pt; - -/** - * Creates and starts a new instance of an endpoint discovery server. - * - * @param discovery [in] the discovery service itself; - * @param context [in] the bundle context; - * @param server [out] the pointer to the created instance. - * @return CELIX_SUCCESS when successful. - */ -celix_status_t endpointDiscoveryServer_create(discovery_pt discovery, bundle_context_pt context, endpoint_discovery_server_pt *server); - -/** - * Stops and destroys a given instance of an endpoint discovery server. - * - * @param server [in] the pointer to the instance to destroy. - * @return CELIX_SUCCESS when successful. - */ -celix_status_t endpointDiscoveryServer_destroy(endpoint_discovery_server_pt server); - -/** - * Adds a given endpoint description to expose through the given discovery server. - * - * @param server [in] the endpoint discovery server to expose the endpoint through; - * @param endpoint [in] the endpoint description to expose. - * @return CELIX_SUCCESS when successful. - */ -celix_status_t endpointDiscoveryServer_addEndpoint(endpoint_discovery_server_pt server, endpoint_description_pt endpoint); - -/** - * Removes a given endpoint description from exposure through the given discovery server. - * - * @param server [in] the endpoint discovery server to remove the endpoint from; - * @param endpoint [in] the endpoint description to remove. - * @return CELIX_SUCCESS when successful. - */ -celix_status_t endpointDiscoveryServer_removeEndpoint( endpoint_discovery_server_pt server, endpoint_description_pt endpoint); - -/** - * Returns the url, which is used by the discovery server to announce the endpoints - * - * @param server [in] the endpoint discovery server to retrieve the url from - * @param url [out] url which is used to announce the endpoints. - * @return CELIX_SUCCESS when successful. - */ -celix_status_t endpointDiscoveryServer_getUrl(endpoint_discovery_server_pt server, char* url); - - -#endif /* ENDPOINT_DISCOVERY_SERVER_H_ */ http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/src/desc.xml ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/src/desc.xml b/remote_services/discovery/private/src/desc.xml deleted file mode 100644 index 5998992..0000000 --- a/remote_services/discovery/private/src/desc.xml +++ /dev/null @@ -1,41 +0,0 @@ -<?xml version="1.0" encoding="UTF-8"?> -<!-- - *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-descriptions xmlns="http://www.osgi.org/xmlns/rsa/v1.0.0";> - <endpoint-description> - <property name="service.intents"> - <list> - <value>SOAP</value> - <value>HTTP</value> - </list> - </property> - <property name="endpoint.id" value="http://ws.acme.com:9000/hello"; /> - <property name="objectClass" value="com.acme.Foo" /> - <property name="endpoint.package.version.com.acme" value="4.2" /> - <property name="service.imported.configs" value="com.acme" /> - <property name="com.acme.ws.xml"> - <xml> - <config xmlns="http://acme.com/defs";> - <port>1029</port> - <host>www.acme.com</host> - </config> - </xml> - </property> - </endpoint-description> -</endpoint-descriptions> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/celix/blob/2a670f26/remote_services/discovery/private/src/discovery.c ---------------------------------------------------------------------- diff --git a/remote_services/discovery/private/src/discovery.c b/remote_services/discovery/private/src/discovery.c deleted file mode 100644 index e40a887..0000000 --- a/remote_services/discovery/private/src/discovery.c +++ /dev/null @@ -1,233 +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. - */ -/* - * discovery.c - * - * \date Aug 8, 2014 - * \author <a href="mailto:[email protected]";>Apache Celix Project Team</a> - * \copyright Apache License, Version 2.0 - */ -#include <stdio.h> -#include <stdbool.h> -#include <netdb.h> -#include <string.h> - -#include "celix_threads.h" -#include "bundle_context.h" -#include "log_helper.h" -#include "discovery.h" -#include "discovery_impl.h" - - -celix_status_t discovery_endpointAdded(void *handle, endpoint_description_pt endpoint, char *matchedFilter) { - celix_status_t status; - discovery_pt discovery = handle; - - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "Endpoint for %s, with filter \"%s\" added...", endpoint->service, matchedFilter); - - status = endpointDiscoveryServer_addEndpoint(discovery->server, endpoint); - - return status; -} - -celix_status_t discovery_endpointRemoved(void *handle, endpoint_description_pt endpoint, char *matchedFilter) { - celix_status_t status; - discovery_pt discovery = handle; - - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "Endpoint for %s, with filter \"%s\" removed...", endpoint->service, matchedFilter); - - status = endpointDiscoveryServer_removeEndpoint(discovery->server, endpoint); - - return status; -} - -celix_status_t discovery_endpointListenerAdding(void* handle, service_reference_pt reference, void** service) { - celix_status_t status = CELIX_SUCCESS; - discovery_pt discovery = handle; - - bundleContext_getService(discovery->context, reference, service); - - return status; -} - -celix_status_t discovery_endpointListenerAdded(void* handle, service_reference_pt reference, void* service) { - celix_status_t status = CELIX_SUCCESS; - discovery_pt discovery = handle; - - const char *discoveryListener = NULL; - serviceReference_getProperty(reference, "DISCOVERY", &discoveryListener); - const char *scope = NULL; - serviceReference_getProperty(reference, OSGI_ENDPOINT_LISTENER_SCOPE, &scope); - - filter_pt filter = filter_create(scope); - - if (discoveryListener != NULL && strcmp(discoveryListener, "true") == 0) { - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "EndpointListener Ignored - Discovery listener"); - } else { - celixThreadMutex_lock(&discovery->discoveredServicesMutex); - - hash_map_iterator_pt iter = hashMapIterator_create(discovery->discoveredServices); - while (hashMapIterator_hasNext(iter)) { - endpoint_description_pt endpoint = hashMapIterator_nextValue(iter); - - bool matchResult = false; - filter_match(filter, endpoint->properties, &matchResult); - if (matchResult) { - endpoint_listener_pt listener = service; - - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "EndpointListener Added - Add Scope"); - - listener->endpointAdded(listener->handle, endpoint, NULL); - } - } - hashMapIterator_destroy(iter); - - celixThreadMutex_unlock(&discovery->discoveredServicesMutex); - - celixThreadMutex_lock(&discovery->listenerReferencesMutex); - - hashMap_put(discovery->listenerReferences, reference, NULL); - - celixThreadMutex_unlock(&discovery->listenerReferencesMutex); - } - - filter_destroy(filter); - - return status; -} - -celix_status_t discovery_endpointListenerModified(void * handle, service_reference_pt reference, void * service) { - celix_status_t status; - - status = discovery_endpointListenerRemoved(handle, reference, service); - if (status == CELIX_SUCCESS) { - status = discovery_endpointListenerAdded(handle, reference, service); - } - - return status; -} - -celix_status_t discovery_endpointListenerRemoved(void * handle, service_reference_pt reference, void * service) { - celix_status_t status; - discovery_pt discovery = handle; - - status = celixThreadMutex_lock(&discovery->listenerReferencesMutex); - - if (status == CELIX_SUCCESS) { - if (discovery->listenerReferences != NULL) { - if (hashMap_remove(discovery->listenerReferences, reference)) { - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "EndpointListener Removed"); - } - } - - status = celixThreadMutex_unlock(&discovery->listenerReferencesMutex); - } - - return status; -} - -celix_status_t discovery_informEndpointListeners(discovery_pt discovery, endpoint_description_pt endpoint, bool endpointAdded) { - celix_status_t status; - - // Inform listeners of new endpoint - status = celixThreadMutex_lock(&discovery->listenerReferencesMutex); - - if (status == CELIX_SUCCESS) { - if (discovery->listenerReferences != NULL) { - hash_map_iterator_pt iter = hashMapIterator_create(discovery->listenerReferences); - while (hashMapIterator_hasNext(iter)) { - hash_map_entry_pt entry = hashMapIterator_nextEntry(iter); - - service_reference_pt reference = hashMapEntry_getKey(entry); - endpoint_listener_pt listener = NULL; - - const char* scope = NULL; - serviceReference_getProperty(reference, OSGI_ENDPOINT_LISTENER_SCOPE, &scope); - - filter_pt filter = filter_create(scope); - bool matchResult = false; - - status = filter_match(filter, endpoint->properties, &matchResult); - if (status == CELIX_SUCCESS) { - if (matchResult) { - bundleContext_getService(discovery->context, reference, (void **) &listener); - if (endpointAdded) { - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "Adding service (%s)", endpoint->service); - - listener->endpointAdded(listener->handle, endpoint, (char*)scope); - } else { - logHelper_log(discovery->loghelper, OSGI_LOGSERVICE_INFO, "Removing service (%s)", endpoint->service); - - listener->endpointRemoved(listener->handle, endpoint, (char*)scope); - } - } - - filter_destroy(filter); - } - } - hashMapIterator_destroy(iter); - } - - status = celixThreadMutex_unlock(&discovery->listenerReferencesMutex); - } - - return status; -} - -celix_status_t discovery_addDiscoveredEndpoint(discovery_pt discovery, endpoint_description_pt endpoint) { - celix_status_t status; - - status = celixThreadMutex_lock(&discovery->discoveredServicesMutex); - - if (status == CELIX_SUCCESS) { - char *endpointId = endpoint->id; - bool exists = hashMap_get(discovery->discoveredServices, endpointId) != NULL; - if (!exists) { - hashMap_put(discovery->discoveredServices, endpointId, endpoint); - } - - status = celixThreadMutex_unlock(&discovery->discoveredServicesMutex); - - if (!exists) { - // notify our listeners that a new endpoint is available... - discovery_informEndpointListeners(discovery, endpoint, true /* addingService */); - } - } - - return status; -} - -celix_status_t discovery_removeDiscoveredEndpoint(discovery_pt discovery, endpoint_description_pt endpoint) { - celix_status_t status; - - status = celixThreadMutex_lock(&discovery->discoveredServicesMutex); - - if (status == CELIX_SUCCESS) { - char *endpointId = endpoint->id; - void *oldValue = hashMap_remove(discovery->discoveredServices, endpointId); - - status = celixThreadMutex_unlock(&discovery->discoveredServicesMutex); - - if (oldValue) { - status = discovery_informEndpointListeners(discovery, endpoint, false /* removeService */); - } - } - - return status; -}
