xuzhenbao commented on code in PR #773:
URL: https://github.com/apache/celix/pull/773#discussion_r1853163798
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -415,6 +454,131 @@ int
celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, c
return CELIX_SUCCESS;
}
+int celix_eventAdmin_addRemoteProviderService(void* handle, void* svc, const
celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ celix_event_admin_t* ea = handle;
+ long serviceId = celix_properties_getAsLong(props,
CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Remote provider service id is
invalid.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_longHashMap_put(ea->remoteProviderServices, serviceId, svc);
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeRemoteProviderService(void* handle, void* svc,
const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ celix_event_admin_t* ea = handle;
+ long serviceId = celix_properties_getAsLong(props,
CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Remote provider service id is
invalid.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_longHashMap_remove(ea->remoteProviderServices, serviceId);
+ return CELIX_SUCCESS;
+}
+
+static void
celix_eventAdmin_retrieveLongTimeUnusedEventSeqIdCache(celix_event_admin_t* ea)
{
+ if (celix_stringHashMap_size(ea->eventSeqIdCache) > 16) {
+ celix_string_hash_map_iterator_t iter =
celix_stringHashMap_begin(ea->eventSeqIdCache);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_seq_id_cache_t* cache = iter.value.ptrValue;
+ if (celix_elapsedtime(CLOCK_MONOTONIC, cache->lastModified) >
60*60/*1h*/) {
+ celix_stringHashMapIterator_remove(&iter);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+ }
+ return;
+}
+
+static bool celix_eventAdmin_isDuplicateEvent(celix_event_admin_t* ea, const
char* topic CELIX_UNUSED, const celix_properties_t* properties) {
+ const char* remoteFwUUID = celix_properties_get(properties,
CELIX_EVENT_REMOTE_FRAMEWORK_UUID, NULL);
+ if (remoteFwUUID == NULL) {
+ return false;
+ }
+ long seqId = celix_properties_getAsLong(properties,
CELIX_EVENT_REMOTE_SEQ_ID, -1L);
+ if (seqId <= 0) {
+ return false;
+ }
+ long seqIdMod = seqId % CELIX_EVENT_ADMIN_MAX_EVENT_SEQ_ID_CACHE_SIZE;
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_seq_id_cache_t* seqIdCache =
celix_stringHashMap_get(ea->eventSeqIdCache, remoteFwUUID);
+ if (seqIdCache == NULL) {
+ celix_autofree celix_event_seq_id_cache_t* cache = calloc(1,
sizeof(*cache));
+ if (cache == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event seq
id cache for %s.", remoteFwUUID);
+ return false;
+ }
+ celix_status_t status = celix_stringHashMap_put(ea->eventSeqIdCache,
remoteFwUUID, cache);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event seq id
cache for %s.", remoteFwUUID);
+ return false;
+ }
+ seqIdCache = celix_steal_ptr(cache);
+ }
+ seqIdCache->lastModified = celix_gettime(CLOCK_MONOTONIC);
+ if (seqIdCache->seqIdBuffer[seqIdMod] == seqId) {
+ return true;
+ }
+ seqIdCache->seqIdBuffer[seqIdMod] = seqId;
+
+ celix_eventAdmin_retrieveLongTimeUnusedEventSeqIdCache(ea);
+
+ return false;
+}
+
+static long celix_eventAdmin_getEventSeqId(celix_event_admin_t* ea) {
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard =
celixRwlockWlockGuard_init(&ea->lock);
+ long seqId = ea->nextSeqId++;
+ if (seqId <= 0) {
+ seqId = 1;
+ ea->nextSeqId = seqId + 1;
+ }
+ return seqId;
+}
+
+static int celix_eventAdmin_deliverEventToRemote(celix_event_admin_t* ea,
const char* topic, const celix_properties_t* props, bool async) {
+ celix_autoptr(celix_properties_t) remoteProps =
celix_properties_copy(props);
+ if (remoteProps == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to copy remote properties
for event %s.", topic);
+ return ENOMEM;
+ }
+ celix_properties_unset(remoteProps, CELIX_EVENT_REMOTE_ENABLE);
+ celix_status_t status = celix_properties_set(remoteProps,
CELIX_EVENT_REMOTE_FRAMEWORK_UUID, ea->fwUUID);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to set remote framework
uuid for event %s.", topic);
+ return status;
+ }
+ long seqId = celix_eventAdmin_getEventSeqId(ea);
+ status = celix_properties_setLong(remoteProps, CELIX_EVENT_REMOTE_SEQ_ID,
seqId);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to set remote seq id for
event %s.", topic);
+ return status;
+ }
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard =
celixRwlockRlockGuard_init(&ea->lock);
+ CELIX_LONG_HASH_MAP_ITERATE(ea->remoteProviderServices, iter) {
+ celix_event_remote_provider_service_t* remoteProvider =
iter.value.ptrValue;
+ if (async) {
+ celix_logHelper_trace(ea->logHelper, "Post event %s to remote
provider.", topic);
+ status = remoteProvider->postEvent(remoteProvider->handle, topic,
remoteProps);
Review Comment:
Is there a risk of deadlock even if a read lock is used here? (In POSIX,
`pthread_rwlock_t` prefer reader by default)
--
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.
To unsubscribe, e-mail: dev-unsubscr...@celix.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
