Oipo commented on a change in pull request #286:
URL: https://github.com/apache/celix/pull/286#discussion_r501061104
##########
File path: bundles/pubsub/pubsub_spi/src/pubsub_endpoint.c
##########
@@ -170,8 +170,9 @@ celix_properties_t*
pubsubEndpoint_createFromPublisherTrackerInfo(bundle_context
data.topic = topic;
celix_bundleContext_useBundle(ctx, bundleId, &data,
retrieveTopicProperties);
+ pubsubEndpoint_setFields(ep, fwUUID, scope, topic,
PUBSUB_PUBLISHER_ENDPOINT_TYPE, NULL, NULL, NULL, data.props);
+
if (data.props != NULL) {
Review comment:
You can just remove this if statement, celix_properties_destroy also
checks for NULL.
##########
File path: libs/framework/gtest/src/bundle_context_bundles_tests.cpp
##########
@@ -378,6 +378,90 @@ TEST_F(CelixBundleContextBundlesTests, trackBundlesTest) {
celix_bundleContext_stopTracker(ctx, trackerId);
};
+
+TEST_F(CelixBundleContextBundlesTests, trackBundlesTestAsync) {
+ struct data {
+ std::atomic<int> installedCount{0};
+ std::atomic<int> startedCount{0};
+ std::atomic<int> stoppedCount{0};
+ };
+ struct data data;
+
+ auto installed = [](void *handle, const bundle_t *bnd) {
+ auto *d = static_cast<struct data*>(handle);
+ EXPECT_TRUE(bnd != nullptr);
+ d->installedCount.fetch_add(1);
Review comment:
Stylistic choice: `installedCount.fetch_add(1)` is semantically the same
as `installedCount += 1`. It's only when you want to specifiy memory order that
the former is the only option.
##########
File path: libs/framework/gtest/src/bundle_context_bundles_tests.cpp
##########
@@ -378,6 +378,90 @@ TEST_F(CelixBundleContextBundlesTests, trackBundlesTest) {
celix_bundleContext_stopTracker(ctx, trackerId);
};
+
+TEST_F(CelixBundleContextBundlesTests, trackBundlesTestAsync) {
+ struct data {
+ std::atomic<int> installedCount{0};
+ std::atomic<int> startedCount{0};
+ std::atomic<int> stoppedCount{0};
+ };
+ struct data data;
+
+ auto installed = [](void *handle, const bundle_t *bnd) {
+ auto *d = static_cast<struct data*>(handle);
+ EXPECT_TRUE(bnd != nullptr);
+ d->installedCount.fetch_add(1);
+ };
+
+ auto started = [](void *handle, const bundle_t *bnd) {
+ auto *d = static_cast<struct data*>(handle);
+ EXPECT_TRUE(bnd != nullptr);
+ d->startedCount.fetch_add(1);
+ };
+
+ auto stopped = [](void *handle, const bundle_t *bnd) {
+ auto *d = static_cast<struct data*>(handle);
+ if (bnd == nullptr) {
+ celix_logUtils_logToStdout("test", CELIX_LOG_LEVEL_ERROR, "bnd
should not be null");
+ }
+ EXPECT_TRUE(bnd != nullptr);
+ d->stoppedCount.fetch_add(1);
+ };
+
+ celix_bundle_tracking_options_t opts{};
+ opts.callbackHandle = static_cast<void*>(&data);
+ opts.onInstalled = installed;
+ opts.onStarted = started;
+ opts.onStopped = stopped;
+
+ long bundleId1 = celix_bundleContext_installBundle(ctx, TEST_BND1_LOC,
true);
+ celix_framework_waitForEmptyEventQueue(fw);
+ EXPECT_TRUE(bundleId1 >= 0);
+
+ /*
+ * NOTE for bundles already installed (TEST_BND1) the callbacks are called
on the
+ * thread of celix_bundleContext_trackBundlesWithOptions.
+ * For Bundles installed after the
celix_bundleContext_trackBundlesWithOptions function
+ * the called are called on the Celix framework event queue thread.
+ */
+ long trackerId = celix_bundleContext_trackBundlesWithOptionsAsync(ctx,
&opts);
+ celix_bundleContext_waitForAsyncTracker(ctx, trackerId);
+ EXPECT_EQ(1, data.installedCount.load());
Review comment:
Same stylistic choice here, the implicit conversion to int is the same
as `.load()`.
##########
File path: libs/utils/src/utils.c
##########
@@ -150,9 +150,9 @@ int utils_compareServiceIdsAndRanking(unsigned long servId,
long servRank, unsig
if (servId == otherServId) {
result = 0;
} else if (servRank != otherServRank) {
- result = servRank < otherServRank ? -1 : 1;
Review comment:
Is this a bug or a breaking change?
##########
File path: libs/framework/include/celix_framework.h
##########
@@ -156,6 +156,12 @@ void
celix_framework_waitForEmptyEventQueue(celix_framework_t *fw);
void celix_framework_setLogCallback(celix_framework_t* fw, void* logHandle,
void (*logFunction)(void* handle, celix_log_level_e level, const char* file,
const char *function, int line, const char *format, va_list formatArgs));
+/**
+ * wait till all events for the bundle identified by the bndId are processed.
+ */
+void celix_framework_waitForEvents(celix_framework_t* fw, long bndId);
Review comment:
A better name would probably be
`celix_framework_waitUntilNoEventsForBnd` or something
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -382,30 +546,74 @@ typedef struct celix_service_tracking_options {
.removeWithProperties = NULL, \
.setWithOwner = NULL, \
.addWithOwner = NULL, \
- .removeWithOwner = NULL}
+ .removeWithOwner = NULL, \
+ .trackerCreatedCallbackData = NULL, \
+ .trackerCreatedCallback = NULL }
#endif
/**
* Tracks services using the provided tracker options.
* The tracker options are only using during this call and can safely be
freed/reused after this call returns.
*
+ * The service tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
+ * @param ctx The bundle context.
+ * @param opts The pointer to the tracker options.
+ * @return the tracker id (>=0) or < 0 if unsuccessful.
+ */
+long celix_bundleContext_trackServicesWithOptionsAsync(celix_bundle_context_t
*ctx, const celix_service_tracking_options_t *opts);
Review comment:
Does `celix_service_tracking_options_t` contain a completion handler?
Would be useful here as well.
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -248,15 +353,39 @@ long
celix_bundleContext_findServiceWithOptions(celix_bundle_context_t *ctx, con
*/
celix_array_list_t*
celix_bundleContext_findServicesWithOptions(celix_bundle_context_t *ctx, const
celix_service_filter_options_t *opts);
+/**
+ * track the highest ranking service with the provided serviceName.
+ * The highest ranking services will used for the callback.
+ * If a new and higher ranking services the callback with be called again with
the new service.
+ * If a service is removed a the callback with be called with next highest
ranking service or NULL as service.
+ *
+ * The service tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
+ * @param ctx The bundle context.
+ * @param serviceName The required service name to track.
+ * If NULL is all service are tracked.
+ * @param callbackHandle The data pointer, which will be used in the callbacks
+ * @param set is a required callback, which will be called when a new highest
ranking service is set.
+ * @return the tracker id (>=0) or < 0 if unsuccessful.
+ */
+long celix_bundleContext_trackServiceAsync(
+ celix_bundle_context_t* ctx,
+ const char* serviceName,
+ void* callbackHandle,
+ void (*set)(void* handle, void* svc)
Review comment:
A completion handler would be useful here as well
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -137,18 +190,50 @@ typedef struct celix_service_registration_options {
.serviceName = NULL, \
.properties = NULL, \
.serviceLanguage = NULL, \
- .serviceVersion = NULL }
+ .serviceVersion = NULL, \
+ .asyncData = NULL, \
+ .asyncCallback = NULL }
#endif
+/**
+ * Register a service to the Celix framework using the provided service
registration options.
+ *
+ * The service will be registered async on the Celix event loop thread. This
means that service registration is (probably)
+ * not yet concluded when this function returns, but is added to the event
loop..
+ * Use celix_bundleContext_waitForAsyncRegistration to synchronise with the
+ * actual service registration in the framework's service registry.
+ *
+ * @param ctx The bundle context
+ * @param opts The pointer to the registration options. The options are only
in the during registration call.
+ * @return The serviceId (>= 0) or -1 if the registration was unsuccessful and
-2 if the registration was cancelled (@see celix_bundleContext_reserveSvcId).
+ */
+long
celix_bundleContext_registerServiceWithOptionsAsync(celix_bundle_context_t
*ctx, const celix_service_registration_options_t *opts);
Review comment:
Does `celix_service_registration_options_t` contain a completion
handler? Would be useful here as well.
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -40,21 +40,63 @@ extern "C" {
#define OPTS_INIT
#endif
+
+/**
+ * Register a service to the Celix framework.
+ *
+ * The service will be registered async on the Celix event loop thread. This
means that service registration is (probably)
+ * not yet concluded when this function returns, but is added to the event
loop.
+ * Use celix_bundleContext_waitForAsyncRegistration to synchronise with the
Review comment:
What would probably fit the event loop idiom better is a completion
handler. Though `celix_bundleContext_waitForAsyncRegistration` could still be
allowed for when sleeping is fine.
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -266,13 +395,37 @@ long celix_bundleContext_trackService(
const char* serviceName,
void* callbackHandle,
void (*set)(void* handle, void* svc)
+); //__attribute__((deprecated("Use celix_bundleContext_trackServiceSync
instead!")));
+
+/**
+ * track services with the provided serviceName.
+ *
+ * The service tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
+ * @param ctx The bundle context.
+ * @param serviceName The required service name to track
+ * If NULL is all service are tracked.
+ * @param callbackHandle The data pointer, which will be used in the callbacks
+ * @param add is a required callback, which will be called when a service is
added and initially for the existing service.
+ * @param remove is a required callback, which will be called when a service
is removed
+ * @return the tracker id (>=0) or < 0 if unsuccessful.
+ */
+long celix_bundleContext_trackServicesAsync(
+ celix_bundle_context_t* ctx,
+ const char* serviceName,
+ void* callbackHandle,
+ void (*add)(void* handle, void* svc),
+ void (*remove)(void* handle, void* svc)
Review comment:
A completion handler would be useful here as well
##########
File path: libs/framework/include/service_registry.h
##########
@@ -134,6 +144,37 @@ bool celix_serviceRegistry_getServiceInfo(
long celix_serviceRegistry_nextSvcId(celix_service_registry_t* registry);
+/**
+ * Unregister service for the provided service id (owned by bnd).
+ * Will print an error if the service id is invalid
+ */
+void celix_serviceRegistry_unregisterService(celix_service_registry_t*
registry, celix_bundle_t* bnd, long serviceId);
+
+
+/**
+ * Create a LDAP filter for the provided filter parts.
+ * @param serviceName The optional service name
+ * @param versionRange The optional version range
+ * @param filter The optional filter
Review comment:
name typo, should be `additionalFilter`. Also probably missing is why it
is additional?
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -655,23 +863,46 @@ bool
celix_bundleContext_startBundle(celix_bundle_context_t *ctx, long bndId);
*/
char* celix_bundleContext_getBundleSymbolicName(celix_bundle_context_t *ctx,
long bndId);
+
/**
* track bundles
* The add bundle callback will also be called for already installed bundles.
Review comment:
add bundle callback?
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -655,23 +863,46 @@ bool
celix_bundleContext_startBundle(celix_bundle_context_t *ctx, long bndId);
*/
char* celix_bundleContext_getBundleSymbolicName(celix_bundle_context_t *ctx,
long bndId);
+
/**
* track bundles
* The add bundle callback will also be called for already installed bundles.
*
+ * The bundle tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
* @param ctx The bundle context.
* @param callbackHandle The data pointer, which will be used in the
callbacks
* @param add The callback which will be called for started
bundles.
* @param remove The callback which will be called when bundles are
stopped.
* @return The bundle tracker id or < 0 if unsuccessful.
*/
-long celix_bundleContext_trackBundles(
+long celix_bundleContext_trackBundlesAsync(
celix_bundle_context_t* ctx,
void* callbackHandle,
void (*onStarted)(void* handle, const celix_bundle_t *bundle),
void (*onStopped)(void *handle, const celix_bundle_t *bundle)
);
+/**
+ * track bundles
+ * The add bundle callback will also be called for already installed bundles.
+ *
+ * Note: please use celix_bundleContext_trackBundlesAsync instead.
+ *
+ * @param ctx The bundle context.
+ * @param callbackHandle The data pointer, which will be used in the
callbacks
+ * @param add The callback which will be called for started
bundles.
Review comment:
Yeah, probably onStarted/onStopped.
##########
File path: libs/framework/src/framework.c
##########
@@ -341,7 +331,7 @@ celix_status_t framework_destroy(framework_pt framework) {
const char *bndName = celix_bundle_getSymbolicName(bnd);
fw_log(framework->logger, CELIX_LOG_LEVEL_FATAL, "Cannot destroy
framework. The use count of bundle %s (bnd id %li) is not 0, but %u.", bndName,
entry->bndId, count);
celixThreadMutex_lock(&framework->dispatcher.mutex);
- int nrOfRequests =
celix_arrayList_size(framework->dispatcher.requests);
+ int nrOfRequests = framework->dispatcher.eventQueueSize +
celix_arrayList_size(framework->dispatcher.dynamicEventQueue);
Review comment:
Can't comment on line 369, but because it's destroying a locked mutex,
thread sanitizer crashes.
##########
File path: libs/framework/src/service_registry.c
##########
@@ -1046,6 +920,137 @@ static inline void
celix_waitAndDestroyServiceListener(celix_service_registry_se
free(entry);
}
+/**
+ * Create a LDAP filter for the provided filter parts.
Review comment:
This documentation should only be available in the header file, no?
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -655,23 +863,46 @@ bool
celix_bundleContext_startBundle(celix_bundle_context_t *ctx, long bndId);
*/
char* celix_bundleContext_getBundleSymbolicName(celix_bundle_context_t *ctx,
long bndId);
+
/**
* track bundles
* The add bundle callback will also be called for already installed bundles.
*
+ * The bundle tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
* @param ctx The bundle context.
* @param callbackHandle The data pointer, which will be used in the
callbacks
* @param add The callback which will be called for started
bundles.
Review comment:
Probably onStarted/onStopped?
##########
File path: libs/framework/src/framework.c
##########
@@ -2528,13 +2631,108 @@ bool celix_framework_startBundle(celix_framework_t
*fw, long bndId) {
}
void celix_framework_waitForEmptyEventQueue(celix_framework_t *fw) {
+ assert(!celix_framework_isCurrentThreadTheEventLoop(fw));
+
celixThreadMutex_lock(&fw->dispatcher.mutex);
- while ((celix_arrayList_size(fw->dispatcher.requests) +
fw->dispatcher.nrOfLocalRequest) != 0) {
+ while (fw->dispatcher.eventQueueSize > 0 ||
celix_arrayList_size(fw->dispatcher.dynamicEventQueue) > 0) {
celixThreadCondition_wait(&fw->dispatcher.cond, &fw->dispatcher.mutex);
}
celixThreadMutex_unlock(&fw->dispatcher.mutex);
}
+void celix_framework_waitForEvents(celix_framework_t* fw, long bndId) {
+ assert(!celix_framework_isCurrentThreadTheEventLoop(fw));
+
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ bool eventInProgress = true;
+ while (eventInProgress) {
+ eventInProgress = false;
+ for (int i = 0; i < fw->dispatcher.eventQueueSize; ++i) {
+ int index = (fw->dispatcher.eventQueueFirstEntry + i) %
CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE;
+ celix_framework_event_t* e = &fw->dispatcher.eventQueue[index];
+ if (e->bndEntry != NULL && e->bndEntry->bndId == bndId) {
+ eventInProgress = true;
+ break;
+ }
+ }
+ for (int i = 0; !eventInProgress && i <
celix_arrayList_size(fw->dispatcher.dynamicEventQueue); ++i) {
+ celix_framework_event_t* e =
celix_arrayList_get(fw->dispatcher.dynamicEventQueue, i);
+ if (e->bndEntry != NULL && e->bndEntry->bndId == bndId) {
+ eventInProgress = true;
+ break;
+ }
+ }
+ if (eventInProgress) {
+ celixThreadCondition_timedwaitRelative(&fw->dispatcher.cond,
&fw->dispatcher.mutex, 5, 0);
+ }
+ }
+ celixThreadMutex_unlock(&fw->dispatcher.mutex);
+}
+
+
void celix_framework_setLogCallback(celix_framework_t* fw, void* logHandle,
void (*logFunction)(void* handle, celix_log_level_e level, const char* file,
const char *function, int line, const char *format, va_list formatArgs)) {
celix_frameworkLogger_setLogCallback(fw->logger, logHandle, logFunction);
+}
+
+
+long celix_framework_fireGenericEvent(framework_t* fw, long eventId, long
bndId, const char *eventName, void* processData, void (*processCallback)(void
*data), void* doneData, void (*doneCallback)(void* doneData)) {
+ celix_framework_bundle_entry_t* bndEntry = NULL;
+ if (bndId >=0) {
+ bndEntry = fw_bundleEntry_getBundleEntryAndIncreaseUseCount(fw, bndId);
+ if (bndEntry == NULL) {
+ fw_log(fw->logger, CELIX_LOG_LEVEL_ERROR, "Cannot find bundle for
id %li", bndId);
+ return -1L;
+ }
+ }
+
+ if (eventId < 0) {
+ eventId = celix_framework_nextEventId(fw);
+ }
+
+ celix_framework_event_t event;
+ memset(&event, 0, sizeof(event));
+ event.type = CELIX_GENERIC_EVENT;
+ event.bndEntry = bndEntry;
+ event.genericEventId = eventId;
+ event.genericEventName = eventName;
+ event.genericProcessData = processData;
+ event.genericProcess = processCallback;
+ event.doneData = doneData;
+ event.doneCallback = doneCallback;
+ celix_framework_addToEventQueue(fw, &event);
+
+ return eventId;
+}
+
+long celix_framework_nextEventId(framework_t *fw) {
+ return __atomic_fetch_add(&fw->nextGenericEventId, 1, __ATOMIC_SEQ_CST);;
Review comment:
Technically, this can even be `__ATOMIC_RELAXED`, because it's a simple
counter. See https://stackoverflow.com/a/48148318/1460998 for a more detailed
answer.
##########
File path: libs/framework/src/framework.c
##########
@@ -2015,67 +1934,112 @@ static void fw_handleEventRequest(celix_framework_t
*framework, request_t* reque
for (int i = 0; i < celix_arrayList_size(localListeners); ++i) {
fw_bundle_listener_pt listener = arrayList_get(localListeners, i);
- bundle_event_t event;
- memset(&event, 0, sizeof(event));
- event.bnd = request->bndEntry->bnd;
- event.type = request->eventType;
- fw_invokeBundleListener(framework, listener->listener, &event,
listener->bundle);
+ bundle_event_t bEvent;
+ memset(&bEvent, 0, sizeof(bEvent));
+ bEvent.bnd = event->bndEntry->bnd;
+ bEvent.type = event->bundleEvent;
+ fw_invokeBundleListener(framework, listener->listener, &bEvent,
listener->bundle);
fw_bundleListener_decreaseUseCount(listener);
}
celix_arrayList_destroy(localListeners);
- } else if (request->type == FRAMEWORK_EVENT_TYPE) {
+ } else if (event->type == CELIX_FRAMEWORK_EVENT_TYPE) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
event");
celixThreadMutex_lock(&framework->frameworkListenersLock);
- //TODO refactor use of framework listeners to use a useCount +
conditition.
for (int i = 0; i <
celix_arrayList_size(framework->frameworkListeners); ++i) {
fw_framework_listener_pt listener =
celix_arrayList_get(framework->frameworkListeners, i);
- framework_event_t event;
- memset(&event, 0, sizeof(event));
- event.type = request->eventType;
- event.error = request->error;
- event.errorCode = request->errorCode;
+ framework_event_t fEvent;
+ memset(&fEvent, 0, sizeof(fEvent));
+ fEvent.type = event->fwEvent;
+ fEvent.error = event->error;
+ fEvent.errorCode = event->errorCode;
- fw_invokeFrameworkListener(framework, listener->listener, &event,
listener->bundle);
+ fw_invokeFrameworkListener(framework, listener->listener, &fEvent,
listener->bundle);
}
celixThreadMutex_unlock(&framework->frameworkListenersLock);
+ } else if (event->type == CELIX_REGISTER_SERVICE_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
service registration event for service '%s' with svc id %li",
event->serviceName, event->registerServiceId);
+ service_registration_t* reg = NULL;
+ celix_status_t status;
+ if (event->factory != NULL) {
+ status =
celix_serviceRegistry_registerServiceFactory(framework->registry,
event->bndEntry->bnd, event->serviceName, event->factory, event->properties,
event->registerServiceId, ®);
+ } else {
+ status =
celix_serviceRegistry_registerService(framework->registry,
event->bndEntry->bnd, event->serviceName, event->svc, event->properties,
event->registerServiceId, ®);
+ }
+ if (status != CELIX_SUCCESS) {
+ fw_log(framework->logger, CELIX_LOG_LEVEL_ERROR, "Could not
register service async. svc name is %s, error is %s", event->serviceName,
celix_strerror(status));
+ } else if (event->registerCallback != NULL) {
+ event->registerCallback(event->registerData,
serviceRegistration_getServiceId(reg));
+ }
+ } else if (event->type == CELIX_UNREGISTER_SERVICE_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
service unregister event for service id %li", event->unregisterServiceId);
+ celix_serviceRegistry_unregisterService(framework->registry,
event->bndEntry->bnd, event->unregisterServiceId);
+ } else if (event->type == CELIX_GENERIC_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling event
%s", event->eventName);
+ if (event->genericProcess != NULL) {
+ event->genericProcess(event->genericProcessData);
+ }
+ }
+
+ if (event->doneCallback != NULL) {
+ event->doneCallback(event->doneData);
}
}
-static inline void fw_handleEvents(celix_framework_t* framework,
celix_array_list_t* localRequests) {
+static inline celix_framework_event_t* fw_topEventFromQueue(celix_framework_t*
fw) {
+ celix_framework_event_t* e = NULL;
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ if (fw->dispatcher.eventQueueSize > 0) {
+ e = &fw->dispatcher.eventQueue[fw->dispatcher.eventQueueFirstEntry];
+ } else if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) > 0) {
+ e = celix_arrayList_get(fw->dispatcher.dynamicEventQueue, 0);
+ }
+ celixThreadMutex_unlock(&fw->dispatcher.mutex);
+ return e;
+}
+
+static inline bool fw_removeTopEventFromQueue(celix_framework_t* fw) {
+ bool dynamicallyAllocated = false;
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ if (fw->dispatcher.eventQueueSize > 0) {
+ fw->dispatcher.eventQueueFirstEntry =
(fw->dispatcher.eventQueueFirstEntry+1) %
CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE;
+ fw->dispatcher.eventQueueSize -= 1;
Review comment:
Hmm, I think a dynamically sized ring buffer solves this quirk of having
two separate internal event queues.
Basically, have one `void *data`, one `uint64_t size`, one `uint64_t
beginIndex` and one `uint64_t endIndex`. The range between begin/end signify
the relevant events. If `beginIndex == endIndex` the ringbuffer is empty and
when inserting, `endIndex < beginIndex && endIndex + 1 == beginIndex ||
endIndex + 1 == size && beginIndex == 0` signifies it's full. In the latter
case, allocate a new `void*` with twice the size and memcpy everything between
begin/end. Insert a new event is also simply a memcpy.
##########
File path: libs/framework/include/service_registry.h
##########
@@ -95,6 +93,17 @@ celix_status_t
celix_serviceRegistry_addServiceListener(celix_service_registry_t
celix_status_t
celix_serviceRegistry_removeServiceListener(celix_service_registry_t *reg,
celix_service_listener_t *listener);
+bool celix_serviceRegistry_isServiceRegistered(celix_service_registry_t* reg,
long serviceId);
Review comment:
Missing doxygen-style documentation here. One of the questions I have,
for example, is what does `reserveId` do?
##########
File path: libs/framework/src/bundle_context.c
##########
@@ -80,15 +86,19 @@ celix_status_t bundleContext_destroy(bundle_context_pt
context) {
celixThreadMutex_lock(&context->mutex);
- bundleContext_cleanupBundleTrackers(context);
- bundleContext_cleanupServiceTrackers(context);
- bundleContext_cleanupServiceTrackerTrackers(context);
+ assert(hashMap_size(context->bundleTrackers) == 0);
+ hashMap_destroy(context->bundleTrackers, false, false);
+ assert(hashMap_size(context->serviceTrackers) == 0);
+ hashMap_destroy(context->serviceTrackers, false, false);
+ assert(hashMap_size(context->metaTrackers) == 0);
+ hashMap_destroy(context->metaTrackers, false, false);
+ assert(celix_arrayList_size(context->svcRegistrations) == 0);
+ celix_arrayList_destroy(context->svcRegistrations);
- //NOTE still present service registrations will be cleared during
bundle stop in the
+ //NOTE still present service registrations will be cleared during
bundle stop in the
Review comment:
But the context->svcRegistrations assert only 3 lines above says
otherwise?
##########
File path: libs/framework/include/celix_bundle_context.h
##########
@@ -655,23 +863,46 @@ bool
celix_bundleContext_startBundle(celix_bundle_context_t *ctx, long bndId);
*/
char* celix_bundleContext_getBundleSymbolicName(celix_bundle_context_t *ctx,
long bndId);
+
/**
* track bundles
* The add bundle callback will also be called for already installed bundles.
*
+ * The bundle tracker will be created async on the Celix event loop thread.
This means that the function can return
+ * before the tracker is created.
+ *
* @param ctx The bundle context.
* @param callbackHandle The data pointer, which will be used in the
callbacks
* @param add The callback which will be called for started
bundles.
* @param remove The callback which will be called when bundles are
stopped.
* @return The bundle tracker id or < 0 if unsuccessful.
*/
-long celix_bundleContext_trackBundles(
+long celix_bundleContext_trackBundlesAsync(
celix_bundle_context_t* ctx,
void* callbackHandle,
void (*onStarted)(void* handle, const celix_bundle_t *bundle),
void (*onStopped)(void *handle, const celix_bundle_t *bundle)
);
+/**
+ * track bundles
+ * The add bundle callback will also be called for already installed bundles.
Review comment:
add bundle callback? Do you mean onStarted callback here?
##########
File path: libs/framework/src/framework.c
##########
@@ -1922,88 +1861,68 @@ bundle_context_t*
framework_getContext(const_framework_pt framework) {
return result;
}
-celix_status_t fw_fireBundleEvent(framework_pt framework, bundle_event_type_e
eventType, celix_framework_bundle_entry_t* entry) {
- celix_status_t status = CELIX_SUCCESS;
-
+void fw_fireBundleEvent(framework_pt framework, bundle_event_type_e eventType,
celix_framework_bundle_entry_t* entry) {
if (eventType == OSGI_FRAMEWORK_BUNDLE_EVENT_STOPPING || eventType ==
OSGI_FRAMEWORK_BUNDLE_EVENT_UNINSTALLED || eventType ==
OSGI_FRAMEWORK_BUNDLE_EVENT_STOPPED) {
if (entry->bndId == framework->bundleId) {
//NOTE for framework bundle not triggering events while framework
is stopped (and as result in use)
- return CELIX_SUCCESS;
+ return;
}
}
- request_t* request = (request_t*) calloc(1, sizeof(*request));
- if (!request) {
- status = CELIX_ENOMEM;
- } else {
- fw_bundleEntry_increaseUseCount(entry);
+ fw_bundleEntry_increaseUseCount(entry);
- request->eventType = eventType;
- request->filter = NULL;
- request->type = BUNDLE_EVENT_TYPE;
- request->error = NULL;
- request->bndEntry = entry;
+ celix_framework_event_t event;
+ memset(&event, 0, sizeof(event));
+ event.type = CELIX_BUNDLE_EVENT_TYPE;
+ event.bndEntry = entry;
+ event.bundleEvent = eventType;
+ celix_framework_addToEventQueue(framework, &event);
+}
- celixThreadMutex_lock(&framework->dispatcher.mutex);
- if (framework->dispatcher.active) {
- //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Adding
dispatcher bundle event request for bnd id %li with event type %i",
entry->bndId, eventType);
- celix_arrayList_add(framework->dispatcher.requests, request);
- celixThreadCondition_broadcast(&framework->dispatcher.cond);
- } else {
- /*
- * NOTE because stopping the framework is done through stopping
the framework bundle,
- * most bundle stopping / stopped events cannot be fired.
- *
- * TBD if this needs to addressed.
- */
- fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Cannot fire
event dispatcher not active. Event is %x for bundle %s", eventType,
celix_bundle_getSymbolicName(entry->bnd));
- fw_bundleEntry_decreaseUseCount(entry);
- free(request);
- }
- celixThreadMutex_unlock(&framework->dispatcher.mutex);
+void fw_fireFrameworkEvent(framework_pt framework, framework_event_type_e
eventType, celix_status_t errorCode) {
+ celix_framework_event_t event;
+ memset(&event, 0, sizeof(event));
+ event.type = CELIX_FRAMEWORK_EVENT_TYPE;
+ event.fwEvent = eventType;
+ event.errorCode = errorCode;
+ event.error = "";
+ if (errorCode != CELIX_SUCCESS) {
+ event.error = celix_strerror(errorCode);
}
- framework_logIfError(framework->logger, status, NULL, "Failed to fire
bundle event");
-
- return status;
+ celix_framework_addToEventQueue(framework, &event);
}
-celix_status_t fw_fireFrameworkEvent(framework_pt framework,
framework_event_type_e eventType, celix_status_t errorCode) {
- celix_status_t status = CELIX_SUCCESS;
-
- request_t* request = calloc(1, sizeof(*request));
- if (!request) {
- status = CELIX_ENOMEM;
+static void celix_framework_addToEventQueue(celix_framework_t *fw, const
celix_framework_event_t* event) {
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ //try to add to static queue
+ if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) > 0) { //always
to dynamic queue if not empty (to ensure order)
+ celix_framework_event_t *e = malloc(sizeof(*e));
+ *e = *event; //shallow copy
+ celix_arrayList_add(fw->dispatcher.dynamicEventQueue, e);
+ if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) % 100 == 0)
{
+ fw_log(fw->logger, CELIX_LOG_LEVEL_WARNING, "dynamic event queue
size is %i. Is there a bundle blocking on the event loop thread?",
celix_arrayList_size(fw->dispatcher.dynamicEventQueue));
+ }
+ } else if (fw->dispatcher.eventQueueSize <
CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE) {
+ size_t index = (fw->dispatcher.eventQueueFirstEntry +
fw->dispatcher.eventQueueSize) %
+ CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE;
+ fw->dispatcher.eventQueue[index] = *event; //shallow copy
+ fw->dispatcher.eventQueueSize += 1;
} else {
- request->eventType = eventType;
- request->filter = NULL;
- request->type = FRAMEWORK_EVENT_TYPE;
- request->errorCode = errorCode;
- request->error = "";
-
- if (errorCode != CELIX_SUCCESS) {
- request->error = celix_strerror(errorCode);
- }
-
- celixThreadMutex_lock(&framework->dispatcher.mutex);
- if (framework->dispatcher.active) {
- //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Adding
dispatcher framework event request for event type %i", eventType);
- celix_arrayList_add(framework->dispatcher.requests, request);
- celixThreadCondition_broadcast(&framework->dispatcher.cond);
- } else {
- free(request);
- }
- celixThreadMutex_unlock(&framework->dispatcher.mutex);
- }
-
- framework_logIfError(framework->logger, status, NULL, "Failed to fire
framework event");
-
- return status;
+ //static queue is full, dynamics queue is empty. Add first entry to
dynamic queue
+ fw_log(fw->logger, CELIX_LOG_LEVEL_WARNING,
+ "Static event queue for celix framework is full, falling back
to dynamic allocated events. Increase static event queue size, current size is
%i", CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE);
+ celix_framework_event_t *e = malloc(sizeof(*e));
+ *e = *event; //shallow copy
+ celix_arrayList_add(fw->dispatcher.dynamicEventQueue, e);
+ }
+ celixThreadCondition_broadcast(&fw->dispatcher.cond);
+ celixThreadMutex_unlock(&fw->dispatcher.mutex);
}
-
-static void fw_handleEventRequest(celix_framework_t *framework, request_t*
request) {
- if (request->type == BUNDLE_EVENT_TYPE) {
+static void fw_handleEventRequest(celix_framework_t *framework,
celix_framework_event_t* event) {
+ if (event->type == CELIX_BUNDLE_EVENT_TYPE) {
Review comment:
This if is best done with a switch/case: https://gcc.godbolt.org/z/oKKnEY
##########
File path: libs/framework/src/framework.c
##########
@@ -2015,67 +1934,112 @@ static void fw_handleEventRequest(celix_framework_t
*framework, request_t* reque
for (int i = 0; i < celix_arrayList_size(localListeners); ++i) {
fw_bundle_listener_pt listener = arrayList_get(localListeners, i);
- bundle_event_t event;
- memset(&event, 0, sizeof(event));
- event.bnd = request->bndEntry->bnd;
- event.type = request->eventType;
- fw_invokeBundleListener(framework, listener->listener, &event,
listener->bundle);
+ bundle_event_t bEvent;
+ memset(&bEvent, 0, sizeof(bEvent));
+ bEvent.bnd = event->bndEntry->bnd;
+ bEvent.type = event->bundleEvent;
+ fw_invokeBundleListener(framework, listener->listener, &bEvent,
listener->bundle);
fw_bundleListener_decreaseUseCount(listener);
}
celix_arrayList_destroy(localListeners);
- } else if (request->type == FRAMEWORK_EVENT_TYPE) {
+ } else if (event->type == CELIX_FRAMEWORK_EVENT_TYPE) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
event");
celixThreadMutex_lock(&framework->frameworkListenersLock);
- //TODO refactor use of framework listeners to use a useCount +
conditition.
for (int i = 0; i <
celix_arrayList_size(framework->frameworkListeners); ++i) {
fw_framework_listener_pt listener =
celix_arrayList_get(framework->frameworkListeners, i);
- framework_event_t event;
- memset(&event, 0, sizeof(event));
- event.type = request->eventType;
- event.error = request->error;
- event.errorCode = request->errorCode;
+ framework_event_t fEvent;
+ memset(&fEvent, 0, sizeof(fEvent));
+ fEvent.type = event->fwEvent;
+ fEvent.error = event->error;
+ fEvent.errorCode = event->errorCode;
- fw_invokeFrameworkListener(framework, listener->listener, &event,
listener->bundle);
+ fw_invokeFrameworkListener(framework, listener->listener, &fEvent,
listener->bundle);
}
celixThreadMutex_unlock(&framework->frameworkListenersLock);
+ } else if (event->type == CELIX_REGISTER_SERVICE_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
service registration event for service '%s' with svc id %li",
event->serviceName, event->registerServiceId);
+ service_registration_t* reg = NULL;
+ celix_status_t status;
+ if (event->factory != NULL) {
+ status =
celix_serviceRegistry_registerServiceFactory(framework->registry,
event->bndEntry->bnd, event->serviceName, event->factory, event->properties,
event->registerServiceId, ®);
+ } else {
+ status =
celix_serviceRegistry_registerService(framework->registry,
event->bndEntry->bnd, event->serviceName, event->svc, event->properties,
event->registerServiceId, ®);
+ }
+ if (status != CELIX_SUCCESS) {
+ fw_log(framework->logger, CELIX_LOG_LEVEL_ERROR, "Could not
register service async. svc name is %s, error is %s", event->serviceName,
celix_strerror(status));
+ } else if (event->registerCallback != NULL) {
+ event->registerCallback(event->registerData,
serviceRegistration_getServiceId(reg));
+ }
+ } else if (event->type == CELIX_UNREGISTER_SERVICE_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw
service unregister event for service id %li", event->unregisterServiceId);
+ celix_serviceRegistry_unregisterService(framework->registry,
event->bndEntry->bnd, event->unregisterServiceId);
+ } else if (event->type == CELIX_GENERIC_EVENT) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling event
%s", event->eventName);
+ if (event->genericProcess != NULL) {
+ event->genericProcess(event->genericProcessData);
+ }
+ }
+
+ if (event->doneCallback != NULL) {
+ event->doneCallback(event->doneData);
}
}
-static inline void fw_handleEvents(celix_framework_t* framework,
celix_array_list_t* localRequests) {
+static inline celix_framework_event_t* fw_topEventFromQueue(celix_framework_t*
fw) {
+ celix_framework_event_t* e = NULL;
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ if (fw->dispatcher.eventQueueSize > 0) {
+ e = &fw->dispatcher.eventQueue[fw->dispatcher.eventQueueFirstEntry];
+ } else if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) > 0) {
+ e = celix_arrayList_get(fw->dispatcher.dynamicEventQueue, 0);
+ }
+ celixThreadMutex_unlock(&fw->dispatcher.mutex);
+ return e;
+}
+
+static inline bool fw_removeTopEventFromQueue(celix_framework_t* fw) {
+ bool dynamicallyAllocated = false;
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ if (fw->dispatcher.eventQueueSize > 0) {
+ fw->dispatcher.eventQueueFirstEntry =
(fw->dispatcher.eventQueueFirstEntry+1) %
CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE;
+ fw->dispatcher.eventQueueSize -= 1;
Review comment:
Am I correct in saying that if the ring buffer is overflowed and dynamic
events have been allocated, that the FIFO principle gets violated? As in, if an
event gets inserted into the ring buffer (once at least one event has been
handled) before the ring buffer is empty, the dynamically allocated events have
to wait?
##########
File path: libs/framework/src/framework.c
##########
@@ -1922,88 +1861,68 @@ bundle_context_t*
framework_getContext(const_framework_pt framework) {
return result;
}
-celix_status_t fw_fireBundleEvent(framework_pt framework, bundle_event_type_e
eventType, celix_framework_bundle_entry_t* entry) {
- celix_status_t status = CELIX_SUCCESS;
-
+void fw_fireBundleEvent(framework_pt framework, bundle_event_type_e eventType,
celix_framework_bundle_entry_t* entry) {
if (eventType == OSGI_FRAMEWORK_BUNDLE_EVENT_STOPPING || eventType ==
OSGI_FRAMEWORK_BUNDLE_EVENT_UNINSTALLED || eventType ==
OSGI_FRAMEWORK_BUNDLE_EVENT_STOPPED) {
if (entry->bndId == framework->bundleId) {
//NOTE for framework bundle not triggering events while framework
is stopped (and as result in use)
- return CELIX_SUCCESS;
+ return;
}
}
- request_t* request = (request_t*) calloc(1, sizeof(*request));
- if (!request) {
- status = CELIX_ENOMEM;
- } else {
- fw_bundleEntry_increaseUseCount(entry);
+ fw_bundleEntry_increaseUseCount(entry);
- request->eventType = eventType;
- request->filter = NULL;
- request->type = BUNDLE_EVENT_TYPE;
- request->error = NULL;
- request->bndEntry = entry;
+ celix_framework_event_t event;
+ memset(&event, 0, sizeof(event));
+ event.type = CELIX_BUNDLE_EVENT_TYPE;
+ event.bndEntry = entry;
+ event.bundleEvent = eventType;
+ celix_framework_addToEventQueue(framework, &event);
+}
- celixThreadMutex_lock(&framework->dispatcher.mutex);
- if (framework->dispatcher.active) {
- //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Adding
dispatcher bundle event request for bnd id %li with event type %i",
entry->bndId, eventType);
- celix_arrayList_add(framework->dispatcher.requests, request);
- celixThreadCondition_broadcast(&framework->dispatcher.cond);
- } else {
- /*
- * NOTE because stopping the framework is done through stopping
the framework bundle,
- * most bundle stopping / stopped events cannot be fired.
- *
- * TBD if this needs to addressed.
- */
- fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Cannot fire
event dispatcher not active. Event is %x for bundle %s", eventType,
celix_bundle_getSymbolicName(entry->bnd));
- fw_bundleEntry_decreaseUseCount(entry);
- free(request);
- }
- celixThreadMutex_unlock(&framework->dispatcher.mutex);
+void fw_fireFrameworkEvent(framework_pt framework, framework_event_type_e
eventType, celix_status_t errorCode) {
+ celix_framework_event_t event;
+ memset(&event, 0, sizeof(event));
+ event.type = CELIX_FRAMEWORK_EVENT_TYPE;
+ event.fwEvent = eventType;
+ event.errorCode = errorCode;
+ event.error = "";
+ if (errorCode != CELIX_SUCCESS) {
+ event.error = celix_strerror(errorCode);
}
- framework_logIfError(framework->logger, status, NULL, "Failed to fire
bundle event");
-
- return status;
+ celix_framework_addToEventQueue(framework, &event);
}
-celix_status_t fw_fireFrameworkEvent(framework_pt framework,
framework_event_type_e eventType, celix_status_t errorCode) {
- celix_status_t status = CELIX_SUCCESS;
-
- request_t* request = calloc(1, sizeof(*request));
- if (!request) {
- status = CELIX_ENOMEM;
+static void celix_framework_addToEventQueue(celix_framework_t *fw, const
celix_framework_event_t* event) {
+ celixThreadMutex_lock(&fw->dispatcher.mutex);
+ //try to add to static queue
+ if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) > 0) { //always
to dynamic queue if not empty (to ensure order)
+ celix_framework_event_t *e = malloc(sizeof(*e));
+ *e = *event; //shallow copy
+ celix_arrayList_add(fw->dispatcher.dynamicEventQueue, e);
+ if (celix_arrayList_size(fw->dispatcher.dynamicEventQueue) % 100 == 0)
{
+ fw_log(fw->logger, CELIX_LOG_LEVEL_WARNING, "dynamic event queue
size is %i. Is there a bundle blocking on the event loop thread?",
celix_arrayList_size(fw->dispatcher.dynamicEventQueue));
+ }
+ } else if (fw->dispatcher.eventQueueSize <
CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE) {
+ size_t index = (fw->dispatcher.eventQueueFirstEntry +
fw->dispatcher.eventQueueSize) %
+ CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE;
+ fw->dispatcher.eventQueue[index] = *event; //shallow copy
+ fw->dispatcher.eventQueueSize += 1;
} else {
- request->eventType = eventType;
- request->filter = NULL;
- request->type = FRAMEWORK_EVENT_TYPE;
- request->errorCode = errorCode;
- request->error = "";
-
- if (errorCode != CELIX_SUCCESS) {
- request->error = celix_strerror(errorCode);
- }
-
- celixThreadMutex_lock(&framework->dispatcher.mutex);
- if (framework->dispatcher.active) {
- //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Adding
dispatcher framework event request for event type %i", eventType);
- celix_arrayList_add(framework->dispatcher.requests, request);
- celixThreadCondition_broadcast(&framework->dispatcher.cond);
- } else {
- free(request);
- }
- celixThreadMutex_unlock(&framework->dispatcher.mutex);
- }
-
- framework_logIfError(framework->logger, status, NULL, "Failed to fire
framework event");
-
- return status;
+ //static queue is full, dynamics queue is empty. Add first entry to
dynamic queue
+ fw_log(fw->logger, CELIX_LOG_LEVEL_WARNING,
+ "Static event queue for celix framework is full, falling back
to dynamic allocated events. Increase static event queue size, current size is
%i", CELIX_FRAMEWORK_STATIC_EVENT_QUEUE_SIZE);
+ celix_framework_event_t *e = malloc(sizeof(*e));
+ *e = *event; //shallow copy
+ celix_arrayList_add(fw->dispatcher.dynamicEventQueue, e);
+ }
+ celixThreadCondition_broadcast(&fw->dispatcher.cond);
+ celixThreadMutex_unlock(&fw->dispatcher.mutex);
}
-
-static void fw_handleEventRequest(celix_framework_t *framework, request_t*
request) {
- if (request->type == BUNDLE_EVENT_TYPE) {
+static void fw_handleEventRequest(celix_framework_t *framework,
celix_framework_event_t* event) {
+ if (event->type == CELIX_BUNDLE_EVENT_TYPE) {
+ //fw_log(framework->logger, CELIX_LOG_LEVEL_TRACE, "Handling fw bundle
event for bundle %s", event->bndEntry->bnd->symbolicName);
Review comment:
Is this statement still needed?
##########
File path: libs/framework/src/service_tracker.c
##########
@@ -128,345 +111,285 @@ celix_status_t serviceTracker_create(bundle_context_pt
context, const char * ser
return status;
}
-celix_status_t serviceTracker_createWithFilter(bundle_context_pt context,
const char * filter, service_tracker_customizer_pt customizer,
service_tracker_pt *tracker) {
- celix_status_t status = CELIX_SUCCESS;
- *tracker = (service_tracker_pt) calloc(1, sizeof(**tracker));
- if (!*tracker) {
- status = CELIX_ENOMEM;
- } else {
- (*tracker)->context = context;
- (*tracker)->filter = strdup(filter);
- (*tracker)->customizer = customizer;
- }
+celix_status_t serviceTracker_createWithFilter(bundle_context_pt context,
const char * filter, service_tracker_customizer_pt customizer,
service_tracker_pt *out) {
+ service_tracker_t* tracker = calloc(1, sizeof(*tracker));
+ *out = tracker;
+ tracker->context = context;
+ tracker->filter = celix_utils_strdup(filter);
+ tracker->customizer = *customizer;
+ free(customizer);
- framework_logIfError(celix_frameworkLogger_globalLogger(), status,
NULL, "Cannot create service tracker [filter=%s]", filter);
+ celixThreadMutex_create(&tracker->closeSync.mutex, NULL);
+ celixThreadCondition_init(&tracker->closeSync.cond, NULL);
- return status;
+ celixThreadMutex_create(&tracker->mutex, NULL);
+ celixThreadCondition_init(&tracker->cond, NULL);
+ tracker->trackedServices = celix_arrayList_create();
+ tracker->untrackingServices = celix_arrayList_create();
+
+ celixThreadMutex_create(&tracker->mutex, NULL);
+ tracker->currentHighestServiceId = -1;
+
+ tracker->listener.handle = tracker;
+ tracker->listener.serviceChanged = (void *) serviceTracker_serviceChanged;
+
+ tracker->callbackHandle = tracker->callbackHandle;
+
+ tracker->add = tracker->add;
+ tracker->addWithProperties = tracker->addWithProperties;
+ tracker->addWithOwner = tracker->addWithOwner;
+ tracker->set = tracker->set;
+ tracker->setWithProperties = tracker->setWithProperties;
+ tracker->setWithOwner = tracker->setWithOwner;
+ tracker->remove = tracker->remove;
+ tracker->removeWithProperties = tracker->removeWithProperties;
+ tracker->removeWithOwner = tracker->removeWithOwner;
+
+ return CELIX_SUCCESS;
}
celix_status_t serviceTracker_destroy(service_tracker_pt tracker) {
- if (tracker->customizer != NULL) {
- serviceTrackerCustomizer_destroy(tracker->customizer);
- }
-
free(tracker->serviceName);
free(tracker->filter);
- free(tracker);
-
+ celixThreadMutex_destroy(&tracker->closeSync.mutex);
+ celixThreadCondition_destroy(&tracker->closeSync.cond);
+ celixThreadMutex_destroy(&tracker->mutex);
+ celixThreadCondition_destroy(&tracker->cond);
+ celix_arrayList_destroy(tracker->trackedServices);
+ celix_arrayList_destroy(tracker->untrackingServices);
+ free(tracker);
return CELIX_SUCCESS;
}
celix_status_t serviceTracker_open(service_tracker_pt tracker) {
- celix_service_listener_t *listener = NULL;
- celix_service_tracker_instance_t *instance = NULL;
- celix_status_t status = CELIX_SUCCESS;
-
- bool addListener = false;
-
- celixThreadRwlock_writeLock(&tracker->instanceLock);
- if (tracker->instance == NULL) {
- instance = calloc(1, sizeof(*instance));
- instance->context = tracker->context;
-
- instance->closing = false;
- instance->activeServiceChangeCalls = 0;
- celixThreadMutex_create(&instance->closingLock, NULL);
- celixThreadCondition_init(&instance->activeServiceChangeCallsCond,
NULL);
-
-
- celixThreadRwlock_create(&instance->lock, NULL);
- instance->trackedServices = celix_arrayList_create();
-
- celixThreadMutex_create(&instance->mutex, NULL);
- instance->currentHighestServiceId = -1;
-
- instance->listener.handle = instance;
- instance->listener.serviceChanged = (void *)
serviceTracker_serviceChanged;
- listener = &instance->listener;
-
- instance->callbackHandle = tracker->callbackHandle;
- instance->filter = strdup(tracker->filter);
- if (tracker->customizer != NULL) {
- memcpy(&instance->customizer, tracker->customizer,
sizeof(instance->customizer));
- }
- instance->add = tracker->add;
- instance->addWithProperties = tracker->addWithProperties;
- instance->addWithOwner = tracker->addWithOwner;
- instance->set = tracker->set;
- instance->setWithProperties = tracker->setWithProperties;
- instance->setWithOwner = tracker->setWithOwner;
- instance->remove = tracker->remove;
- instance->removeWithProperties = tracker->removeWithProperties;
- instance->removeWithOwner = tracker->removeWithOwner;
-
- tracker->instance = instance;
-
- addListener = true;
- } else {
- //already open
- framework_logIfError(tracker->context->framework->logger, status,
NULL, "Tracker already open");
+ celixThreadMutex_lock(&tracker->mutex);
+ bool alreadyOpen = tracker->open;
+ tracker->open = true;
+ celixThreadMutex_unlock(&tracker->mutex);
+ if (!alreadyOpen) {
+ bundleContext_addServiceListener(tracker->context, &tracker->listener,
tracker->filter);
}
- celixThreadRwlock_unlock(&tracker->instanceLock);
-
- if (addListener) {
- bundleContext_addServiceListener(tracker->context, listener,
tracker->filter);
- }
return CELIX_SUCCESS;
}
-celix_status_t serviceTracker_close(service_tracker_pt tracker) {
+celix_status_t serviceTracker_close(service_tracker_t* tracker) {
//put all tracked entries in tmp array list, so that the untrack (etc)
calls are not blocked.
//set state to close to prevent service listener events
- celixThreadRwlock_writeLock(&tracker->instanceLock);
- celix_service_tracker_instance_t *instance = tracker->instance;
- tracker->instance = NULL;
- if (instance != NULL) {
- celixThreadMutex_lock(&instance->closingLock);
- //prevent service listener events
- instance->closing = true;
- celixThreadMutex_unlock(&instance->closingLock);
- }
- celixThreadRwlock_unlock(&tracker->instanceLock);
-
- if (instance != NULL) {
- celixThreadRwlock_writeLock(&instance->lock);
- unsigned int size = celix_arrayList_size(instance->trackedServices);
- if(size > 0) {
- celix_tracked_entry_t *trackedEntries[size];
- for (unsigned int i = 0u; i < size; i++) {
- trackedEntries[i] = (celix_tracked_entry_t *)
arrayList_get(instance->trackedServices, i);
- }
- arrayList_clear(instance->trackedServices);
- celixThreadRwlock_unlock(&instance->lock);
+ celixThreadMutex_lock(&tracker->mutex);
+ bool open = tracker->open;
+ tracker->open = false;
+ celixThreadMutex_unlock(&tracker->mutex);
- //loop trough tracked entries an untrack
- for (unsigned int i = 0u; i < size; i++) {
- serviceTracker_untrackTracked(instance, trackedEntries[i]);
+ if (!open) {
+ return CELIX_SUCCESS;
+ }
+
+
+ //indicate that the service tracking is closing and wait for the still
pending service registration events.
+ celixThreadMutex_lock(&tracker->closeSync.mutex);
+ tracker->closeSync.closing = true;
+ while (tracker->closeSync.activeCalls > 0) {
+ celixThreadCondition_wait(&tracker->closeSync.cond,
&tracker->closeSync.mutex);
+ }
+ celixThreadMutex_unlock(&tracker->closeSync.mutex);
+
+ int nrOfTrackedEntries;
+ do {
+ celixThreadMutex_lock(&tracker->mutex);
+ celix_tracked_entry_t* tracked = NULL;
+ nrOfTrackedEntries = celix_arrayList_size(tracker->trackedServices);
+ if (nrOfTrackedEntries > 0) {
+ tracked = celix_arrayList_get(tracker->trackedServices, 0);
+ celix_arrayList_removeAt(tracker->trackedServices, 0);
+ celix_arrayList_add(tracker->untrackingServices, tracked);
+ }
+ celixThreadMutex_unlock(&tracker->mutex);
+
+ if (tracked != NULL) {
+ int currentSize = nrOfTrackedEntries - 1;
+ if (currentSize == 0) {
+ serviceTracker_checkAndInvokeSetService(tracker, NULL, NULL,
NULL);
+ } else {
+ celix_serviceTracker_useHighestRankingService(tracker,
tracked->serviceName, tracker, NULL, NULL,
serviceTracker_checkAndInvokeSetService);
}
- } else {
- celixThreadRwlock_unlock(&instance->lock);
- }
- celixThreadMutex_lock(&instance->closingLock);
- while (instance->activeServiceChangeCalls > 0) {
- celixThreadCondition_wait(&instance->activeServiceChangeCallsCond,
&instance->closingLock);
+ serviceTracker_untrackTracked(tracker, tracked);
+ celixThreadMutex_lock(&tracker->mutex);
+ celix_arrayList_remove(tracker->untrackingServices, tracked);
+ celixThreadCondition_broadcast(&tracker->cond);
+ celixThreadMutex_unlock(&tracker->mutex);
}
- celixThreadMutex_unlock(&instance->closingLock);
-#ifdef CELIX_SERVICE_TRACKER_USE_SHUTDOWN_THREAD
- //NOTE Separate thread is needed to prevent deadlock where closing is
triggered from a serviceChange event and the
- // untrack -> removeServiceListener will try to remove a service
listener which is being invoked and is the
- // actual thread calling the removeServiceListener.
- //
- // This can be detached -> because service listener events are ignored
(closing=true) and so no callbacks
- // are made back to the celix framework / tracker owner.
- serviceTracker_addInstanceFromShutdownList(instance);
- celix_thread_t localThread;
- celixThread_create(&localThread, NULL,
shutdownServiceTrackerInstanceHandler, instance);
- celixThread_detach(localThread);
-#else
- fw_removeServiceListener(instance->context->framework,
instance->context->bundle, &instance->listener);
-
- celixThreadMutex_destroy(&instance->closingLock);
- celixThreadCondition_destroy(&instance->activeServiceChangeCallsCond);
- celixThreadMutex_destroy(&instance->mutex);
- celixThreadRwlock_destroy(&instance->lock);
- celix_arrayList_destroy(instance->trackedServices);
- free(instance->filter);
- free(instance);
-#endif
- }
+ celixThreadMutex_lock(&tracker->mutex);
+ nrOfTrackedEntries = celix_arrayList_size(tracker->trackedServices);
+ celixThreadMutex_unlock(&tracker->mutex);
+ } while (nrOfTrackedEntries > 0);
+
+
+ fw_removeServiceListener(tracker->context->framework,
tracker->context->bundle, &tracker->listener);
return CELIX_SUCCESS;
}
-service_reference_pt serviceTracker_getServiceReference(service_tracker_pt
tracker) {
+service_reference_pt serviceTracker_getServiceReference(service_tracker_t*
tracker) {
//TODO deprecated warning -> not locked
- celix_tracked_entry_t *tracked;
+
service_reference_pt result = NULL;
- unsigned int i;
-
- celixThreadRwlock_readLock(&tracker->instanceLock);
- celix_service_tracker_instance_t *instance = tracker->instance;
- if (instance != NULL) {
- celixThreadRwlock_readLock(&instance->lock);
- for (i = 0; i < arrayList_size(instance->trackedServices); ++i) {
- tracked = (celix_tracked_entry_t *)
arrayList_get(instance->trackedServices, i);
- result = tracked->reference;
- break;
- }
- celixThreadRwlock_unlock(&instance->lock);
+
+ celixThreadMutex_lock(&tracker->mutex);
+ for (int i = 0; i < celix_arrayList_size(tracker->trackedServices); ++i) {
+ celix_tracked_entry_t *tracked =
celix_arrayList_get(tracker->trackedServices, i);
+ result = tracked->reference;
+ break;
}
- celixThreadRwlock_unlock(&tracker->instanceLock);
+ celixThreadMutex_unlock(&tracker->mutex);
return result;
}
-array_list_pt serviceTracker_getServiceReferences(service_tracker_pt tracker) {
+array_list_pt serviceTracker_getServiceReferences(service_tracker_t* tracker) {
//TODO deprecated warning -> not locked
Review comment:
Is this comment still relevant?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
