You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@celix.apache.org by "xuzhenbao (via GitHub)" <gi...@apache.org> on 2024/03/11 14:58:58 UTC
[PR] Event admin (celix)
xuzhenbao opened a new pull request, #738:
URL: https://github.com/apache/celix/pull/738
This PR implements a local event admin and maps celix-framework-event to event-admin-event. See the PR's README.md for details.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option (build_xxx) in turn.
> Is every option enabled by default?
All `build_xxx` options default to be off. This is intentional.
In my day job, we deploy dozens (if not hundreds) of Conan packages together.
All default off together with automatic dependency deduction prove to be very powerful.
The ultimate downstream integrator only has to choose what he/she wants, then will automatically get a minimal system fulfilling his/her needs.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build? Is every option enabled by default?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option in turn.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option (build_xxx) in turn.
> Is every option enabled by default?
All `build_xxx` options default to be off. This is intentional.
In my day job, we deploy dozens (if not hundreds) of Conan packages together.
All default off plug automatic dependency deduction prove to be very powerful.
The ultimate downstream integrator only has to choose what he/she wants, then will automatically get a minimal system fulfilling his/her needs.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "pnoltes (via GitHub)" <gi...@apache.org>.
pnoltes commented on PR #738:
URL: https://github.com/apache/celix/pull/738#issuecomment-1995352717
Very nice addition for Apache Celix :). I had a short look into the event_admin_api and this looks good; Clean api and well documented. I will try to find some time this week and next week to more thoroughly review this.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build? Is every option enabled by default?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option (build_xxx) in turn.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option (build_xxx) in turn.
> Is every option enabled by default?
All `build_xxx` options default to be off. This is intentional.
In my day job, we deploy tens (if not hundreds) of Conan packages together.
All default off together with automatic dependency deduction prove to be very powerful.
The ultimate downstream integrator only has to choose what he/she wants, then will automatically get a minimal system fulfilling his/her needs.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "rlenferink (via GitHub)" <gi...@apache.org>.
rlenferink commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1530838902
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
How does this work for the conan CI build? Is every option enabled by default?
For the non-conan (ubuntu) build, it must be added to here indeed: https://github.com/apache/celix/blob/master/.github/workflows/ubuntu.yml#L143
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1531502865
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
> How does this work for the conan CI build? Is every option enabled by default?
First, we have `build_all` to enable all options. Second, we have "Dependency Deduction Test" in conan_create.yml, in which we enable each option (and its dependency, which is described by `configure` method of the conanfile.py) in turn.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "PengZheng (via GitHub)" <gi...@apache.org>.
PengZheng commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1525718213
##########
bundles/event_admin/CMakeLists.txt:
##########
@@ -0,0 +1,24 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+celix_subproject(EVENT_ADMIN "Option to enable building the Event Admin bundles" OFF)
Review Comment:
With this option default OFF, this module is not available in non-conan CI build.
##########
bundles/event_admin/event_admin_api/include/celix_event_constants.h:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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 CELIX_EVENT_CONSTANTS_H
+#define CELIX_EVENT_CONSTANTS_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @name event constants
+ * @see https://docs.osgi.org/specification/osgi.cmpn/7.0.0/service.event.html#org.osgi.service.event.EventConstants
+ */
+
+/**
+ * @brief The bundle id of the bundle relevant to the event. The type of the value for this event property is Long.
+ */
+#define CELIX_EVENT_BUNDLE_ID "bundle.id"
+
+/**
+ * @brief The bundle symbolic name of the bundle relevant to the event. The type of the value for this event property is String.
+ */
+#define CELIX_EVENT_BUNDLE_SYMBOLICNAME "bundle.symbolicName"
+
+/**
+ * @brief The version of the bundle relevant to the event. The type of the value for this event property is Version.
+ */
+#define CELIX_EVENT_BUNDLE_VERSION "bundle.version"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_ORDERED "async.ordered"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in any order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_UNORDERED "async.unordered"
+
+/**
+ * @brief Service Registration property specifying the delivery qualities requested by an Event Handler service.
+ * @see CELIX_EVENT_DELIVERY_ASYNC_ORDERED,CELIX_EVENT_DELIVERY_ASYNC_UNORDERED
+ */
+#define CELIX_EVENT_DELIVERY "event.delivery"
+
+/**
+ * @brief Service Registration property specifying a filter to further select Event s of interest to an Event Handler service.
+ * The type of the value for this service property is String.
+ */
+#define CELIX_EVENT_FILTER "event.filter"
+
+/**
+ * @brief Service registration property specifying the Event topics of interest to an Event Handler service.
+ *
+ * The Event Handler must use the following grammar to indicate the topics in which it is interested:\n
+ * topic-description := '*' | topic ( '/*' )? \n
+ * topic := token ( '/' token )* \n
+ *
+ * The event handler also can specify multiple topics by providing a comma separated list of topics.
+ *
+ * @example
+ * - "com/acme/*" Indicates all topics beginning with "com/acme/" \n
+ * - "com/acme/MyEvent" Indicates only the specific event which topic is "com/acme/MyEvent" \n
+ * - "*" Indicates all topics \n
+ * - "com/acme/MyEvent,com/acme/MyOtherEvent" Indicates the specific events with topics "com/acme/MyEvent" and "com/acme/MyOtherEvent"
+ *
+ * @todo Use string array instead of comma-separated list.(depends on https://github.com/apache/celix/issues/674)
+ */
+#define CELIX_EVENT_TOPIC "event.topics"
+
+/**
+ * @brief A service's id. The type of the value for this event property is Long.
+ */
+#define CELIX_EVENT_SERVICE_ID "service.id"
+
+/**
+ * @brief A service's objectClass. The type of the value for this event property is String.
+ */
+#define CELIX_EVENT_SERVICE_OBJECTCLASS "service.objectClass"
+
+/**
+ * @brief A service's persistent identity. The type of the value for this event property is String.
+ */
+#define CELIX_EVENT_SERVICE_PID "service.pid"
+
+/**
+ * @brief The time when the event occurred, as reported by the system time in microseconds. The type of the value for this event property is Long.
Review Comment:
Is it microsecond or millisecond? I suspect the former could be useful.
Note that `System.currentTimeMillis()` is a method in Java that returns the current time in milliseconds.
##########
bundles/event_admin/examples/event_publisher/src/celix_example_event_publisher_activator.c:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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 <unistd.h>
+
+#include "celix_event_admin_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+#include "celix_threads.h"
+
+typedef struct celix_event_publisher_example_activator {
+ celix_thread_rwlock_t svcLock;
+ celix_event_admin_service_t *service;
+ long serviceTrkId;
+ bool running;
+ celix_thread_t sendEventThread;
+} celix_event_publisher_example_activator_t;
+
+static void *celix_eventPublisherExampleActivator_sendEventThread(void *handle) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ while(act->running) {
+ celixThreadRwlock_readLock(&act->svcLock);
+ celix_event_admin_service_t *svc = act->service;
+ if (svc != NULL) {
+ {
+ printf("Sending sync event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/syncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->sendEvent(svc->handle, "example/syncEvent", props);
+ }
+ {
+ printf("Sending async event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/asyncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->postEvent(svc->handle, "example/asyncEvent", props);
+ }
+ }
+ celixThreadRwlock_unlock(&act->svcLock);
+ sleep(3);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventPublisherExampleActivator_addEventAdminService(void *handle, void *svc) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ celixThreadRwlock_writeLock(&act->svcLock);
+ act->service = svc;
+ celixThreadRwlock_unlock(&act->svcLock);
+}
+
+celix_status_t celix_eventPublisherExampleActivator_start(celix_event_publisher_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_status_t status = celixThreadRwlock_create(&act->svcLock, NULL);
+ if (status != CELIX_SUCCESS) {
+ return status;
+ }
+ celix_autoptr(celix_thread_rwlock_t) svcLock = &act->svcLock;
+
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.serviceName = CELIX_EVENT_ADMIN_SERVICE_NAME;
+ opts.filter.versionRange = CELIX_EVENT_ADMIN_SERVICE_USE_RANGE;
+ opts.callbackHandle = act;
+ opts.set = celix_eventPublisherExampleActivator_addEventAdminService;
+ act->serviceTrkId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
Review Comment:
Async prefered.
##########
bundles/event_admin/event_admin/README.md:
##########
@@ -0,0 +1,107 @@
+---
+title: Event Admin Service
+---
+
+<!--
+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.
+-->
+
+## Event Admin
+
+The Event Admin provides the pubsub mechanism for in-process communication. It has implemented two delivery mechanisms:
+- Synchronous delivery
+- Asynchronous delivery
+
+In addition, the Event Admin provides an adapter for the celix framework-level event. It maps the celix framework-level event to the
+event admin pubsub model, and events are delivered asynchronously.
+
+### Cmake options
+
+ BUILD_EVENT_ADMIN=OFF
+
+### Conan options
+
+ build_event_admin=false
+
+### Software Design
+
+#### Synchronous Delivery
+
+In the synchronous delivery, the event handler will be notified in the caller's thread. And the `sendEvent` method will return after all the event handlers have been called.
+The diagram of synchronous delivery is shown in the following figure:
+
+
+
+#### Asynchronous Delivery
+
+In the asynchronous delivery, the event handler will be notified in a separate event-delivery thread. And the `postEvent` method will return immediately.
+The diagram of asynchronous delivery is shown in the following figure:
+
+
+
+In the above figure, if the event handler sets the "event.delivery" property to "async.ordered", each event handler can hold
+at most one event-delivery thread at a time, so that events can be delivered in order. If the event handler sets the
+"event.delivery" property to "async.unordered", the event handler can hold multiple event-delivery threads at the same
+time, so that events can be delivered in parallel.
+
+
+#### Event Adapter
+
+In order to present a consistent view of all the events occurring in the system, the Event Admin provides an adapter for the celix framework-level event.
+The events are mapped as follows:
+
+- **Celix Framework Event**
+
+For celix framework Event, we map the celix framework condition id to the event admin event topic. The event admin event properties are none.
+The details mappings are shown in the following table:
+
+| Celix Framework Condition ID | Event Admin Event topic |
+|------------------------------------|---------------------------------------------|
+| CELIX_CONDITION_ID_FRAMEWORK_READY | "org/osgi/framework/FrameworkEvent/STARTED" |
+| CELIX_CONDITION_ID_FRAMEWORK_ERROR | "org/osgi/framework/FrameworkEvent/ERROR" |
+
+- **Celix Bundle Event**
+
+For celix bundle Event, we map the celix bundle event type to the event admin event topic. And The event admin event properties are as follows:
+- bundle.id
Review Comment:
For remote event, this is not enough, framework UUID is needed.
The same remark applies to other events.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
Review Comment:
Missing `celix_logHelper_logTssErrors`.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_status_t status = celix_longHashMap_put(ea->eventHandlers, handler->serviceId, handler);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler(%s).", handler->serviceDescription);
+ return status;
+ }
+
+ CELIX_STRING_HASH_MAP_ITERATE(topicsMap, iter) {
+ celix_eventAdmin_subscribeTopicFor(ea, iter.key, handler);
+ }
+
+ celix_logHelper_debug(ea->logHelper, "Added event handler(%s) for topics %s", handler->serviceDescription, topics);
+
+ celix_steal_ptr(eventFilter);
+ celix_steal_ptr(handler);
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != 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, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_handler_t *handler = celix_longHashMap_get(ea->eventHandlers, serviceId);
+ if (handler != NULL) {
+ celix_logHelper_debug(ea->logHelper, "Removing event handler(%s)", handler->serviceDescription);
+ celix_eventAdmin_unsubscribeTopicFor(ea, handler);
+ celix_longHashMap_remove(ea->eventHandlers, serviceId);
+ celix_filter_destroy(handler->eventFilter);
+ free(handler);
+ }
+
+ return CELIX_SUCCESS;
+}
+
+static void celix_eventAdmin_collectEventHandlers(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ celix_event_channel_t* channel, celix_long_hash_map_t* eventHandlers) {
+ if (channel == NULL) {
+ return;
+ }
+ size_t size = celix_arrayList_size(channel->eventHandlerSvcIdList);
+ for (int i = 0; i < size; ++i) {
+ long handlerSvcId = celix_arrayList_getLong(channel->eventHandlerSvcIdList, i);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ continue;
+ }
+ if (!celix_filter_match(eventHandler->eventFilter, eventProperties)) {
+ celix_logHelper_debug(ea->logHelper, "Event %s is filtered by filter(%s)", eventTopic,
+ celix_filter_getFilterString(eventHandler->eventFilter));
+ continue;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ continue;
+ }
+ if (celix_longHashMap_put(eventHandlers, handlerSvcId, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to collect event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ }
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEvent(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ int (*deliverAction)(celix_event_admin_t* ea, const char* topic, const celix_properties_t* properties,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers)) {
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = celix_longHashMap_create();//avoid duplicated event handler
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list.");
+ return CELIX_ENOMEM;
+ }
+
+ {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ // Add all handlers for matching everything
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, &ea->channelMatchingAllEvents, eventHandlers);
+
+ // Add the handlers for prefix matches
+ if (celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) != 0) {
+ char topicCopy[256] = {0};
+ if (celix_utils_strlen(eventTopic) >= sizeof(topicCopy)) {
+ celix_logHelper_error(ea->logHelper, "Topic %s too long.", eventTopic);
+ } else {
+ strcpy(topicCopy, eventTopic);
+ char *pos;
+ while ((pos = strrchr(topicCopy, '/')) != NULL) {
+ topicCopy[pos - topicCopy] = '\0';
+ celix_event_channel_t *channel = celix_stringHashMap_get(ea->channelsMatchingPrefixTopic, topicCopy);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+ }
+ }
+
+ // Add the handlers for matching topic names
+ celix_event_channel_t* channel = celix_stringHashMap_get(ea->channelsMatchingTopic, eventTopic);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+
+ if (celix_longHashMap_size(eventHandlers) == 0) {//no event handlers found, discard event
+ celix_logHelper_trace(ea->logHelper, "No event handlers found for topic %s", eventTopic);
+ return CELIX_SUCCESS;
+ }
+ bool stealEventHandlers = false;
+ celix_status_t status = deliverAction(ea, eventTopic, eventProperties, eventHandlers, &stealEventHandlers);
+ if (stealEventHandlers) {
+ celix_steal_ptr(eventHandlers);
+ }
+ return status;
+}
+
+static void celix_eventAdmin_deliverEventToHandler(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_event_handler_t* eventHandler) {
+ celix_logHelper_trace(ea->logHelper, "Delivering event %s to handler(%s)", topic, eventHandler->serviceDescription);
+ celix_event_handler_service_t* svc = eventHandler->service;
+ struct timespec startTime = celix_gettime(CLOCK_MONOTONIC);
+ svc->handleEvent(svc->handle, topic, props);
+ double elapsedTime = celix_elapsedtime(CLOCK_MONOTONIC, startTime);
+ if (elapsedTime > CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME) {
+ celix_logHelper_error(ea->logHelper, "Event handler for topic %s took %f seconds, %s", topic, elapsedTime, eventHandler->serviceDescription);
+ __atomic_store_n(&eventHandler->blackListed, true, __ATOMIC_RELEASE);
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEventSyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ CELIX_LONG_HASH_MAP_ITERATE(eventHandlers, iter) {
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, iter.key);
+ if (eventHandler) {
+ celix_eventAdmin_deliverEventToHandler(ea, topic, props, eventHandler);
+ }
+ }
+ *stealEventHandlers = false;
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_sendEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventSyncDo);
+}
+
+static int celix_eventAdmin_deliverEventAsyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_autofree celix_event_entry_t* entry = calloc(1, sizeof(*entry));
+ if (entry == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event entry %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ celix_autoptr(celix_event_t) event = entry->event = celix_event_create(topic, props);
+ if (event == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ entry->eventHandlers = eventHandlers;
+ celix_auto(celix_mutex_lock_guard_t) mutexGuard = celixMutexLockGuard_init(&ea->eventsMutex);
+ if (celix_arrayList_size(ea->asyncEventQueue) >= CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE) {
+ celix_logHelper_error(ea->logHelper, "Event queue is full. Dropping event %s.", topic);
+ return CELIX_ILLEGAL_STATE;
+ }
+ int status = celix_arrayList_add(ea->asyncEventQueue, entry);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event(%s) to event queue.", topic);
+ return status;
+ }
+ *stealEventHandlers = true;
+ celix_steal_ptr(event);
+ celix_steal_ptr(entry);
+ celixThreadCondition_signal(&ea->eventsTriggerCond);
+
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_postEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventAsyncDo);
+}
+
+static bool celix_eventAdmin_getPendingEvent(celix_event_admin_t* ea, celix_event_t** event, long* eventHandlerSvcId) {
+ bool notFound = true;
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; (i < size) && notFound; ++i) {
+ celix_event_entry_t* eventEntry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_long_hash_map_iterator_t iter = celix_longHashMap_begin(eventEntry->eventHandlers);
+ while (!celix_longHashMapIterator_isEnd(&iter) && notFound) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ long handlerSvcId = iter.key;
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ celix_longHashMapIterator_remove(&iter);
+ continue;
+ }
+ unsigned int handlingEventCnt = __atomic_fetch_add(&eventHandler->handlingAsyncEventCnt, 1, __ATOMIC_SEQ_CST);
+ if (handlingEventCnt == 0 || (!eventHandler->asyncOrdered && handlingEventCnt < CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER)) {
+ *event = celix_event_retain(eventEntry->event);
+ *eventHandlerSvcId = handlerSvcId;
+ celix_longHashMapIterator_remove(&iter);
+ notFound = false;
+ continue;
+ }
+ __atomic_fetch_sub(&eventHandler->handlingAsyncEventCnt, 1, __ATOMIC_SEQ_CST);
+
+ celix_longHashMapIterator_next(&iter);
+ }
+ if (celix_longHashMap_size(eventEntry->eventHandlers) == 0) {
+ celix_arrayList_removeAt(ea->asyncEventQueue, i);
+ i--;
+ size--;
+ celix_event_release(eventEntry->event);
+ celix_longHashMap_destroy(eventEntry->eventHandlers);
+ free(eventEntry);
Review Comment:
Entry dtor duplicated.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_status_t status = celix_longHashMap_put(ea->eventHandlers, handler->serviceId, handler);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler(%s).", handler->serviceDescription);
+ return status;
+ }
+
+ CELIX_STRING_HASH_MAP_ITERATE(topicsMap, iter) {
+ celix_eventAdmin_subscribeTopicFor(ea, iter.key, handler);
+ }
+
+ celix_logHelper_debug(ea->logHelper, "Added event handler(%s) for topics %s", handler->serviceDescription, topics);
+
+ celix_steal_ptr(eventFilter);
+ celix_steal_ptr(handler);
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != 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, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_handler_t *handler = celix_longHashMap_get(ea->eventHandlers, serviceId);
+ if (handler != NULL) {
+ celix_logHelper_debug(ea->logHelper, "Removing event handler(%s)", handler->serviceDescription);
+ celix_eventAdmin_unsubscribeTopicFor(ea, handler);
+ celix_longHashMap_remove(ea->eventHandlers, serviceId);
+ celix_filter_destroy(handler->eventFilter);
+ free(handler);
+ }
+
+ return CELIX_SUCCESS;
+}
+
+static void celix_eventAdmin_collectEventHandlers(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ celix_event_channel_t* channel, celix_long_hash_map_t* eventHandlers) {
+ if (channel == NULL) {
+ return;
+ }
+ size_t size = celix_arrayList_size(channel->eventHandlerSvcIdList);
+ for (int i = 0; i < size; ++i) {
+ long handlerSvcId = celix_arrayList_getLong(channel->eventHandlerSvcIdList, i);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ continue;
+ }
+ if (!celix_filter_match(eventHandler->eventFilter, eventProperties)) {
+ celix_logHelper_debug(ea->logHelper, "Event %s is filtered by filter(%s)", eventTopic,
+ celix_filter_getFilterString(eventHandler->eventFilter));
+ continue;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ continue;
+ }
+ if (celix_longHashMap_put(eventHandlers, handlerSvcId, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to collect event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ }
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEvent(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ int (*deliverAction)(celix_event_admin_t* ea, const char* topic, const celix_properties_t* properties,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers)) {
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = celix_longHashMap_create();//avoid duplicated event handler
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list.");
+ return CELIX_ENOMEM;
+ }
+
+ {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ // Add all handlers for matching everything
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, &ea->channelMatchingAllEvents, eventHandlers);
+
+ // Add the handlers for prefix matches
+ if (celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) != 0) {
+ char topicCopy[256] = {0};
+ if (celix_utils_strlen(eventTopic) >= sizeof(topicCopy)) {
+ celix_logHelper_error(ea->logHelper, "Topic %s too long.", eventTopic);
+ } else {
+ strcpy(topicCopy, eventTopic);
+ char *pos;
+ while ((pos = strrchr(topicCopy, '/')) != NULL) {
+ topicCopy[pos - topicCopy] = '\0';
+ celix_event_channel_t *channel = celix_stringHashMap_get(ea->channelsMatchingPrefixTopic, topicCopy);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+ }
+ }
+
+ // Add the handlers for matching topic names
+ celix_event_channel_t* channel = celix_stringHashMap_get(ea->channelsMatchingTopic, eventTopic);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+
+ if (celix_longHashMap_size(eventHandlers) == 0) {//no event handlers found, discard event
+ celix_logHelper_trace(ea->logHelper, "No event handlers found for topic %s", eventTopic);
+ return CELIX_SUCCESS;
+ }
+ bool stealEventHandlers = false;
+ celix_status_t status = deliverAction(ea, eventTopic, eventProperties, eventHandlers, &stealEventHandlers);
+ if (stealEventHandlers) {
+ celix_steal_ptr(eventHandlers);
+ }
+ return status;
+}
+
+static void celix_eventAdmin_deliverEventToHandler(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_event_handler_t* eventHandler) {
+ celix_logHelper_trace(ea->logHelper, "Delivering event %s to handler(%s)", topic, eventHandler->serviceDescription);
+ celix_event_handler_service_t* svc = eventHandler->service;
+ struct timespec startTime = celix_gettime(CLOCK_MONOTONIC);
+ svc->handleEvent(svc->handle, topic, props);
Review Comment:
It might be useful for implementing remote event admin. An local proxy may return `ETIMEOUT` or other errors.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_status_t status = celix_longHashMap_put(ea->eventHandlers, handler->serviceId, handler);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler(%s).", handler->serviceDescription);
+ return status;
+ }
+
+ CELIX_STRING_HASH_MAP_ITERATE(topicsMap, iter) {
+ celix_eventAdmin_subscribeTopicFor(ea, iter.key, handler);
+ }
+
+ celix_logHelper_debug(ea->logHelper, "Added event handler(%s) for topics %s", handler->serviceDescription, topics);
+
+ celix_steal_ptr(eventFilter);
+ celix_steal_ptr(handler);
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != 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, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_handler_t *handler = celix_longHashMap_get(ea->eventHandlers, serviceId);
+ if (handler != NULL) {
+ celix_logHelper_debug(ea->logHelper, "Removing event handler(%s)", handler->serviceDescription);
+ celix_eventAdmin_unsubscribeTopicFor(ea, handler);
+ celix_longHashMap_remove(ea->eventHandlers, serviceId);
+ celix_filter_destroy(handler->eventFilter);
+ free(handler);
+ }
+
+ return CELIX_SUCCESS;
+}
+
+static void celix_eventAdmin_collectEventHandlers(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ celix_event_channel_t* channel, celix_long_hash_map_t* eventHandlers) {
+ if (channel == NULL) {
+ return;
+ }
+ size_t size = celix_arrayList_size(channel->eventHandlerSvcIdList);
+ for (int i = 0; i < size; ++i) {
+ long handlerSvcId = celix_arrayList_getLong(channel->eventHandlerSvcIdList, i);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ continue;
+ }
+ if (!celix_filter_match(eventHandler->eventFilter, eventProperties)) {
+ celix_logHelper_debug(ea->logHelper, "Event %s is filtered by filter(%s)", eventTopic,
+ celix_filter_getFilterString(eventHandler->eventFilter));
+ continue;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ continue;
+ }
+ if (celix_longHashMap_put(eventHandlers, handlerSvcId, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to collect event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ }
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEvent(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ int (*deliverAction)(celix_event_admin_t* ea, const char* topic, const celix_properties_t* properties,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers)) {
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = celix_longHashMap_create();//avoid duplicated event handler
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list.");
+ return CELIX_ENOMEM;
+ }
+
+ {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ // Add all handlers for matching everything
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, &ea->channelMatchingAllEvents, eventHandlers);
+
+ // Add the handlers for prefix matches
+ if (celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) != 0) {
+ char topicCopy[256] = {0};
+ if (celix_utils_strlen(eventTopic) >= sizeof(topicCopy)) {
+ celix_logHelper_error(ea->logHelper, "Topic %s too long.", eventTopic);
+ } else {
+ strcpy(topicCopy, eventTopic);
+ char *pos;
+ while ((pos = strrchr(topicCopy, '/')) != NULL) {
+ topicCopy[pos - topicCopy] = '\0';
+ celix_event_channel_t *channel = celix_stringHashMap_get(ea->channelsMatchingPrefixTopic, topicCopy);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+ }
+ }
+
+ // Add the handlers for matching topic names
+ celix_event_channel_t* channel = celix_stringHashMap_get(ea->channelsMatchingTopic, eventTopic);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+
+ if (celix_longHashMap_size(eventHandlers) == 0) {//no event handlers found, discard event
+ celix_logHelper_trace(ea->logHelper, "No event handlers found for topic %s", eventTopic);
+ return CELIX_SUCCESS;
+ }
+ bool stealEventHandlers = false;
+ celix_status_t status = deliverAction(ea, eventTopic, eventProperties, eventHandlers, &stealEventHandlers);
+ if (stealEventHandlers) {
+ celix_steal_ptr(eventHandlers);
+ }
+ return status;
+}
+
+static void celix_eventAdmin_deliverEventToHandler(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_event_handler_t* eventHandler) {
+ celix_logHelper_trace(ea->logHelper, "Delivering event %s to handler(%s)", topic, eventHandler->serviceDescription);
+ celix_event_handler_service_t* svc = eventHandler->service;
+ struct timespec startTime = celix_gettime(CLOCK_MONOTONIC);
+ svc->handleEvent(svc->handle, topic, props);
+ double elapsedTime = celix_elapsedtime(CLOCK_MONOTONIC, startTime);
+ if (elapsedTime > CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME) {
+ celix_logHelper_error(ea->logHelper, "Event handler for topic %s took %f seconds, %s", topic, elapsedTime, eventHandler->serviceDescription);
+ __atomic_store_n(&eventHandler->blackListed, true, __ATOMIC_RELEASE);
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEventSyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ CELIX_LONG_HASH_MAP_ITERATE(eventHandlers, iter) {
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, iter.key);
+ if (eventHandler) {
+ celix_eventAdmin_deliverEventToHandler(ea, topic, props, eventHandler);
+ }
+ }
+ *stealEventHandlers = false;
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_sendEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventSyncDo);
+}
+
+static int celix_eventAdmin_deliverEventAsyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_autofree celix_event_entry_t* entry = calloc(1, sizeof(*entry));
+ if (entry == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event entry %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ celix_autoptr(celix_event_t) event = entry->event = celix_event_create(topic, props);
+ if (event == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ entry->eventHandlers = eventHandlers;
+ celix_auto(celix_mutex_lock_guard_t) mutexGuard = celixMutexLockGuard_init(&ea->eventsMutex);
+ if (celix_arrayList_size(ea->asyncEventQueue) >= CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE) {
+ celix_logHelper_error(ea->logHelper, "Event queue is full. Dropping event %s.", topic);
+ return CELIX_ILLEGAL_STATE;
+ }
+ int status = celix_arrayList_add(ea->asyncEventQueue, entry);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event(%s) to event queue.", topic);
+ return status;
+ }
+ *stealEventHandlers = true;
+ celix_steal_ptr(event);
+ celix_steal_ptr(entry);
+ celixThreadCondition_signal(&ea->eventsTriggerCond);
+
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_postEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventAsyncDo);
+}
+
+static bool celix_eventAdmin_getPendingEvent(celix_event_admin_t* ea, celix_event_t** event, long* eventHandlerSvcId) {
+ bool notFound = true;
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; (i < size) && notFound; ++i) {
+ celix_event_entry_t* eventEntry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_long_hash_map_iterator_t iter = celix_longHashMap_begin(eventEntry->eventHandlers);
+ while (!celix_longHashMapIterator_isEnd(&iter) && notFound) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ long handlerSvcId = iter.key;
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ celix_longHashMapIterator_remove(&iter);
+ continue;
+ }
+ unsigned int handlingEventCnt = __atomic_fetch_add(&eventHandler->handlingAsyncEventCnt, 1, __ATOMIC_SEQ_CST);
+ if (handlingEventCnt == 0 || (!eventHandler->asyncOrdered && handlingEventCnt < CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER)) {
+ *event = celix_event_retain(eventEntry->event);
+ *eventHandlerSvcId = handlerSvcId;
+ celix_longHashMapIterator_remove(&iter);
+ notFound = false;
+ continue;
+ }
+ __atomic_fetch_sub(&eventHandler->handlingAsyncEventCnt, 1, __ATOMIC_SEQ_CST);
+
+ celix_longHashMapIterator_next(&iter);
+ }
+ if (celix_longHashMap_size(eventEntry->eventHandlers) == 0) {
+ celix_arrayList_removeAt(ea->asyncEventQueue, i);
+ i--;
+ size--;
+ celix_event_release(eventEntry->event);
+ celix_longHashMap_destroy(eventEntry->eventHandlers);
+ free(eventEntry);
+ }
+ }
+ return !notFound;
+}
+
+static void celix_eventAdmin_deliverPendingEvent(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props, long eventHandlerSvcId) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, eventHandlerSvcId);
+ if(eventHandler == NULL) {
+ return;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", topic,
+ eventHandler->serviceDescription);
Review Comment:
Missing `__atomic_fetch_sub(&eventHandler->handlingAsyncEventCnt, 1, __ATOMIC_SEQ_CST)` ?
##########
bundles/event_admin/event_admin_api/include/celix_event_admin_service.h:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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 CELIX_EVENT_ADMIN_SERVICE_H
+#define CELIX_EVENT_ADMIN_SERVICE_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "celix_properties.h"
+#include "celix_errno.h"
+
+/**
+ * @brief The Event Admin service name
+ */
+#define CELIX_EVENT_ADMIN_SERVICE_NAME "org.osgi.service.event.EventAdmin"
+
+/**
+ * @brief The Event Admin service version
+ */
+#define CELIX_EVENT_ADMIN_SERVICE_VERSION "1.0.0"
+#define CELIX_EVENT_ADMIN_SERVICE_USE_RANGE "[1.0.0,2)"
+
+/**
+ * @brief The Event Admin service
+ * @see https://osgi.org/specification/osgi.cmpn/7.0.0/service.event.html#org.osgi.service.event.EventAdmin
+ */
+typedef struct celix_event_admin_service {
+ void* handle;
+ /**
+ * @brief Asynchronous event delivery . This method returns to the caller before delivery of the event is completed.
+ * @param[in] handle The handle as provided by the service registration.
+ * @param[in] topic The topic of the event.
+ * @param[in] properties The properties of the event. It can be NULL.
+ * @return Status code indicating failure or success. CELIX_SUCCESS if no errors are encountered. If an error is encountered, it should be return celix errno.
+ */
+ celix_status_t (*postEvent)(void* handle, const char* topic, const celix_properties_t* properties);
+ /**
+ * @brief Synchronous event delivery. This method does not return to the caller until delivery of the event is completed.
+ * @param[in] handle The handle as provided by the service registration.
+ * @param[in] topic The topic of the event.
+ * @param[in] properties The properties of the event. It can be NULL.
+ * @return Status code indicating failure or success. CELIX_SUCCESS if no errors are encountered. If an error is encountered, it should be return celix errno.
Review Comment:
```suggestion
* @return Status code indicating failure or success. CELIX_SUCCESS if no errors are encountered. If an error is encountered, it should be a celix errno.
```
##########
bundles/event_admin/event_admin_api/include/celix_event_constants.h:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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 CELIX_EVENT_CONSTANTS_H
+#define CELIX_EVENT_CONSTANTS_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @name event constants
+ * @see https://docs.osgi.org/specification/osgi.cmpn/7.0.0/service.event.html#org.osgi.service.event.EventConstants
+ */
+
+/**
+ * @brief The bundle id of the bundle relevant to the event. The type of the value for this event property is Long.
+ */
+#define CELIX_EVENT_BUNDLE_ID "bundle.id"
+
+/**
+ * @brief The bundle symbolic name of the bundle relevant to the event. The type of the value for this event property is String.
+ */
+#define CELIX_EVENT_BUNDLE_SYMBOLICNAME "bundle.symbolicName"
+
+/**
+ * @brief The version of the bundle relevant to the event. The type of the value for this event property is Version.
+ */
+#define CELIX_EVENT_BUNDLE_VERSION "bundle.version"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_ORDERED "async.ordered"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in any order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_UNORDERED "async.unordered"
+
+/**
+ * @brief Service Registration property specifying the delivery qualities requested by an Event Handler service.
+ * @see CELIX_EVENT_DELIVERY_ASYNC_ORDERED,CELIX_EVENT_DELIVERY_ASYNC_UNORDERED
+ */
+#define CELIX_EVENT_DELIVERY "event.delivery"
+
+/**
+ * @brief Service Registration property specifying a filter to further select Event s of interest to an Event Handler service.
+ * The type of the value for this service property is String.
+ */
+#define CELIX_EVENT_FILTER "event.filter"
+
+/**
+ * @brief Service registration property specifying the Event topics of interest to an Event Handler service.
+ *
+ * The Event Handler must use the following grammar to indicate the topics in which it is interested:\n
+ * topic-description := '*' | topic ( '/*' )? \n
+ * topic := token ( '/' token )* \n
+ *
+ * The event handler also can specify multiple topics by providing a comma separated list of topics.
+ *
+ * @example
+ * - "com/acme/*" Indicates all topics beginning with "com/acme/" \n
Review Comment:
```suggestion
* - "com/acme/‍*" Indicates all topics beginning with "com/acme/" \n
```
See above readability improvement.
##########
bundles/event_admin/event_admin_api/include/celix_event_admin_service.h:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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 CELIX_EVENT_ADMIN_SERVICE_H
+#define CELIX_EVENT_ADMIN_SERVICE_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "celix_properties.h"
+#include "celix_errno.h"
+
+/**
+ * @brief The Event Admin service name
+ */
+#define CELIX_EVENT_ADMIN_SERVICE_NAME "org.osgi.service.event.EventAdmin"
+
+/**
+ * @brief The Event Admin service version
+ */
+#define CELIX_EVENT_ADMIN_SERVICE_VERSION "1.0.0"
+#define CELIX_EVENT_ADMIN_SERVICE_USE_RANGE "[1.0.0,2)"
+
+/**
+ * @brief The Event Admin service
+ * @see https://osgi.org/specification/osgi.cmpn/7.0.0/service.event.html#org.osgi.service.event.EventAdmin
+ */
+typedef struct celix_event_admin_service {
+ void* handle;
+ /**
+ * @brief Asynchronous event delivery . This method returns to the caller before delivery of the event is completed.
+ * @param[in] handle The handle as provided by the service registration.
+ * @param[in] topic The topic of the event.
+ * @param[in] properties The properties of the event. It can be NULL.
+ * @return Status code indicating failure or success. CELIX_SUCCESS if no errors are encountered. If an error is encountered, it should be return celix errno.
Review Comment:
```suggestion
* @return Status code indicating failure or success. CELIX_SUCCESS if no errors are encountered. If an error is encountered, it should be a celix errno.
```
##########
bundles/event_admin/examples/event_handler/src/celix_example_event_handler_activator.c:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <stdio.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+
+typedef struct celix_event_handler_example_activator {
+ celix_event_handler_service_t service;
+ long serviceId;
+} celix_event_handler_example_activator_t;
+
+static celix_status_t celix_eventHandlerExampleActivator_handleEvent(void *handle, const char *topic, const celix_properties_t *properties) {
+ (void)handle;
+ (void)properties;
+ printf("Received event with topic %s\n\n", topic);
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventHandlerExampleActivator_start(celix_event_handler_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_properties_t *properties = celix_properties_create();
+ celix_properties_set(properties, CELIX_EVENT_TOPIC, "example/*");
+ celix_properties_set(properties, CELIX_FRAMEWORK_SERVICE_VERSION, CELIX_EVENT_HANDLER_SERVICE_VERSION);
+ act->service.handle = act;
+ act->service.handleEvent = celix_eventHandlerExampleActivator_handleEvent;
+ act->serviceId = celix_bundleContext_registerService(ctx, &act->service, CELIX_EVENT_HANDLER_SERVICE_NAME, properties);
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventHandlerExampleActivator_stop(celix_event_handler_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_bundleContext_unregisterService(ctx, act->serviceId);
Review Comment:
Sync APIs do harm to program's startup speed, and thus they are discouraged. I suggest new code use Async one instead unless absolutely necessary (for example testing).
Note that this remark applies to several places.
##########
bundles/event_admin/examples/event_publisher/src/celix_example_event_publisher_activator.c:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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 <unistd.h>
+
+#include "celix_event_admin_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+#include "celix_threads.h"
+
+typedef struct celix_event_publisher_example_activator {
+ celix_thread_rwlock_t svcLock;
+ celix_event_admin_service_t *service;
+ long serviceTrkId;
+ bool running;
+ celix_thread_t sendEventThread;
+} celix_event_publisher_example_activator_t;
+
+static void *celix_eventPublisherExampleActivator_sendEventThread(void *handle) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ while(act->running) {
+ celixThreadRwlock_readLock(&act->svcLock);
+ celix_event_admin_service_t *svc = act->service;
+ if (svc != NULL) {
+ {
+ printf("Sending sync event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/syncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->sendEvent(svc->handle, "example/syncEvent", props);
+ }
+ {
+ printf("Sending async event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/asyncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->postEvent(svc->handle, "example/asyncEvent", props);
+ }
+ }
+ celixThreadRwlock_unlock(&act->svcLock);
+ sleep(3);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventPublisherExampleActivator_addEventAdminService(void *handle, void *svc) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ celixThreadRwlock_writeLock(&act->svcLock);
+ act->service = svc;
+ celixThreadRwlock_unlock(&act->svcLock);
+}
+
+celix_status_t celix_eventPublisherExampleActivator_start(celix_event_publisher_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_status_t status = celixThreadRwlock_create(&act->svcLock, NULL);
+ if (status != CELIX_SUCCESS) {
+ return status;
+ }
+ celix_autoptr(celix_thread_rwlock_t) svcLock = &act->svcLock;
+
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.serviceName = CELIX_EVENT_ADMIN_SERVICE_NAME;
+ opts.filter.versionRange = CELIX_EVENT_ADMIN_SERVICE_USE_RANGE;
+ opts.callbackHandle = act;
+ opts.set = celix_eventPublisherExampleActivator_addEventAdminService;
+ act->serviceTrkId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
+ if (act->serviceTrkId < 0) {
+ return CELIX_BUNDLE_EXCEPTION;
+ }
+
+ act->running = true;
+ status = celixThread_create(&act->sendEventThread, NULL, celix_eventPublisherExampleActivator_sendEventThread, act);
+ if (status != CELIX_SUCCESS) {
+ celix_bundleContext_stopTracker(ctx, act->serviceTrkId);
Review Comment:
Async prefered.
##########
bundles/event_admin/event_admin_api/include/celix_event_constants.h:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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 CELIX_EVENT_CONSTANTS_H
+#define CELIX_EVENT_CONSTANTS_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/**
+ * @name event constants
+ * @see https://docs.osgi.org/specification/osgi.cmpn/7.0.0/service.event.html#org.osgi.service.event.EventConstants
+ */
+
+/**
+ * @brief The bundle id of the bundle relevant to the event. The type of the value for this event property is Long.
+ */
+#define CELIX_EVENT_BUNDLE_ID "bundle.id"
+
+/**
+ * @brief The bundle symbolic name of the bundle relevant to the event. The type of the value for this event property is String.
+ */
+#define CELIX_EVENT_BUNDLE_SYMBOLICNAME "bundle.symbolicName"
+
+/**
+ * @brief The version of the bundle relevant to the event. The type of the value for this event property is Version.
+ */
+#define CELIX_EVENT_BUNDLE_VERSION "bundle.version"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_ORDERED "async.ordered"
+
+/**
+ * @brief Event handler delivery quality value specifying the Event Handler requires asynchronously delivered events be delivered in any order.
+ * @see CELIX_EVENT_DELIVERY
+ */
+#define CELIX_EVENT_DELIVERY_ASYNC_UNORDERED "async.unordered"
+
+/**
+ * @brief Service Registration property specifying the delivery qualities requested by an Event Handler service.
+ * @see CELIX_EVENT_DELIVERY_ASYNC_ORDERED,CELIX_EVENT_DELIVERY_ASYNC_UNORDERED
+ */
+#define CELIX_EVENT_DELIVERY "event.delivery"
+
+/**
+ * @brief Service Registration property specifying a filter to further select Event s of interest to an Event Handler service.
+ * The type of the value for this service property is String.
+ */
+#define CELIX_EVENT_FILTER "event.filter"
+
+/**
+ * @brief Service registration property specifying the Event topics of interest to an Event Handler service.
+ *
+ * The Event Handler must use the following grammar to indicate the topics in which it is interested:\n
+ * topic-description := '*' | topic ( '/*' )? \n
Review Comment:
```suggestion
* topic-description := '*' | topic ( '/‍*' )? \n
```
Though `∗ ` also works, I believe the suggested way is more readable in the source code form (at lease we have a explicit '*' there). And both are better than `*`, which is completely unreadable in the source form.
Check https://www.doxygen.nl/manual/htmlcmds.html#htmltagcmds for more html commands.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
Review Comment:
I guess our users would be interested to adjust some of the above configuration?
Shall we make some of the above configurable at build time or run time?
If so, what do you think are the most interesting ones?
Are there any missing configuration options (like scheduling policy of event dilivery)?
Of course, we can delay this question until real need arises.
##########
bundles/event_admin/examples/event_handler/src/celix_example_event_handler_activator.c:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <stdio.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+
+typedef struct celix_event_handler_example_activator {
+ celix_event_handler_service_t service;
+ long serviceId;
+} celix_event_handler_example_activator_t;
+
+static celix_status_t celix_eventHandlerExampleActivator_handleEvent(void *handle, const char *topic, const celix_properties_t *properties) {
+ (void)handle;
+ (void)properties;
+ printf("Received event with topic %s\n\n", topic);
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventHandlerExampleActivator_start(celix_event_handler_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_properties_t *properties = celix_properties_create();
+ celix_properties_set(properties, CELIX_EVENT_TOPIC, "example/*");
+ celix_properties_set(properties, CELIX_FRAMEWORK_SERVICE_VERSION, CELIX_EVENT_HANDLER_SERVICE_VERSION);
+ act->service.handle = act;
+ act->service.handleEvent = celix_eventHandlerExampleActivator_handleEvent;
+ act->serviceId = celix_bundleContext_registerService(ctx, &act->service, CELIX_EVENT_HANDLER_SERVICE_NAME, properties);
Review Comment:
Async prefered.
##########
bundles/event_admin/examples/event_publisher/src/celix_example_event_publisher_activator.c:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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 <unistd.h>
+
+#include "celix_event_admin_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+#include "celix_threads.h"
+
+typedef struct celix_event_publisher_example_activator {
+ celix_thread_rwlock_t svcLock;
+ celix_event_admin_service_t *service;
+ long serviceTrkId;
+ bool running;
+ celix_thread_t sendEventThread;
+} celix_event_publisher_example_activator_t;
+
+static void *celix_eventPublisherExampleActivator_sendEventThread(void *handle) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ while(act->running) {
+ celixThreadRwlock_readLock(&act->svcLock);
+ celix_event_admin_service_t *svc = act->service;
+ if (svc != NULL) {
+ {
+ printf("Sending sync event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/syncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->sendEvent(svc->handle, "example/syncEvent", props);
+ }
+ {
+ printf("Sending async event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/asyncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->postEvent(svc->handle, "example/asyncEvent", props);
+ }
+ }
+ celixThreadRwlock_unlock(&act->svcLock);
+ sleep(3);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventPublisherExampleActivator_addEventAdminService(void *handle, void *svc) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ celixThreadRwlock_writeLock(&act->svcLock);
+ act->service = svc;
+ celixThreadRwlock_unlock(&act->svcLock);
+}
+
+celix_status_t celix_eventPublisherExampleActivator_start(celix_event_publisher_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_status_t status = celixThreadRwlock_create(&act->svcLock, NULL);
+ if (status != CELIX_SUCCESS) {
+ return status;
+ }
+ celix_autoptr(celix_thread_rwlock_t) svcLock = &act->svcLock;
+
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.serviceName = CELIX_EVENT_ADMIN_SERVICE_NAME;
+ opts.filter.versionRange = CELIX_EVENT_ADMIN_SERVICE_USE_RANGE;
+ opts.callbackHandle = act;
+ opts.set = celix_eventPublisherExampleActivator_addEventAdminService;
Review Comment:
Misleading name. It is "set" not "add".
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_status_t status = celix_longHashMap_put(ea->eventHandlers, handler->serviceId, handler);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler(%s).", handler->serviceDescription);
+ return status;
+ }
+
+ CELIX_STRING_HASH_MAP_ITERATE(topicsMap, iter) {
+ celix_eventAdmin_subscribeTopicFor(ea, iter.key, handler);
+ }
+
+ celix_logHelper_debug(ea->logHelper, "Added event handler(%s) for topics %s", handler->serviceDescription, topics);
+
+ celix_steal_ptr(eventFilter);
+ celix_steal_ptr(handler);
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != 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, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_handler_t *handler = celix_longHashMap_get(ea->eventHandlers, serviceId);
+ if (handler != NULL) {
+ celix_logHelper_debug(ea->logHelper, "Removing event handler(%s)", handler->serviceDescription);
+ celix_eventAdmin_unsubscribeTopicFor(ea, handler);
+ celix_longHashMap_remove(ea->eventHandlers, serviceId);
+ celix_filter_destroy(handler->eventFilter);
+ free(handler);
+ }
+
+ return CELIX_SUCCESS;
+}
+
+static void celix_eventAdmin_collectEventHandlers(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ celix_event_channel_t* channel, celix_long_hash_map_t* eventHandlers) {
+ if (channel == NULL) {
+ return;
+ }
+ size_t size = celix_arrayList_size(channel->eventHandlerSvcIdList);
+ for (int i = 0; i < size; ++i) {
+ long handlerSvcId = celix_arrayList_getLong(channel->eventHandlerSvcIdList, i);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ continue;
+ }
+ if (!celix_filter_match(eventHandler->eventFilter, eventProperties)) {
+ celix_logHelper_debug(ea->logHelper, "Event %s is filtered by filter(%s)", eventTopic,
+ celix_filter_getFilterString(eventHandler->eventFilter));
+ continue;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ continue;
+ }
+ if (celix_longHashMap_put(eventHandlers, handlerSvcId, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to collect event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ }
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEvent(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ int (*deliverAction)(celix_event_admin_t* ea, const char* topic, const celix_properties_t* properties,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers)) {
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = celix_longHashMap_create();//avoid duplicated event handler
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list.");
+ return CELIX_ENOMEM;
+ }
+
+ {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ // Add all handlers for matching everything
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, &ea->channelMatchingAllEvents, eventHandlers);
+
+ // Add the handlers for prefix matches
+ if (celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) != 0) {
+ char topicCopy[256] = {0};
+ if (celix_utils_strlen(eventTopic) >= sizeof(topicCopy)) {
+ celix_logHelper_error(ea->logHelper, "Topic %s too long.", eventTopic);
+ } else {
+ strcpy(topicCopy, eventTopic);
+ char *pos;
+ while ((pos = strrchr(topicCopy, '/')) != NULL) {
+ topicCopy[pos - topicCopy] = '\0';
+ celix_event_channel_t *channel = celix_stringHashMap_get(ea->channelsMatchingPrefixTopic, topicCopy);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+ }
+ }
+
+ // Add the handlers for matching topic names
+ celix_event_channel_t* channel = celix_stringHashMap_get(ea->channelsMatchingTopic, eventTopic);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+
+ if (celix_longHashMap_size(eventHandlers) == 0) {//no event handlers found, discard event
+ celix_logHelper_trace(ea->logHelper, "No event handlers found for topic %s", eventTopic);
+ return CELIX_SUCCESS;
+ }
+ bool stealEventHandlers = false;
+ celix_status_t status = deliverAction(ea, eventTopic, eventProperties, eventHandlers, &stealEventHandlers);
+ if (stealEventHandlers) {
+ celix_steal_ptr(eventHandlers);
+ }
+ return status;
+}
+
+static void celix_eventAdmin_deliverEventToHandler(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_event_handler_t* eventHandler) {
+ celix_logHelper_trace(ea->logHelper, "Delivering event %s to handler(%s)", topic, eventHandler->serviceDescription);
+ celix_event_handler_service_t* svc = eventHandler->service;
+ struct timespec startTime = celix_gettime(CLOCK_MONOTONIC);
+ svc->handleEvent(svc->handle, topic, props);
Review Comment:
If you don't care about return value, why bother adding it to the service interface?
##########
bundles/event_admin/examples/event_publisher/src/celix_example_event_publisher_activator.c:
##########
@@ -0,0 +1,104 @@
+/*
+ * 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 <unistd.h>
+
+#include "celix_event_admin_service.h"
+#include "celix_event_constants.h"
+#include "celix_bundle_activator.h"
+#include "celix_threads.h"
+
+typedef struct celix_event_publisher_example_activator {
+ celix_thread_rwlock_t svcLock;
+ celix_event_admin_service_t *service;
+ long serviceTrkId;
+ bool running;
+ celix_thread_t sendEventThread;
+} celix_event_publisher_example_activator_t;
+
+static void *celix_eventPublisherExampleActivator_sendEventThread(void *handle) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ while(act->running) {
+ celixThreadRwlock_readLock(&act->svcLock);
+ celix_event_admin_service_t *svc = act->service;
+ if (svc != NULL) {
+ {
+ printf("Sending sync event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/syncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->sendEvent(svc->handle, "example/syncEvent", props);
+ }
+ {
+ printf("Sending async event\n");
+ celix_autoptr(celix_properties_t) props = celix_properties_create();
+ celix_properties_set(props, CELIX_EVENT_TOPIC, "example/asyncEvent");
+ celix_properties_set(props, "example", "data");
+ svc->postEvent(svc->handle, "example/asyncEvent", props);
+ }
+ }
+ celixThreadRwlock_unlock(&act->svcLock);
+ sleep(3);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventPublisherExampleActivator_addEventAdminService(void *handle, void *svc) {
+ celix_event_publisher_example_activator_t *act = (celix_event_publisher_example_activator_t *)handle;
+ celixThreadRwlock_writeLock(&act->svcLock);
+ act->service = svc;
+ celixThreadRwlock_unlock(&act->svcLock);
+}
+
+celix_status_t celix_eventPublisherExampleActivator_start(celix_event_publisher_example_activator_t *act, celix_bundle_context_t *ctx) {
+ celix_status_t status = celixThreadRwlock_create(&act->svcLock, NULL);
+ if (status != CELIX_SUCCESS) {
+ return status;
+ }
+ celix_autoptr(celix_thread_rwlock_t) svcLock = &act->svcLock;
+
+ celix_service_tracking_options_t opts = CELIX_EMPTY_SERVICE_TRACKING_OPTIONS;
+ opts.filter.serviceName = CELIX_EVENT_ADMIN_SERVICE_NAME;
+ opts.filter.versionRange = CELIX_EVENT_ADMIN_SERVICE_USE_RANGE;
+ opts.callbackHandle = act;
+ opts.set = celix_eventPublisherExampleActivator_addEventAdminService;
+ act->serviceTrkId = celix_bundleContext_trackServicesWithOptions(ctx, &opts);
+ if (act->serviceTrkId < 0) {
+ return CELIX_BUNDLE_EXCEPTION;
+ }
+
+ act->running = true;
+ status = celixThread_create(&act->sendEventThread, NULL, celix_eventPublisherExampleActivator_sendEventThread, act);
+ if (status != CELIX_SUCCESS) {
+ celix_bundleContext_stopTracker(ctx, act->serviceTrkId);
+ return status;
+ }
+ celix_steal_ptr(svcLock);
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventPublisherExampleActivator_stop(celix_event_publisher_example_activator_t *act, celix_bundle_context_t *ctx) {
+ act->running = false;
+ celixThread_join(act->sendEventThread, NULL);
+ celix_bundleContext_stopTracker(ctx, act->serviceTrkId);
Review Comment:
Async prefered.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
Review Comment:
`celix_arrayList_createWithOptions` with entry dtor will make above unnecessary.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
Review Comment:
It seems that `celix_arrayList_createWithOptions` together with dtor of `celix_event_entry_t` will recude code duplication.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
Review Comment:
Missing `celix_logHelper_logTssErrors`.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
Review Comment:
This doe not comply with the spec:
> If the filter is an error, then the Event Admin service should log a warning and further ignore the Event Handler.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
Review Comment:
IIRC, it will log to celix_err. We'd better improvement documentation of `celix_arrayList` about this, and flush err to log helper.
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,632 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_create();
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
+ }
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_stop(celix_event_admin_t* ea) {
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++i) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return CELIX_SUCCESS;
+}
+
+void celix_eventAdmin_destroy(celix_event_admin_t* ea) {
+ assert(ea != NULL);
+ size_t size = celix_arrayList_size(ea->asyncEventQueue);
+ for (int i = 0; i < size; ++i) {
+ celix_event_entry_t *entry = celix_arrayList_get(ea->asyncEventQueue, i);
+ celix_event_release(entry->event);
+ celix_longHashMap_destroy(entry->eventHandlers);
+ free(entry);
+ }
+ celix_arrayList_destroy(ea->asyncEventQueue);
+ celixThreadCondition_destroy(&ea->eventsTriggerCond);
+ celixThreadMutex_destroy(&ea->eventsMutex);
+ assert(celix_longHashMap_size(ea->eventHandlers) == 0);
+ celix_longHashMap_destroy(ea->eventHandlers);
+ assert(celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingPrefixTopic);
+ assert(celix_stringHashMap_size(ea->channelsMatchingTopic) == 0);
+ celix_stringHashMap_destroy(ea->channelsMatchingTopic);
+ celix_arrayList_destroy(ea->channelMatchingAllEvents.eventHandlerSvcIdList);
+ celixThreadRwlock_destroy(&ea->lock);
+ celix_logHelper_destroy(ea->logHelper);
+ free(ea);
+ return;
+}
+
+static void celix_eventAdmin_addEventHandlerToChannels(celix_event_admin_t* ea, const char* topic,
+ celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ celix_event_channel_t* channel = celix_stringHashMap_get(channels, topic);
+ if (channel == NULL) {
+ celix_autofree celix_event_channel_t *_channel = channel = calloc(1, sizeof(*channel));
+ if (_channel == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event channel for topic %s", topic);
+ return;
+ }
+ celix_autoptr(celix_array_list_t) eventHandlerSvcIdList = channel->eventHandlerSvcIdList = celix_arrayList_create();
+ if (eventHandlerSvcIdList == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list for topic %s", topic);
+ return;
+ }
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ if (celix_stringHashMap_put(channels, topic, channel) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event channel for topic %s", topic);
+ return;
+ }
+ celix_steal_ptr(eventHandlerSvcIdList);
+ celix_steal_ptr(_channel);
+ } else if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ }
+ return;
+}
+
+static void celix_eventAdmin_subscribeTopicFor(celix_event_admin_t* ea, const char* topic, celix_event_handler_t* eventHandler) {
+ size_t topicLen = celix_utils_strlen(topic);
+ if (celix_utils_stringEquals(topic, "*")) {
+ celix_event_channel_t *channel = &ea->channelMatchingAllEvents;
+ if (celix_arrayList_addLong(channel->eventHandlerSvcIdList, eventHandler->serviceId) != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to subscribe %s for event handler(%s).", topic, eventHandler->serviceDescription);
+ return;
+ }
+ } else if (topicLen > 2 && topic[topicLen-1] == '*' && topic[topicLen-2] == '/'){
+ char prefixTopic[256] = {0};
+ if (topicLen-2 >= sizeof(prefixTopic)) {
+ celix_logHelper_error(ea->logHelper, "Prefix topic %s too long.", topic);
+ return;
+ }
+ memcpy(prefixTopic, topic, topicLen - 2);
+ prefixTopic[topicLen - 2] = '\0';
+ celix_eventAdmin_addEventHandlerToChannels(ea, prefixTopic, eventHandler, ea->channelsMatchingPrefixTopic);
+ } else {
+ celix_eventAdmin_addEventHandlerToChannels(ea, topic, eventHandler, ea->channelsMatchingTopic);
+ }
+ return;
+}
+
+static void celix_eventAdmin_removeEventHandlerFromChannels(celix_event_admin_t* ea, celix_event_handler_t* eventHandler, celix_string_hash_map_t* channels) {
+ (void)ea;//unused
+ celix_string_hash_map_iterator_t iter = celix_stringHashMap_begin(channels);
+ while (!celix_stringHashMapIterator_isEnd(&iter)) {
+ celix_event_channel_t *channel = iter.value.ptrValue;
+ celix_arrayList_removeLong(channel->eventHandlerSvcIdList, eventHandler->serviceId);
+ if (celix_arrayList_size(channel->eventHandlerSvcIdList) == 0) {
+ celix_stringHashMapIterator_remove(&iter);
+ celix_arrayList_destroy(channel->eventHandlerSvcIdList);
+ free(channel);
+ } else {
+ celix_stringHashMapIterator_next(&iter);
+ }
+ }
+}
+
+static void celix_eventAdmin_unsubscribeTopicFor(celix_event_admin_t* ea, celix_event_handler_t* eventHandler) {
+ celix_arrayList_removeLong(ea->channelMatchingAllEvents.eventHandlerSvcIdList, eventHandler->serviceId);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingPrefixTopic);
+ celix_eventAdmin_removeEventHandlerFromChannels(ea, eventHandler, ea->channelsMatchingTopic);
+ return;
+}
+
+int celix_eventAdmin_addEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != NULL);
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ const char *topics = celix_properties_get(props, CELIX_EVENT_TOPIC, NULL);
+ if (topics == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler. No %s property set", CELIX_EVENT_TOPIC);
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ long serviceId = celix_properties_getAsLong(props, CELIX_FRAMEWORK_SERVICE_ID, -1L);
+ if (serviceId < 0) {
+ celix_logHelper_error(ea->logHelper, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_autofree celix_event_handler_t *handler = calloc(1, sizeof(*handler));
+ if (handler == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler.");
+ return CELIX_ENOMEM;
+ }
+ handler->service = svc;
+ handler->serviceId = serviceId;
+ handler->serviceDescription = celix_properties_get(props, CELIX_FRAMEWORK_SERVICE_DESCRIPTION, "Unknown");
+ const char *deliver = celix_properties_get(props, CELIX_EVENT_DELIVERY, NULL);
+ handler->asyncOrdered = (deliver == NULL || strstr(deliver,CELIX_EVENT_DELIVERY_ASYNC_ORDERED) != NULL);
+ handler->eventFilter = NULL;
+ handler->blackListed = false;
+ handler->handlingAsyncEventCnt = 0;
+
+ celix_autofree char* topicsCopy = celix_utils_strdup(topics);
+ if (topicsCopy == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to dup topics %s.", topics);
+ return CELIX_ENOMEM;
+ }
+
+ celix_autoptr(celix_string_hash_map_t) topicsMap = celix_stringHashMap_create();//avoid topic duplicates
+ if (topicsMap == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create topics map.");
+ return CELIX_ENOMEM;
+ }
+ char* savePtr = NULL;
+ char* token = strtok_r(topicsCopy, ",", &savePtr);
+ while (token != NULL) {
+ char *trimmedToken = celix_utils_trimInPlace(token);
+ if (celix_stringHashMap_put(topicsMap, trimmedToken, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add topic %s to topics map.", trimmedToken);
+ return CELIX_ENOMEM;
+ }
+ token = strtok_r(NULL, ",", &savePtr);
+ }
+
+ celix_autoptr(celix_filter_t) eventFilter = NULL;
+ const char* eventFilterStr = celix_properties_get(props, CELIX_EVENT_FILTER, NULL);
+ if (eventFilterStr) {
+ eventFilter = handler->eventFilter = celix_filter_create(eventFilterStr);
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_status_t status = celix_longHashMap_put(ea->eventHandlers, handler->serviceId, handler);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to add event handler(%s).", handler->serviceDescription);
+ return status;
+ }
+
+ CELIX_STRING_HASH_MAP_ITERATE(topicsMap, iter) {
+ celix_eventAdmin_subscribeTopicFor(ea, iter.key, handler);
+ }
+
+ celix_logHelper_debug(ea->logHelper, "Added event handler(%s) for topics %s", handler->serviceDescription, topics);
+
+ celix_steal_ptr(eventFilter);
+ celix_steal_ptr(handler);
+
+ return CELIX_SUCCESS;
+}
+
+int celix_eventAdmin_removeEventHandlerWithProperties(void* handle, void* svc, const celix_properties_t* props) {
+ assert(handle != NULL);
+ assert(svc != NULL);
+ assert(props != 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, "Event handler service id is empty.");
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+
+ celix_auto(celix_rwlock_wlock_guard_t) wLockGuard = celixRwlockWlockGuard_init(&ea->lock);
+ celix_event_handler_t *handler = celix_longHashMap_get(ea->eventHandlers, serviceId);
+ if (handler != NULL) {
+ celix_logHelper_debug(ea->logHelper, "Removing event handler(%s)", handler->serviceDescription);
+ celix_eventAdmin_unsubscribeTopicFor(ea, handler);
+ celix_longHashMap_remove(ea->eventHandlers, serviceId);
+ celix_filter_destroy(handler->eventFilter);
+ free(handler);
+ }
+
+ return CELIX_SUCCESS;
+}
+
+static void celix_eventAdmin_collectEventHandlers(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ celix_event_channel_t* channel, celix_long_hash_map_t* eventHandlers) {
+ if (channel == NULL) {
+ return;
+ }
+ size_t size = celix_arrayList_size(channel->eventHandlerSvcIdList);
+ for (int i = 0; i < size; ++i) {
+ long handlerSvcId = celix_arrayList_getLong(channel->eventHandlerSvcIdList, i);
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, handlerSvcId);
+ if (eventHandler == NULL) {
+ continue;
+ }
+ if (!celix_filter_match(eventHandler->eventFilter, eventProperties)) {
+ celix_logHelper_debug(ea->logHelper, "Event %s is filtered by filter(%s)", eventTopic,
+ celix_filter_getFilterString(eventHandler->eventFilter));
+ continue;
+ }
+ if (__atomic_load_n(&eventHandler->blackListed, __ATOMIC_ACQUIRE)) {
+ celix_logHelper_warning(ea->logHelper, "Skipping blacklisted event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ continue;
+ }
+ if (celix_longHashMap_put(eventHandlers, handlerSvcId, NULL) != CELIX_SUCCESS) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to collect event handler for topic %s, %s", eventTopic,
+ eventHandler->serviceDescription);
+ }
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEvent(celix_event_admin_t* ea, const char* eventTopic, const celix_properties_t* eventProperties,
+ int (*deliverAction)(celix_event_admin_t* ea, const char* topic, const celix_properties_t* properties,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers)) {
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = celix_longHashMap_create();//avoid duplicated event handler
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(ea->logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(ea->logHelper, "Failed to create event handlers list.");
+ return CELIX_ENOMEM;
+ }
+
+ {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ // Add all handlers for matching everything
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, &ea->channelMatchingAllEvents, eventHandlers);
+
+ // Add the handlers for prefix matches
+ if (celix_stringHashMap_size(ea->channelsMatchingPrefixTopic) != 0) {
+ char topicCopy[256] = {0};
+ if (celix_utils_strlen(eventTopic) >= sizeof(topicCopy)) {
+ celix_logHelper_error(ea->logHelper, "Topic %s too long.", eventTopic);
+ } else {
+ strcpy(topicCopy, eventTopic);
+ char *pos;
+ while ((pos = strrchr(topicCopy, '/')) != NULL) {
+ topicCopy[pos - topicCopy] = '\0';
+ celix_event_channel_t *channel = celix_stringHashMap_get(ea->channelsMatchingPrefixTopic, topicCopy);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+ }
+ }
+
+ // Add the handlers for matching topic names
+ celix_event_channel_t* channel = celix_stringHashMap_get(ea->channelsMatchingTopic, eventTopic);
+ celix_eventAdmin_collectEventHandlers(ea, eventTopic, eventProperties, channel, eventHandlers);
+ }
+
+ if (celix_longHashMap_size(eventHandlers) == 0) {//no event handlers found, discard event
+ celix_logHelper_trace(ea->logHelper, "No event handlers found for topic %s", eventTopic);
+ return CELIX_SUCCESS;
+ }
+ bool stealEventHandlers = false;
+ celix_status_t status = deliverAction(ea, eventTopic, eventProperties, eventHandlers, &stealEventHandlers);
+ if (stealEventHandlers) {
+ celix_steal_ptr(eventHandlers);
+ }
+ return status;
+}
+
+static void celix_eventAdmin_deliverEventToHandler(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_event_handler_t* eventHandler) {
+ celix_logHelper_trace(ea->logHelper, "Delivering event %s to handler(%s)", topic, eventHandler->serviceDescription);
+ celix_event_handler_service_t* svc = eventHandler->service;
+ struct timespec startTime = celix_gettime(CLOCK_MONOTONIC);
+ svc->handleEvent(svc->handle, topic, props);
+ double elapsedTime = celix_elapsedtime(CLOCK_MONOTONIC, startTime);
+ if (elapsedTime > CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME) {
+ celix_logHelper_error(ea->logHelper, "Event handler for topic %s took %f seconds, %s", topic, elapsedTime, eventHandler->serviceDescription);
+ __atomic_store_n(&eventHandler->blackListed, true, __ATOMIC_RELEASE);
+ }
+ return;
+}
+
+static int celix_eventAdmin_deliverEventSyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_auto(celix_rwlock_rlock_guard_t) rLockGuard = celixRwlockRlockGuard_init(&ea->lock);
+ CELIX_LONG_HASH_MAP_ITERATE(eventHandlers, iter) {
+ celix_event_handler_t* eventHandler = celix_longHashMap_get(ea->eventHandlers, iter.key);
+ if (eventHandler) {
+ celix_eventAdmin_deliverEventToHandler(ea, topic, props, eventHandler);
+ }
+ }
+ *stealEventHandlers = false;
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_sendEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventSyncDo);
+}
+
+static int celix_eventAdmin_deliverEventAsyncDo(celix_event_admin_t* ea, const char* topic, const celix_properties_t* props,
+ celix_long_hash_map_t* eventHandlers, bool* stealEventHandlers) {
+ celix_autofree celix_event_entry_t* entry = calloc(1, sizeof(*entry));
+ if (entry == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event entry %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ celix_autoptr(celix_event_t) event = entry->event = celix_event_create(topic, props);
+ if (event == NULL) {
+ celix_logHelper_error(ea->logHelper, "Failed to alloc memory for event %s.", topic);
+ return CELIX_ENOMEM;
+ }
+ entry->eventHandlers = eventHandlers;
+ celix_auto(celix_mutex_lock_guard_t) mutexGuard = celixMutexLockGuard_init(&ea->eventsMutex);
+ if (celix_arrayList_size(ea->asyncEventQueue) >= CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE) {
+ celix_logHelper_error(ea->logHelper, "Event queue is full. Dropping event %s.", topic);
+ return CELIX_ILLEGAL_STATE;
+ }
+ int status = celix_arrayList_add(ea->asyncEventQueue, entry);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to add event(%s) to event queue.", topic);
+ return status;
+ }
+ *stealEventHandlers = true;
+ celix_steal_ptr(event);
+ celix_steal_ptr(entry);
+ celixThreadCondition_signal(&ea->eventsTriggerCond);
+
+ return CELIX_SUCCESS;
+}
+
+celix_status_t celix_eventAdmin_postEvent(void* handle, const char* topic, const celix_properties_t* props) {
+ if (handle == NULL || topic == NULL) {
+ return CELIX_ILLEGAL_ARGUMENT;
+ }
+ celix_event_admin_t* ea = (celix_event_admin_t*)handle;
+ return celix_eventAdmin_deliverEvent(ea, topic, props, celix_eventAdmin_deliverEventAsyncDo);
+}
+
+static bool celix_eventAdmin_getPendingEvent(celix_event_admin_t* ea, celix_event_t** event, long* eventHandlerSvcId) {
+ bool notFound = true;
Review Comment:
Double negation impairs readability. `found` will be way better.
##########
bundles/event_admin/event_admin/src/celix_event_admin_activator.c:
##########
@@ -0,0 +1,106 @@
+/*
+ * 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 <assert.h>
+#include <errno.h>
+
+#include "celix_errno.h"
+#include "celix_bundle_activator.h"
+#include "celix_event_admin.h"
+#include "celix_event_adapter.h"
+#include "celix_event_admin_service.h"
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+
+typedef struct celix_event_admin_activator {
+ celix_event_admin_t *eventAdmin;
+ celix_event_admin_service_t eventAdminService;
+ celix_event_adapter_t *eventAdapter;
+} celix_event_admin_activator_t;
+
+celix_status_t celix_eventAdminActivator_start(celix_event_admin_activator_t *act, celix_bundle_context_t *ctx) {
+ assert(act != NULL);
+ assert(ctx != NULL);
+
+ celix_autoptr(celix_dm_component_t) adminCmp = celix_dmComponent_create(ctx, "EVENT_ADMIN_CMP");
+ if (adminCmp == NULL) {
+ return CELIX_ENOMEM;
+ }
+
+ act->eventAdmin = celix_eventAdmin_create(ctx);
+ if (act->eventAdmin == NULL) {
+ assert(errno != 0);
+ return CELIX_ERROR_MAKE(CELIX_FACILITY_CERRNO, errno);
+ }
+ celix_dmComponent_setImplementation(adminCmp, act->eventAdmin);
+ CELIX_DM_COMPONENT_SET_CALLBACKS(adminCmp, celix_event_admin_t, NULL, celix_eventAdmin_start, celix_eventAdmin_stop, NULL);
+ CELIX_DM_COMPONENT_SET_IMPLEMENTATION_DESTROY_FUNCTION(adminCmp, celix_event_admin_t, celix_eventAdmin_destroy);
+
+ celix_dm_service_dependency_t *eventHandlerDep = celix_dmServiceDependency_create();
+ if (eventHandlerDep == NULL) {
+ return CELIX_ENOMEM;
+ }
+ celix_dmServiceDependency_setService(eventHandlerDep, CELIX_EVENT_HANDLER_SERVICE_NAME, CELIX_EVENT_HANDLER_SERVICE_USE_RANGE, "("CELIX_EVENT_TOPIC"=*)");//Event Handlers which have not specified the EVENT_TOPIC service property must not receive events.
+ celix_dmServiceDependency_setStrategy(eventHandlerDep, DM_SERVICE_DEPENDENCY_STRATEGY_LOCKING);
+ celix_dm_service_dependency_callback_options_t opts = CELIX_EMPTY_DM_SERVICE_DEPENDENCY_CALLBACK_OPTIONS;
+ opts.addWithProps = celix_eventAdmin_addEventHandlerWithProperties;
+ opts.removeWithProps = celix_eventAdmin_removeEventHandlerWithProperties;
+ celix_dmServiceDependency_setCallbacksWithOptions(eventHandlerDep, &opts);
+ celix_dmComponent_addServiceDependency(adminCmp, eventHandlerDep);
+
+ act->eventAdminService.handle = act->eventAdmin;
+ act->eventAdminService.postEvent = celix_eventAdmin_postEvent;
+ act->eventAdminService.sendEvent = celix_eventAdmin_sendEvent;
+ celix_dmComponent_addInterface(adminCmp, CELIX_EVENT_ADMIN_SERVICE_NAME, CELIX_EVENT_ADMIN_SERVICE_VERSION, &act->eventAdminService, NULL);
Review Comment:
Nice to know every single error is handled, except this one. `celix_dmComponent_addInterface` can fail.
Of course, there are still some flaws, such as `celix_dependencyManager_addAsync`, which does not handle `celix_arrayList_add` error. But that is other component's flaw.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #738:
URL: https://github.com/apache/celix/pull/738#issuecomment-1988668070
## [Codecov](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `99.07834%` with `6 lines` in your changes are missing coverage. Please review.
> Project coverage is 89.89%. Comparing base [(`d46d983`)](https://app.codecov.io/gh/apache/celix/commit/d46d98383fb60b498415bfa5c0edc5c4f8e17bad?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`eb58535`)](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
| [Files](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [.../event\_admin/event\_admin/src/celix\_event\_adapter.c](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnVuZGxlcy9ldmVudF9hZG1pbi9ldmVudF9hZG1pbi9zcmMvY2VsaXhfZXZlbnRfYWRhcHRlci5j) | 97.56% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...es/event\_admin/event\_admin/src/celix\_event\_admin.c](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnVuZGxlcy9ldmVudF9hZG1pbi9ldmVudF9hZG1pbi9zcmMvY2VsaXhfZXZlbnRfYWRtaW4uYw==) | 99.50% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #738 +/- ##
==========================================
+ Coverage 89.58% 89.89% +0.31%
==========================================
Files 216 221 +5
Lines 25413 26064 +651
==========================================
+ Hits 22765 23431 +666
+ Misses 2648 2633 -15
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/celix/pull/738?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "pnoltes (via GitHub)" <gi...@apache.org>.
pnoltes commented on code in PR #738:
URL: https://github.com/apache/celix/pull/738#discussion_r1534513381
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,652 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+//Belows parameters are not configurable, consider its configurability until a real need arises.
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+static void onAsyncEventQueueRemoved(void* value);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_array_list_create_options_t opts = CELIX_EMPTY_ARRAY_LIST_CREATE_OPTIONS;
+ opts.elementType = CELIX_ARRAY_LIST_ELEMENT_TYPE_POINTER;
+ opts.simpleRemovedCallback = onAsyncEventQueueRemoved;
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_createWithOptions(&opts);
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
+ status = celixThread_create(&ea->eventHandlerThreads[thCnt], NULL, celix_eventAdmin_deliverEventThread, ea);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(ea->logHelper, "Failed to create event handler thread.");
+ celixThreadMutex_lock(&ea->eventsMutex);
+ ea->threadsRunning = false;
+ celixThreadMutex_unlock(&ea->eventsMutex);
+ celixThreadCondition_broadcast(&ea->eventsTriggerCond);
+ for (int i = 0; i < thCnt; i++) {
+ celixThread_join(ea->eventHandlerThreads[i], NULL);
+ }
+ return status;
+ }
+ celixThread_setName(&ea->eventHandlerThreads[thCnt], "EventHandler");
Review Comment:
Good that the threads have a name, The Celix Event is named "CelixEvent".
Maybe this could be named "EventAdmin-1", "EventAdmin-2", etc. Still shorts (< 16 chars) but more specific for the event admin.
##########
bundles/event_admin/event_admin_api/include/celix_event_admin_service.h:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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 CELIX_EVENT_ADMIN_SERVICE_H
+#define CELIX_EVENT_ADMIN_SERVICE_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+#include "celix_properties.h"
+#include "celix_errno.h"
+
+/**
+ * @brief The Event Admin service name
+ */
+#define CELIX_EVENT_ADMIN_SERVICE_NAME "org.osgi.service.event.EventAdmin"
Review Comment:
Although Apache Celix is inspired by OSGi and we did use some OSGi named constants, enums etc.
I currently think it is better not to use the "org.osgi" or "osgi" part and replace that with "celix".
So in this case "celix.service.event.EventAdmin"
##########
bundles/event_admin/event_admin/src/celix_event_admin.c:
##########
@@ -0,0 +1,652 @@
+/*
+ * 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 "celix_event_admin.h"
+
+#include <string.h>
+#include <stdbool.h>
+#include <assert.h>
+#include <errno.h>
+
+#include "celix_event_handler_service.h"
+#include "celix_event_constants.h"
+#include "celix_event.h"
+#include "celix_log_helper.h"
+#include "celix_string_hash_map.h"
+#include "celix_long_hash_map.h"
+#include "celix_filter.h"
+#include "celix_constants.h"
+#include "celix_threads.h"
+#include "celix_utils.h"
+#include "celix_stdlib_cleanup.h"
+
+//Belows parameters are not configurable, consider its configurability until a real need arises.
+#define CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS 5
+#define CELIX_EVENT_ADMIN_MAX_PARALLEL_EVENTS_OF_HANDLER (CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS/3 + 1) //max parallel async event for a single handler
+#define CELIX_EVENT_ADMIN_MAX_HANDLE_EVENT_TIME 60 //seconds
+#define CELIX_EVENT_ADMIN_MAX_EVENT_QUEUE_SIZE 512 //events
+
+typedef struct celix_event_handler {
+ celix_event_handler_service_t* service;
+ long serviceId;
+ const char* serviceDescription;
+ bool asyncOrdered;
+ celix_filter_t* eventFilter;
+ bool blackListed;//Blacklisted handlers must not be notified of any events.
+ unsigned int handlingAsyncEventCnt;
+}celix_event_handler_t;
+
+typedef struct celix_event_channel {
+ celix_array_list_t* eventHandlerSvcIdList;
+}celix_event_channel_t;
+
+typedef struct celix_event_entry {
+ celix_event_t* event;
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: null
+}celix_event_entry_t;
+
+struct celix_event_admin {
+ celix_bundle_context_t* ctx;
+ celix_log_helper_t* logHelper;
+ celix_thread_rwlock_t lock;//projects: channels,eventHandlers
+ celix_event_channel_t channelMatchingAllEvents;
+ celix_string_hash_map_t* channelsMatchingTopic; //key: topic, value: celix_event_channel_t *
+ celix_string_hash_map_t* channelsMatchingPrefixTopic;//key:prefix topic, value: celix_event_channel_t *
+ celix_long_hash_map_t* eventHandlers;//key: event handler service id, value: celix_event_handler_t*
+ celix_thread_mutex_t eventsMutex;// protect belows
+ celix_thread_cond_t eventsTriggerCond;
+ celix_array_list_t* asyncEventQueue;//array_list<celix_event_entry_t*>
+ bool threadsRunning;
+ celix_thread_t eventHandlerThreads[CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS];
+};
+
+static void* celix_eventAdmin_deliverEventThread(void* data);
+static void onAsyncEventQueueRemoved(void* value);
+
+celix_event_admin_t* celix_eventAdmin_create(celix_bundle_context_t* ctx) {
+ celix_autofree celix_event_admin_t* ea = calloc(1, sizeof(*ea));
+ if (ea == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+ ea->ctx = ctx;
+ ea->threadsRunning = false;
+
+ celix_autoptr(celix_log_helper_t) logHelper = ea->logHelper = celix_logHelper_create(ctx, "CelixEventAdmin");
+ if (logHelper == NULL) {
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_status_t status = celixThreadRwlock_create(&ea->lock, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin lock.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_rwlock_t) lock = &ea->lock;
+ celix_autoptr(celix_array_list_t) channelMatchingAllEvents = ea->channelMatchingAllEvents.eventHandlerSvcIdList = celix_arrayList_create();
+ if (channelMatchingAllEvents == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channel matching all events.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingTopic = ea->channelsMatchingTopic = celix_stringHashMap_create();
+ if (channelsMatchingTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_string_hash_map_t) channelsMatchingPrefixTopic = ea->channelsMatchingPrefixTopic = celix_stringHashMap_create();
+ if (channelsMatchingPrefixTopic == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event channels matching prefix topic.");
+ errno = ENOMEM;
+ return NULL;
+ }
+ celix_autoptr(celix_long_hash_map_t) eventHandlers = ea->eventHandlers = celix_longHashMap_create();
+ if (eventHandlers == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create event handler map.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ status = celixThreadMutex_create(&ea->eventsMutex, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events mutex.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_mutex_t) mutex = &ea->eventsMutex;
+ status = celixThreadCondition_init(&ea->eventsTriggerCond, NULL);
+ if (status != CELIX_SUCCESS) {
+ celix_logHelper_error(logHelper, "Failed to create event admin events not empty condition.");
+ errno = status;//The status be ensured to belong to CELIX_FACILITY_CERRNO
+ return NULL;
+ }
+ celix_autoptr(celix_thread_cond_t) cond = &ea->eventsTriggerCond;
+
+ celix_array_list_create_options_t opts = CELIX_EMPTY_ARRAY_LIST_CREATE_OPTIONS;
+ opts.elementType = CELIX_ARRAY_LIST_ELEMENT_TYPE_POINTER;
+ opts.simpleRemovedCallback = onAsyncEventQueueRemoved;
+ celix_autoptr(celix_array_list_t) asyncEventQueue = ea->asyncEventQueue = celix_arrayList_createWithOptions(&opts);
+ if (asyncEventQueue == NULL) {
+ celix_logHelper_logTssErrors(logHelper, CELIX_LOG_LEVEL_ERROR);
+ celix_logHelper_error(logHelper, "Failed to create async event queue.");
+ errno = ENOMEM;
+ return NULL;
+ }
+
+ celix_steal_ptr(asyncEventQueue);
+ celix_steal_ptr(cond);
+ celix_steal_ptr(mutex);
+ celix_steal_ptr(eventHandlers);
+ celix_steal_ptr(channelsMatchingPrefixTopic);
+ celix_steal_ptr(channelsMatchingTopic);
+ celix_steal_ptr(channelMatchingAllEvents);
+ celix_steal_ptr(lock);
+ celix_steal_ptr(logHelper);
+
+ return celix_steal_ptr(ea);
+}
+
+int celix_eventAdmin_start(celix_event_admin_t* ea) {
+ celix_status_t status = CELIX_SUCCESS;
+ ea->threadsRunning = true;
+ for (int thCnt = 0; thCnt < CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS; ++thCnt) {
Review Comment:
nitpick:
The name of `CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS` implies that the event admin spins up a maximum of CELIX_EVENT_ADMIN_MAX_HANDLER_THREADS threads, but these threads are always created (in the happy flow).
I think the name should imply this, something like: `CELIX_EVENT_ADMIN_HANDLER_THREADS`.
I also think this is a good candidate for a configuration parameter, so that users can lower or increase the event handlers threads based on their usage.
##########
bundles/event_admin/examples/event_publisher/src/celix_example_event_publisher_activator.c:
##########
Review Comment:
nice examples. Shows the simplicity (in usage) of the event admin.
##########
bundles/event_admin/event_admin/README.md:
##########
@@ -0,0 +1,107 @@
+---
+title: Event Admin Service
+---
+
+<!--
+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.
+-->
+
+## Event Admin
+
+The Event Admin provides the pubsub mechanism for in-process communication. It has implemented two delivery mechanisms:
+- Synchronous delivery
+- Asynchronous delivery
+
+In addition, the Event Admin provides an adapter for the celix framework-level event. It maps the celix framework-level event to the
+event admin pubsub model, and events are delivered asynchronously.
+
+### Cmake options
+
+ BUILD_EVENT_ADMIN=OFF
+
+### Conan options
+
+ build_event_admin=false
+
+### Software Design
+
+#### Synchronous Delivery
+
+In the synchronous delivery, the event handler will be notified in the caller's thread. And the `sendEvent` method will return after all the event handlers have been called.
+The diagram of synchronous delivery is shown in the following figure:
+
+
+
+#### Asynchronous Delivery
+
+In the asynchronous delivery, the event handler will be notified in a separate event-delivery thread. And the `postEvent` method will return immediately.
+The diagram of asynchronous delivery is shown in the following figure:
+
+
+
+In the above figure, if the event handler sets the "event.delivery" property to "async.ordered", each event handler can hold
+at most one event-delivery thread at a time, so that events can be delivered in order. If the event handler sets the
+"event.delivery" property to "async.unordered", the event handler can hold multiple event-delivery threads at the same
+time, so that events can be delivered in parallel.
+
+
+#### Event Adapter
+
+In order to present a consistent view of all the events occurring in the system, the Event Admin provides an adapter for the celix framework-level event.
+The events are mapped as follows:
+
+- **Celix Framework Event**
+
+For celix framework Event, we map the celix framework condition id to the event admin event topic. The event admin event properties are none.
+The details mappings are shown in the following table:
+
+| Celix Framework Condition ID | Event Admin Event topic |
+|------------------------------------|---------------------------------------------|
+| CELIX_CONDITION_ID_FRAMEWORK_READY | "org/osgi/framework/FrameworkEvent/STARTED" |
+| CELIX_CONDITION_ID_FRAMEWORK_ERROR | "org/osgi/framework/FrameworkEvent/ERROR" |
+
+- **Celix Bundle Event**
+
+For celix bundle Event, we map the celix bundle event type to the event admin event topic. And The event admin event properties are as follows:
+- bundle.id
+- bundle.symbolicName
+- bundle.version
+
+The details mappings are shown in the following table:
+
+| Celix Bundle Event Type | Event Admin Event topic |
+|--------------------------------|----------------------------------------------|
+| CELIX_BUNDLE_EVENT_INSTALLED | "org/osgi/framework/BundleEvent/INSTALLED" |
Review Comment:
Also mentioned for the service names, but I prefer that "org.osgi" or "osgi" is not used (inspired by, but not a implementation of).
So I would prefer topics like "celix/framework/BundleEvent/INSTALLED
##########
bundles/event_admin/event_admin/README.md:
##########
@@ -0,0 +1,107 @@
+---
+title: Event Admin Service
+---
+
+<!--
+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.
+-->
+
+## Event Admin
+
+The Event Admin provides the pubsub mechanism for in-process communication. It has implemented two delivery mechanisms:
+- Synchronous delivery
+- Asynchronous delivery
+
+In addition, the Event Admin provides an adapter for the celix framework-level event. It maps the celix framework-level event to the
+event admin pubsub model, and events are delivered asynchronously.
+
+### Cmake options
+
+ BUILD_EVENT_ADMIN=OFF
+
+### Conan options
+
+ build_event_admin=false
+
+### Software Design
+
+#### Synchronous Delivery
+
+In the synchronous delivery, the event handler will be notified in the caller's thread. And the `sendEvent` method will return after all the event handlers have been called.
+The diagram of synchronous delivery is shown in the following figure:
+
+
+
+#### Asynchronous Delivery
+
+In the asynchronous delivery, the event handler will be notified in a separate event-delivery thread. And the `postEvent` method will return immediately.
+The diagram of asynchronous delivery is shown in the following figure:
+
+
+
+In the above figure, if the event handler sets the "event.delivery" property to "async.ordered", each event handler can hold
+at most one event-delivery thread at a time, so that events can be delivered in order. If the event handler sets the
+"event.delivery" property to "async.unordered", the event handler can hold multiple event-delivery threads at the same
+time, so that events can be delivered in parallel.
+
+
+#### Event Adapter
+
+In order to present a consistent view of all the events occurring in the system, the Event Admin provides an adapter for the celix framework-level event.
+The events are mapped as follows:
+
+- **Celix Framework Event**
+
+For celix framework Event, we map the celix framework condition id to the event admin event topic. The event admin event properties are none.
+The details mappings are shown in the following table:
+
+| Celix Framework Condition ID | Event Admin Event topic |
+|------------------------------------|---------------------------------------------|
+| CELIX_CONDITION_ID_FRAMEWORK_READY | "org/osgi/framework/FrameworkEvent/STARTED" |
Review Comment:
Good choice and prevents additional needs to exposing a additional framework header.
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Event admin (celix)
Posted by "xuzhenbao (via GitHub)" <gi...@apache.org>.
xuzhenbao merged PR #738:
URL: https://github.com/apache/celix/pull/738
--
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-unsubscribe@celix.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org