You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@celix.apache.org by GitBox <gi...@apache.org> on 2021/04/09 09:44:37 UTC
[GitHub] [celix] pnoltes opened a new pull request #336: Feature/async update
pnoltes opened a new pull request #336:
URL: https://github.com/apache/celix/pull/336
Mainly updates C++ async remote service admin so that this is more based on the use of services as shared_ptr instead raw pointers.
Also:
- updates the ServiceDependency of depenency manager to support callbacks with shared pointers. For this the same setup is used as in the celix::ServiceTracker, so using shared_ptr as a way to monitor usage of the service.
- Refactors the add callback functions for celix::ServiceTrackerBuilder to use function template argument (F&&) instead of std::function to create more flexibility.
- extends and updates the rsa service provider interface (spi). Among other this now has a IExportServiceFactory and IImportServiceFactory interface which is more opaque than the currenlty implemented version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on pull request #336:
URL: https://github.com/apache/celix/pull/336#issuecomment-833610466
Note also update the github ci configuration to use ubuntu 20.04 instead of 18.04.
Because for ubuntu 18.04 the install of libcmzq is not working anymore -> resulting in failed builds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618362878
##########
File path: libs/framework/include/celix/BundleActivator.h
##########
@@ -84,9 +82,9 @@ namespace celix {
celix_status_t destroyActivator(void *userData) {
auto *data = static_cast<BundleActivatorData<I> *>(userData);
std::weak_ptr<celix::BundleContext> ctx = data->ctx;
- std::weak_ptr<celix::dm::DependencyManager> dm = data->dm;
+ std::weak_ptr<celix::dm::DependencyManager> dm = data->ctx->getDependencyManager();
auto bndId = data->bndId;
- data->dm->clear();
+ data->ctx->getDependencyManager()->clear();
Review comment:
Nitpick: `dm->clear();` would be clearer here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618353987
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
- return;
- }
+ std::lock_guard l(_m);
- std::unique_lock l(_m);
+ //remove export interface (if present)
auto instanceIt = _exportedServiceInstances.find(svcId);
-
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
+}
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
+void celix::async_rsa::AsyncAdmin::createImportServices() {
+ for (auto it = _toBeImportedServices.begin(); it != _toBeImportedServices.end(); ++it) {
+ auto interface = (*it)->getExportedInterfaces();
+ auto existingFactory = _importedServiceFactories.find(interface);
+ if (existingFactory == end(_importedServiceFactories)) {
+ L_DEBUG("Adding endpoint to be imported but no factory available yet, delaying import");
Review comment:
missing `continue` statement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r610665854
##########
File path: bundles/async_remote_services/rsa_spi/include/celix/rsa/IExportServiceFactory.h
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+#pragma once
+
+#include <memory>
+
+namespace celix::rsa {
+
+ /**
+ * @brief ExportService class which represent a (opaque) exported service.
+ * The lifetime of this object should be coupled with the lifetime of the exported service.
+ */
+ class ExportedService {
+ public:
+ virtual ~ExportedService() noexcept = default;
+
+ /**
+ * @brief returns the endpoint for this exported service (to be used in discovery).
+ */
+ virtual const celix::rsa::Endpoint& getEndpoint() const = 0;
+ };
+
+ /**
+ * @brief A export service factory for a specific service type.
+ *
+ * The service type which this export service factory targets is provided with
+ * the mandatory celix::rsa::IExportServiceFactory::TARGET_SERVICE_NAME service property.
+ *
+ */
+ class IExportServiceFactory {
Review comment:
- Correct. The factory knows the type and can get the svc id from the service properties, as result it can create a ExportedService which a service dependency to the service.
- If I am correct the current setup is that an exported service is marked using a IExportedService interface. So the ExportServiceFactory creates a ExportService which is a component with a publish, subscriber, dependency on the "service to export" and a IExportedService provided interface.
The IExportedService services can be tracked by the topology manager or discovery manager so that it can be announced in the network. The async admin does not need to know this, it only has to keep the std::unique_ptr<ExportService> as long as the "service to export" and the Factory which created the ExportService is tracked/available.
Note that for now the "announce in network" is not relevant; configured discovery only handles imports.
- Yes, exactly. And an IMO additional benefit is that the 'dm' overview will also cleary show which service is exported by a ExportService component.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627575321
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
Review comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627495742
##########
File path: libs/framework/include/celix/dm/ServiceDependency_Impl.h
##########
@@ -403,22 +531,64 @@ void ServiceDependency<T,I>::setupCallbacks() {
int(*cadd)(void*, void *, const celix_properties_t*) {nullptr};
int(*crem)(void*, void *, const celix_properties_t*) {nullptr};
- if (setFp != nullptr) {
- cset = [](void* handle, void *service, const celix_properties_t* props) -> int {
+ if (setFp || setFpUsingSharedPtr) {
+ cset = [](void* handle, void* rawSvc, const celix_properties_t* rawProps) -> int {
+ int rc = 0;
auto dep = (ServiceDependency<T,I>*) handle;
- return dep->invokeCallback(dep->setFp, props, service);
+ if (dep->setFp) {
+ rc = dep->invokeCallback(dep->setFp, rawProps, rawSvc);
+ }
+ if (dep->setFpUsingSharedPtr) {
+ auto svcId = dep->setService.second ? dep->setService.second->getAsLong(celix::SERVICE_ID, -1) : -1;
+ std::weak_ptr<I> currentSvc = dep->setService.first;
+ std::weak_ptr<const celix::Properties> currentProps = dep->setService.second;
+ auto svc = std::shared_ptr<I>{static_cast<I*>(rawSvc), [](I*){/*nop*/}};
+ auto props = rawProps ? celix::Properties::wrap(rawProps) : nullptr;
+ dep->setService = std::make_pair(std::move(svc), std::move(props));
+ dep->setFpUsingSharedPtr(dep->setService.first, dep->setService.second);
+ dep->waitForExpired(currentSvc, svcId, "service pointer");
Review comment:
Yes. If a shared_ptr is not expired we can have a bug (removal is not correctly done) or the svc is still in use on an other thread (also not desirable).
Note that this can also happen with raw pointers, but in the case of shared_ptr a warning can be printed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] codecov-commenter commented on pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #336:
URL: https://github.com/apache/celix/pull/336#issuecomment-833618338
# [Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#336](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ee23f57) into [master](https://codecov.io/gh/apache/celix/commit/39a3d2e794e4583da4c884a4b316688907035a8e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (39a3d2e) will **decrease** coverage by `0.00%`.
> The diff coverage is `91.15%`.
[](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #336 +/- ##
==========================================
- Coverage 70.91% 70.91% -0.01%
==========================================
Files 186 183 -3
Lines 34614 34624 +10
==========================================
+ Hits 24546 24552 +6
- Misses 10068 10072 +4
```
| [Impacted Files](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [libs/framework/include/celix/dm/types.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS90eXBlcy5o) | `100.00% <ø> (ø)` | |
| [...ramework/include/celix/dm/ServiceDependency\_Impl.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS9TZXJ2aWNlRGVwZW5kZW5jeV9JbXBsLmg=) | `93.30% <90.09%> (-2.51%)` | :arrow_down: |
| [libs/framework/include/celix/BundleActivator.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9CdW5kbGVBY3RpdmF0b3IuaA==) | `66.66% <100.00%> (-1.08%)` | :arrow_down: |
| [libs/framework/include/celix/TrackerBuilders.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9UcmFja2VyQnVpbGRlcnMuaA==) | `100.00% <100.00%> (ø)` | |
| [...ibs/framework/include/celix/dm/ServiceDependency.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS9TZXJ2aWNlRGVwZW5kZW5jeS5o) | `92.85% <100.00%> (+0.54%)` | :arrow_up: |
| [libs/promises/api/celix/PromiseTimeoutException.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9wcm9taXNlcy9hcGkvY2VsaXgvUHJvbWlzZVRpbWVvdXRFeGNlcHRpb24uaA==) | `0.00% <0.00%> (-33.34%)` | :arrow_down: |
| [libs/framework/include/celix/Exception.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9FeGNlcHRpb24uaA==) | `33.33% <0.00%> (-16.67%)` | :arrow_down: |
| [libs/promises/api/celix/impl/SharedPromiseState.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9wcm9taXNlcy9hcGkvY2VsaXgvaW1wbC9TaGFyZWRQcm9taXNlU3RhdGUuaA==) | `84.04% <0.00%> (-0.70%)` | :arrow_down: |
| [libs/framework/include/celix/Trackers.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9UcmFja2Vycy5o) | `90.68% <0.00%> (-0.25%)` | :arrow_down: |
| ... and [28 more](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [39a3d2e...ee23f57](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] codecov-commenter edited a comment on pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
codecov-commenter edited a comment on pull request #336:
URL: https://github.com/apache/celix/pull/336#issuecomment-833618338
# [Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#336](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ee23f57) into [master](https://codecov.io/gh/apache/celix/commit/39a3d2e794e4583da4c884a4b316688907035a8e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (39a3d2e) will **decrease** coverage by `0.00%`.
> The diff coverage is `91.15%`.
[](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #336 +/- ##
==========================================
- Coverage 70.91% 70.91% -0.01%
==========================================
Files 186 183 -3
Lines 34614 34624 +10
==========================================
+ Hits 24546 24552 +6
- Misses 10068 10072 +4
```
| [Impacted Files](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [libs/framework/include/celix/dm/types.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS90eXBlcy5o) | `100.00% <ø> (ø)` | |
| [...ramework/include/celix/dm/ServiceDependency\_Impl.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS9TZXJ2aWNlRGVwZW5kZW5jeV9JbXBsLmg=) | `93.30% <90.09%> (-2.51%)` | :arrow_down: |
| [libs/framework/include/celix/BundleActivator.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9CdW5kbGVBY3RpdmF0b3IuaA==) | `66.66% <100.00%> (-1.08%)` | :arrow_down: |
| [libs/framework/include/celix/TrackerBuilders.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9UcmFja2VyQnVpbGRlcnMuaA==) | `100.00% <100.00%> (ø)` | |
| [...ibs/framework/include/celix/dm/ServiceDependency.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9kbS9TZXJ2aWNlRGVwZW5kZW5jeS5o) | `92.85% <100.00%> (+0.54%)` | :arrow_up: |
| [libs/promises/api/celix/PromiseTimeoutException.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9wcm9taXNlcy9hcGkvY2VsaXgvUHJvbWlzZVRpbWVvdXRFeGNlcHRpb24uaA==) | `0.00% <0.00%> (-33.34%)` | :arrow_down: |
| [libs/framework/include/celix/Exception.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9FeGNlcHRpb24uaA==) | `33.33% <0.00%> (-16.67%)` | :arrow_down: |
| [libs/promises/api/celix/impl/SharedPromiseState.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9wcm9taXNlcy9hcGkvY2VsaXgvaW1wbC9TaGFyZWRQcm9taXNlU3RhdGUuaA==) | `84.04% <0.00%> (-0.70%)` | :arrow_down: |
| [libs/framework/include/celix/Trackers.h](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bGlicy9mcmFtZXdvcmsvaW5jbHVkZS9jZWxpeC9UcmFja2Vycy5o) | `90.68% <0.00%> (-0.25%)` | :arrow_down: |
| ... and [28 more](https://codecov.io/gh/apache/celix/pull/336/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [39a3d2e...ee23f57](https://codecov.io/gh/apache/celix/pull/336?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627498943
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
Review comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627501889
##########
File path: libs/framework/include/celix/BundleActivator.h
##########
@@ -84,9 +82,9 @@ namespace celix {
celix_status_t destroyActivator(void *userData) {
auto *data = static_cast<BundleActivatorData<I> *>(userData);
std::weak_ptr<celix::BundleContext> ctx = data->ctx;
- std::weak_ptr<celix::dm::DependencyManager> dm = data->dm;
+ std::weak_ptr<celix::dm::DependencyManager> dm = data->ctx->getDependencyManager();
auto bndId = data->bndId;
- data->dm->clear();
+ data->ctx->getDependencyManager()->clear();
Review comment:
note that the clear is done before the `delete data`, so I think a lock() on a weak ptr is not needed and prefer using the data->ctx->getDependencyManager().
I moved the statement a bit higher, hopefully that is more clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r610665854
##########
File path: bundles/async_remote_services/rsa_spi/include/celix/rsa/IExportServiceFactory.h
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+#pragma once
+
+#include <memory>
+
+namespace celix::rsa {
+
+ /**
+ * @brief ExportService class which represent a (opaque) exported service.
+ * The lifetime of this object should be coupled with the lifetime of the exported service.
+ */
+ class ExportedService {
+ public:
+ virtual ~ExportedService() noexcept = default;
+
+ /**
+ * @brief returns the endpoint for this exported service (to be used in discovery).
+ */
+ virtual const celix::rsa::Endpoint& getEndpoint() const = 0;
+ };
+
+ /**
+ * @brief A export service factory for a specific service type.
+ *
+ * The service type which this export service factory targets is provided with
+ * the mandatory celix::rsa::IExportServiceFactory::TARGET_SERVICE_NAME service property.
+ *
+ */
+ class IExportServiceFactory {
Review comment:
- Correct. The factory knows the type and can get the svc id from the service properties, as result it can create a ExportedService which a service dependency to the service.
- If I am correct the current setup is that an exported service is marked using a IExportedService interface. So the ExportServiceFactory creates a ExportService which is a component with a publish, subscriber, dependency on the "service to export" and a IExportedService provided interface.
The IExportedService services can be tracked by the topology manager or discovery manager so that it can be announced in the network. The async admin does not need to know this, it "only" has to keep the std::unique_ptr<ExportService> as long as the "service to export" and the Factory which created the ExportService is tracked/available.
Note that for now the "announce in network" is not relevant; configured discovery only handles imports.
- Yes, exactly. And an IMO additional benefit is that the 'dm' overview will also cleary show which service is exported by a ExportService component.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618349004
##########
File path: libs/framework/include/celix/dm/ServiceDependency_Impl.h
##########
@@ -403,22 +531,64 @@ void ServiceDependency<T,I>::setupCallbacks() {
int(*cadd)(void*, void *, const celix_properties_t*) {nullptr};
int(*crem)(void*, void *, const celix_properties_t*) {nullptr};
- if (setFp != nullptr) {
- cset = [](void* handle, void *service, const celix_properties_t* props) -> int {
+ if (setFp || setFpUsingSharedPtr) {
+ cset = [](void* handle, void* rawSvc, const celix_properties_t* rawProps) -> int {
+ int rc = 0;
auto dep = (ServiceDependency<T,I>*) handle;
- return dep->invokeCallback(dep->setFp, props, service);
+ if (dep->setFp) {
+ rc = dep->invokeCallback(dep->setFp, rawProps, rawSvc);
+ }
+ if (dep->setFpUsingSharedPtr) {
+ auto svcId = dep->setService.second ? dep->setService.second->getAsLong(celix::SERVICE_ID, -1) : -1;
+ std::weak_ptr<I> currentSvc = dep->setService.first;
+ std::weak_ptr<const celix::Properties> currentProps = dep->setService.second;
+ auto svc = std::shared_ptr<I>{static_cast<I*>(rawSvc), [](I*){/*nop*/}};
+ auto props = rawProps ? celix::Properties::wrap(rawProps) : nullptr;
+ dep->setService = std::make_pair(std::move(svc), std::move(props));
+ dep->setFpUsingSharedPtr(dep->setService.first, dep->setService.second);
+ dep->waitForExpired(currentSvc, svcId, "service pointer");
Review comment:
Does the user stuff happen on another thread then?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r610651520
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
}
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
-
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
std::unique_lock l(_m);
- auto instanceIt = _exportedServiceInstances.find(svcId);
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ //remove export interface (if present)
+ auto instanceIt = _exportedServiceInstances.find(svcId);
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
-
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
- }
+}
- auto existingFactory = _importedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::createExportServices() {
+ std::lock_guard<std::mutex> lock{_m};
- if(existingFactory == end(_importedServiceFactories)) {
- L_DEBUG("Adding endpoint but no factory available yet, delaying creation");
- _toBeCreatedImportedEndpoints.emplace_back(celix::rsa::Endpoint{properties});
- return;
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ auto serviceName = it->second->get(celix::SERVICE_NAME, "");
+ auto svcId = it->second->getAsLong(celix::SERVICE_ID, -1);
+ if (serviceName.empty()) {
+ L_WARN("Adding service to be exported but missing objectclass for svc id %li", svcId);
+ } else {
+ auto factory = _exportedServiceFactories.find(serviceName);
+ if (factory != _exportedServiceFactories.end()) {
+ auto endpointId = _ctx->getFramework()->getUUID() + "-" + std::to_string(svcId);
+ auto cmpUUID = factory->second->create(it->first.get(), std::move(endpointId)).getUUID();
+ _exportedServiceInstances.emplace(svcId, std::move(cmpUUID));
+ _toBeExportedServices.erase(it--);
+ } else {
+ L_DEBUG("Adding service to be exported but no factory available yet, delaying creation");
+ }
+ }
}
+}
- L_DEBUG("Adding endpoint, created service");
- _importedServiceInstances.emplace(endpointId, existingFactory->second->create(endpointId));
+void celix::async_rsa::AsyncAdmin::setLogger(const std::shared_ptr<celix_log_service>& logService) {
+ _logService = logService;
}
class AdminActivator {
public:
- explicit AdminActivator(std::shared_ptr<celix::dm::DependencyManager> mng) :
- _cmp(mng->createComponent(std::make_unique<celix::async_rsa::AsyncAdmin>(mng))), _mng(std::move(mng)) {
+ explicit AdminActivator(const std::shared_ptr<celix::BundleContext>& ctx) {
+ auto admin = std::make_shared<celix::async_rsa::AsyncAdmin>(ctx);
- _cmp.createServiceDependency<celix::rsa::Endpoint>()
+ auto& cmp = ctx->getDependencyManager()->createComponent(admin);
+ cmp.createServiceDependency<celix::rsa::Endpoint>()
.setRequired(false)
- .setCallbacks(&celix::async_rsa::AsyncAdmin::addEndpoint, &celix::async_rsa::AsyncAdmin::removeEndpoint)
- .build();
-
- _cmp.createServiceDependency<celix::async_rsa::IImportedServiceFactory>()
+ .setStrategy(celix::dm::DependencyUpdateStrategy::locking)
Review comment:
I see that this is missing some documentation on the enum values.
For service dependency the suspend strategy will stop and start the component for every update (note that this is not really relevant for the admin, because it has no stop/start).
The locking strategy will use the dependency is protected with a lock, but has no locking on it self.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618356212
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
Review comment:
`REMOTE_SERVICE_PROPERTY_NAME` implies a string, if it is a string, `get` is probably a better match than `getAsBool`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618356403
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
Review comment:
`REMOTE_SERVICE_PROPERTY_NAME` implies a string, if it is a string, `get` is probably a better match than `getAsBool`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes merged pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes merged pull request #336:
URL: https://github.com/apache/celix/pull/336
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627498292
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
- return;
- }
+ std::lock_guard l(_m);
- std::unique_lock l(_m);
+ //remove export interface (if present)
auto instanceIt = _exportedServiceInstances.find(svcId);
-
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
Review comment:
nice catch
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r610655552
##########
File path: libs/framework/include/celix/dm/ServiceDependency_Impl.h
##########
@@ -403,22 +531,64 @@ void ServiceDependency<T,I>::setupCallbacks() {
int(*cadd)(void*, void *, const celix_properties_t*) {nullptr};
int(*crem)(void*, void *, const celix_properties_t*) {nullptr};
- if (setFp != nullptr) {
- cset = [](void* handle, void *service, const celix_properties_t* props) -> int {
+ if (setFp || setFpUsingSharedPtr) {
+ cset = [](void* handle, void* rawSvc, const celix_properties_t* rawProps) -> int {
+ int rc = 0;
auto dep = (ServiceDependency<T,I>*) handle;
- return dep->invokeCallback(dep->setFp, props, service);
+ if (dep->setFp) {
+ rc = dep->invokeCallback(dep->setFp, rawProps, rawSvc);
+ }
+ if (dep->setFpUsingSharedPtr) {
+ auto svcId = dep->setService.second ? dep->setService.second->getAsLong(celix::SERVICE_ID, -1) : -1;
+ std::weak_ptr<I> currentSvc = dep->setService.first;
+ std::weak_ptr<const celix::Properties> currentProps = dep->setService.second;
+ auto svc = std::shared_ptr<I>{static_cast<I*>(rawSvc), [](I*){/*nop*/}};
+ auto props = rawProps ? celix::Properties::wrap(rawProps) : nullptr;
+ dep->setService = std::make_pair(std::move(svc), std::move(props));
+ dep->setFpUsingSharedPtr(dep->setService.first, dep->setService.second);
+ dep->waitForExpired(currentSvc, svcId, "service pointer");
Review comment:
The currentSvc is the service "currently" set and which just has been replaced (Maybe rename it so replacedSvc?).
So that waitForExpired will wait until the user has no instance of that service anymore.
With raw pointer this check is not possible, but we assume that the user will not use the old service anymore after a new service is set. With shared_ptr this can be checked.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618355390
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
- return;
- }
+ std::lock_guard l(_m);
- std::unique_lock l(_m);
+ //remove export interface (if present)
auto instanceIt = _exportedServiceInstances.find(svcId);
-
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
Review comment:
`getAsLong` is probably a better match here instead of `getAsBool`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r618362878
##########
File path: libs/framework/include/celix/BundleActivator.h
##########
@@ -84,9 +82,9 @@ namespace celix {
celix_status_t destroyActivator(void *userData) {
auto *data = static_cast<BundleActivatorData<I> *>(userData);
std::weak_ptr<celix::BundleContext> ctx = data->ctx;
- std::weak_ptr<celix::dm::DependencyManager> dm = data->dm;
+ std::weak_ptr<celix::dm::DependencyManager> dm = data->ctx->getDependencyManager();
auto bndId = data->bndId;
- data->dm->clear();
+ data->ctx->getDependencyManager()->clear();
Review comment:
Nitpick: `dm->lock()->clear();` would be clearer here
##########
File path: libs/framework/include/celix/BundleActivator.h
##########
@@ -84,9 +82,9 @@ namespace celix {
celix_status_t destroyActivator(void *userData) {
auto *data = static_cast<BundleActivatorData<I> *>(userData);
std::weak_ptr<celix::BundleContext> ctx = data->ctx;
- std::weak_ptr<celix::dm::DependencyManager> dm = data->dm;
+ std::weak_ptr<celix::dm::DependencyManager> dm = data->ctx->getDependencyManager();
auto bndId = data->bndId;
- data->dm->clear();
+ data->ctx->getDependencyManager()->clear();
Review comment:
Nitpick: `dm.lock()->clear();` would be clearer here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] Oipo commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
Oipo commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r610585464
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,82 +21,80 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
+
}
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
-}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
+ return;
+ }
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
return;
}
+ _toBeImportedServices.emplace_back(endpoint);
- std::unique_lock l(_m);
+ l.unlock();
+ createImportServices();
+}
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
+ std::unique_lock l(_m);
Review comment:
The lock is taken before id is checked, leading to potentially useless locking and unlocking. Please create the lock after the check.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
Review comment:
`std::make_pair` is superfluous here, removing it would lead to more readable code.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
}
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
-
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
std::unique_lock l(_m);
- auto instanceIt = _exportedServiceInstances.find(svcId);
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ //remove export interface (if present)
+ auto instanceIt = _exportedServiceInstances.find(svcId);
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
-
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
- }
+}
- auto existingFactory = _importedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::createExportServices() {
+ std::lock_guard<std::mutex> lock{_m};
- if(existingFactory == end(_importedServiceFactories)) {
- L_DEBUG("Adding endpoint but no factory available yet, delaying creation");
- _toBeCreatedImportedEndpoints.emplace_back(celix::rsa::Endpoint{properties});
- return;
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ auto serviceName = it->second->get(celix::SERVICE_NAME, "");
+ auto svcId = it->second->getAsLong(celix::SERVICE_ID, -1);
+ if (serviceName.empty()) {
Review comment:
Please invert the if and use `continue` and remove the else branch. This way, the amount of indentation is minimized and leads to IMO more readable code.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
}
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
-
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
std::unique_lock l(_m);
- auto instanceIt = _exportedServiceInstances.find(svcId);
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ //remove export interface (if present)
+ auto instanceIt = _exportedServiceInstances.find(svcId);
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
-
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
- }
+}
- auto existingFactory = _importedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::createExportServices() {
+ std::lock_guard<std::mutex> lock{_m};
- if(existingFactory == end(_importedServiceFactories)) {
- L_DEBUG("Adding endpoint but no factory available yet, delaying creation");
- _toBeCreatedImportedEndpoints.emplace_back(celix::rsa::Endpoint{properties});
- return;
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ auto serviceName = it->second->get(celix::SERVICE_NAME, "");
+ auto svcId = it->second->getAsLong(celix::SERVICE_ID, -1);
+ if (serviceName.empty()) {
+ L_WARN("Adding service to be exported but missing objectclass for svc id %li", svcId);
+ } else {
+ auto factory = _exportedServiceFactories.find(serviceName);
+ if (factory != _exportedServiceFactories.end()) {
Review comment:
Please invert the if and use `continue` and remove the else branch. This way, the amount of indentation is minimized and leads to IMO more readable code.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -112,64 +110,59 @@ void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::I
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ l.unlock();
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
std::unique_lock l(_m);
_importedServiceFactories.erase(interface);
-}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
+}
- if(interface.empty()) {
- L_WARN("Adding exported factory but missing exported interfaces");
- return;
+void celix::async_rsa::AsyncAdmin::createImportServices() {
Review comment:
Would you mind moving this function near to `createExportServices`? That way the semantically-close functions are also closer together in the source file.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -112,64 +110,59 @@ void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::I
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ l.unlock();
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
std::unique_lock l(_m);
_importedServiceFactories.erase(interface);
-}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
+}
- if(interface.empty()) {
- L_WARN("Adding exported factory but missing exported interfaces");
- return;
+void celix::async_rsa::AsyncAdmin::createImportServices() {
+ std::lock_guard<std::mutex> lock{_m};
+
+ for (auto it = _toBeImportedServices.begin(); it != _toBeImportedServices.end(); ++it) {
+ auto interface = (*it)->getExportedInterfaces();
+ auto existingFactory = _importedServiceFactories.find(interface);
+ if (existingFactory != end(_importedServiceFactories)) {
+ auto endpointId = (*it)->getEndpointId();
+ L_DEBUG("Adding endpoint, created service");
+ _importedServiceInstances.emplace(endpointId, existingFactory->second->create(endpointId).getUUID());
+ _toBeImportedServices.erase(it--);
+ } else {
+ L_DEBUG("Adding endpoint to be imported but no factory available yet, delaying import");
+ }
}
+}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
+ if (interface.empty()) {
+ L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- _exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
+ {
+ std::lock_guard<std::mutex> lock{_m};
Review comment:
`createExportServices` is only called in situations where the same mutex is locked just before. Please extend the lock lifetime to include `createExportServices` and remove the lock from said function.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,82 +21,80 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
+
}
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
std::unique_lock l(_m);
Review comment:
The lock is now taken before interface and endpoint are checked, leading to potentially useless locking and unlocking. Please create the lock after these checks.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,82 +21,80 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
+
}
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
-}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
+ return;
+ }
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
return;
}
+ _toBeImportedServices.emplace_back(endpoint);
- std::unique_lock l(_m);
+ l.unlock();
Review comment:
Using a manual unlock is error-prone. On top of that, `createImportServices` is only called in situations where the same mutex is locked just before. Please extend the lock lifetime to include `createImportServices` and remove the lock from said function.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -112,64 +110,59 @@ void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::I
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ l.unlock();
Review comment:
Using a manual unlock is error-prone. On top of that, `createImportServices` is only called in situations where the same mutex is locked just before. Please extend the lock lifetime to include `createImportServices` and remove the lock from said function.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
}
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
-
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
std::unique_lock l(_m);
- auto instanceIt = _exportedServiceInstances.find(svcId);
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ //remove export interface (if present)
+ auto instanceIt = _exportedServiceInstances.find(svcId);
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
-
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
- }
+}
- auto existingFactory = _importedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::createExportServices() {
+ std::lock_guard<std::mutex> lock{_m};
- if(existingFactory == end(_importedServiceFactories)) {
- L_DEBUG("Adding endpoint but no factory available yet, delaying creation");
- _toBeCreatedImportedEndpoints.emplace_back(celix::rsa::Endpoint{properties});
- return;
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ auto serviceName = it->second->get(celix::SERVICE_NAME, "");
+ auto svcId = it->second->getAsLong(celix::SERVICE_ID, -1);
+ if (serviceName.empty()) {
+ L_WARN("Adding service to be exported but missing objectclass for svc id %li", svcId);
+ } else {
+ auto factory = _exportedServiceFactories.find(serviceName);
+ if (factory != _exportedServiceFactories.end()) {
+ auto endpointId = _ctx->getFramework()->getUUID() + "-" + std::to_string(svcId);
+ auto cmpUUID = factory->second->create(it->first.get(), std::move(endpointId)).getUUID();
+ _exportedServiceInstances.emplace(svcId, std::move(cmpUUID));
+ _toBeExportedServices.erase(it--);
Review comment:
Please set `it` to the return value of `erase` instead of decrementing it yourself. Iterator stability is not guaranteed for `unordered_map::erase`. Taken from [cppreference](https://en.cppreference.com/w/cpp/container/unordered_map/erase):
`References and iterators to the erased elements are invalidated.`
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
Review comment:
`createExportServices` is only called in situations where the same mutex is locked just before. Please extend the lock lifetime to include `createExportServices` and remove the lock from said function.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -112,64 +110,59 @@ void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::I
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ l.unlock();
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
std::unique_lock l(_m);
_importedServiceFactories.erase(interface);
-}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
+}
- if(interface.empty()) {
- L_WARN("Adding exported factory but missing exported interfaces");
- return;
+void celix::async_rsa::AsyncAdmin::createImportServices() {
+ std::lock_guard<std::mutex> lock{_m};
+
+ for (auto it = _toBeImportedServices.begin(); it != _toBeImportedServices.end(); ++it) {
+ auto interface = (*it)->getExportedInterfaces();
+ auto existingFactory = _importedServiceFactories.find(interface);
+ if (existingFactory != end(_importedServiceFactories)) {
Review comment:
Please invert the if and use `continue` and remove the else branch. This way, the amount of indentation is minimized and leads to IMO more readable code.
##########
File path: libs/framework/include/celix/dm/ServiceDependency_Impl.h
##########
@@ -403,22 +531,64 @@ void ServiceDependency<T,I>::setupCallbacks() {
int(*cadd)(void*, void *, const celix_properties_t*) {nullptr};
int(*crem)(void*, void *, const celix_properties_t*) {nullptr};
- if (setFp != nullptr) {
- cset = [](void* handle, void *service, const celix_properties_t* props) -> int {
+ if (setFp || setFpUsingSharedPtr) {
+ cset = [](void* handle, void* rawSvc, const celix_properties_t* rawProps) -> int {
+ int rc = 0;
auto dep = (ServiceDependency<T,I>*) handle;
- return dep->invokeCallback(dep->setFp, props, service);
+ if (dep->setFp) {
+ rc = dep->invokeCallback(dep->setFp, rawProps, rawSvc);
+ }
+ if (dep->setFpUsingSharedPtr) {
+ auto svcId = dep->setService.second ? dep->setService.second->getAsLong(celix::SERVICE_ID, -1) : -1;
+ std::weak_ptr<I> currentSvc = dep->setService.first;
+ std::weak_ptr<const celix::Properties> currentProps = dep->setService.second;
+ auto svc = std::shared_ptr<I>{static_cast<I*>(rawSvc), [](I*){/*nop*/}};
+ auto props = rawProps ? celix::Properties::wrap(rawProps) : nullptr;
+ dep->setService = std::make_pair(std::move(svc), std::move(props));
+ dep->setFpUsingSharedPtr(dep->setService.first, dep->setService.second);
+ dep->waitForExpired(currentSvc, svcId, "service pointer");
Review comment:
Am I correct in understanding that this call will hang until the service to which it is being added is completely stopped? I don't see another time when the weak_ptrs expire.
##########
File path: bundles/async_remote_services/rsa_spi/include/celix/rsa/IExportServiceFactory.h
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+#pragma once
+
+#include <memory>
+
+namespace celix::rsa {
+
+ /**
+ * @brief ExportService class which represent a (opaque) exported service.
+ * The lifetime of this object should be coupled with the lifetime of the exported service.
+ */
+ class ExportedService {
+ public:
+ virtual ~ExportedService() noexcept = default;
+
+ /**
+ * @brief returns the endpoint for this exported service (to be used in discovery).
+ */
+ virtual const celix::rsa::Endpoint& getEndpoint() const = 0;
+ };
+
+ /**
+ * @brief A export service factory for a specific service type.
+ *
+ * The service type which this export service factory targets is provided with
+ * the mandatory celix::rsa::IExportServiceFactory::TARGET_SERVICE_NAME service property.
+ *
+ */
+ class IExportServiceFactory {
Review comment:
The impact of switching over to an opaque factory (either imported or exported) I think requires some more thought. Some of the impact that I can foresee right now is:
* Factory implementations would add service dependencies instead of passing the `void*` directly, for each service created.
* This would mean that the `exportService` function below should not return the exportedService. The admin should add a tracker for that as well.
* Factories should probably also have the new services have a dependency on the factory itself. This and the previous point would solve the problem with creating/destroying bundles IMO
##########
File path: bundles/async_remote_services/rsa_spi/include/celix/rsa/Exception.h
##########
@@ -0,0 +1,21 @@
+#pragma once
+
+#include <exception>
+
+namespace celix::rsa {
+
+ /**
+ * @brief Celix Remote Service Admin Exception
+ */
+ class Exception : public std::exception {
Review comment:
We should consider giving this exception a better name. It currently is not thrown anywhere, but once there is a place where it is thrown, the use case should be grokkable from the name.
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -178,143 +171,121 @@ void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa
std::unique_lock l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
+ {
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(std::make_pair(std::move(svc), std::move(props)));
}
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
-
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
std::unique_lock l(_m);
- auto instanceIt = _exportedServiceInstances.find(svcId);
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ //remove export interface (if present)
+ auto instanceIt = _exportedServiceInstances.find(svcId);
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
-
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
- }
+}
- auto existingFactory = _importedServiceFactories.find(interface);
+void celix::async_rsa::AsyncAdmin::createExportServices() {
+ std::lock_guard<std::mutex> lock{_m};
- if(existingFactory == end(_importedServiceFactories)) {
- L_DEBUG("Adding endpoint but no factory available yet, delaying creation");
- _toBeCreatedImportedEndpoints.emplace_back(celix::rsa::Endpoint{properties});
- return;
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ auto serviceName = it->second->get(celix::SERVICE_NAME, "");
+ auto svcId = it->second->getAsLong(celix::SERVICE_ID, -1);
+ if (serviceName.empty()) {
+ L_WARN("Adding service to be exported but missing objectclass for svc id %li", svcId);
+ } else {
+ auto factory = _exportedServiceFactories.find(serviceName);
+ if (factory != _exportedServiceFactories.end()) {
+ auto endpointId = _ctx->getFramework()->getUUID() + "-" + std::to_string(svcId);
+ auto cmpUUID = factory->second->create(it->first.get(), std::move(endpointId)).getUUID();
+ _exportedServiceInstances.emplace(svcId, std::move(cmpUUID));
+ _toBeExportedServices.erase(it--);
+ } else {
+ L_DEBUG("Adding service to be exported but no factory available yet, delaying creation");
+ }
+ }
}
+}
- L_DEBUG("Adding endpoint, created service");
- _importedServiceInstances.emplace(endpointId, existingFactory->second->create(endpointId));
+void celix::async_rsa::AsyncAdmin::setLogger(const std::shared_ptr<celix_log_service>& logService) {
+ _logService = logService;
}
class AdminActivator {
public:
- explicit AdminActivator(std::shared_ptr<celix::dm::DependencyManager> mng) :
- _cmp(mng->createComponent(std::make_unique<celix::async_rsa::AsyncAdmin>(mng))), _mng(std::move(mng)) {
+ explicit AdminActivator(const std::shared_ptr<celix::BundleContext>& ctx) {
+ auto admin = std::make_shared<celix::async_rsa::AsyncAdmin>(ctx);
- _cmp.createServiceDependency<celix::rsa::Endpoint>()
+ auto& cmp = ctx->getDependencyManager()->createComponent(admin);
+ cmp.createServiceDependency<celix::rsa::Endpoint>()
.setRequired(false)
- .setCallbacks(&celix::async_rsa::AsyncAdmin::addEndpoint, &celix::async_rsa::AsyncAdmin::removeEndpoint)
- .build();
-
- _cmp.createServiceDependency<celix::async_rsa::IImportedServiceFactory>()
+ .setStrategy(celix::dm::DependencyUpdateStrategy::locking)
Review comment:
For my information: given that all these functions use a mutex themselves, isn't using a suspend strategy possible here and reduces some overhead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [celix] pnoltes commented on a change in pull request #336: Feature/async update
Posted by GitBox <gi...@apache.org>.
pnoltes commented on a change in pull request #336:
URL: https://github.com/apache/celix/pull/336#discussion_r627497949
##########
File path: bundles/async_remote_services/admin/src/admin.cc
##########
@@ -21,300 +21,256 @@
#include <celix_api.h>
#include <pubsub_message_serialization_service.h>
#include <ConfiguredEndpoint.h>
+#include "celix/rsa/Constants.h"
#define L_DEBUG(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_DEBUG, __VA_ARGS__)
+ if (_logService) { \
+ _logService->debug(_logService->handle, __VA_ARGS__); \
+ }
#define L_INFO(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_INFO, __VA_ARGS__)
+ if (_logService) { \
+ _logService->info(_logService->handle, __VA_ARGS__); \
+ }
#define L_WARN(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_WARNING, __VA_ARGS__)
+ if (_logService) { \
+ _logService->warning(_logService->handle, __VA_ARGS__); \
+ }
#define L_ERROR(...) \
- celix_logHelper_log(_logger, CELIX_LOG_LEVEL_ERROR, __VA_ARGS__)
-
-celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::dm::DependencyManager> &mng) noexcept : _mng(mng), _logger(celix_logHelper_create(mng->bundleContext(), "celix_async_rsa_admin")) {
-
- // C++ version of tracking services without type not possible (yet?)
- celix_service_tracking_options_t opts{};
- opts.callbackHandle = this;
- opts.addWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->addService(svc, props);
- };
- opts.removeWithProperties = [](void *handle, void *svc, const celix_properties_t *props){
- auto *admin = static_cast<AsyncAdmin*>(handle);
- admin->removeService(svc, props);
- };
- _serviceTrackerId = celix_bundleContext_trackServicesWithOptions(_mng->bundleContext(), &opts);
-
- if(_serviceTrackerId < 0) {
- L_ERROR("couldn't register tracker");
+ if (_logService) { \
+ _logService->error(_logService->handle, __VA_ARGS__); \
}
-}
-celix::async_rsa::AsyncAdmin::~AsyncAdmin() {
- celix_logHelper_destroy(_logger);
- celix_bundleContext_stopTracker(_mng->bundleContext(), _serviceTrackerId);
-}
+celix::async_rsa::AsyncAdmin::AsyncAdmin(std::shared_ptr<celix::BundleContext> ctx) noexcept : _ctx{std::move(ctx)} {
-void celix::async_rsa::AsyncAdmin::addEndpoint(celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- std::unique_lock l(_m);
- addEndpointInternal(*endpoint);
}
-void celix::async_rsa::AsyncAdmin::removeEndpoint([[maybe_unused]] celix::rsa::Endpoint* endpoint, [[maybe_unused]] Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(interface.empty()) {
- L_DEBUG("Removing endpoint but no exported interfaces");
+ auto interface = endpoint->getExportedInterfaces();
+ if (interface.empty()) {
+ L_WARN("Adding endpoint but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ auto endpointId = endpoint->getEndpointId();
+ if (endpointId.empty()) {
+ L_WARN("Adding endpoint but missing service id");
+ return;
+ }
- _toBeCreatedImportedEndpoints.erase(std::remove_if(_toBeCreatedImportedEndpoints.begin(), _toBeCreatedImportedEndpoints.end(), [&interface](auto const &endpoint){
- auto endpointInterface = endpoint.getProperties().get(ENDPOINT_EXPORTS);
- return !endpointInterface.empty() && endpointInterface == interface;
- }), _toBeCreatedImportedEndpoints.end());
+ std::lock_guard l(_m);
+ _toBeImportedServices.emplace_back(endpoint);
+ createImportServices();
+}
- auto svcId = properties.get(ENDPOINT_IDENTIFIER);
+void celix::async_rsa::AsyncAdmin::removeEndpoint(const std::shared_ptr<celix::rsa::Endpoint>& endpoint) {
+ assert(endpoint);
- if(svcId.empty()) {
- L_DEBUG("Removing endpoint but no service instance");
+ auto id = endpoint->getEndpointId();
+ if (id.empty()) {
+ L_WARN("Cannot remove endpoint without a valid id");
return;
}
- auto instanceIt = _importedServiceInstances.find(svcId);
+ std::lock_guard l(_m);
+
+ _toBeImportedServices.erase(std::remove_if(_toBeImportedServices.begin(), _toBeImportedServices.end(), [&id](auto const &endpoint){
+ return id == endpoint->getEndpointId();
+ }), _toBeImportedServices.end());
+
- if(instanceIt == end(_importedServiceInstances)) {
+ auto instanceIt = _importedServiceInstances.find(id);
+ if (instanceIt == end(_importedServiceInstances)) {
return;
}
- _mng->destroyComponent(instanceIt->second);
+ _ctx->getDependencyManager()->removeComponentAsync(instanceIt->second);
_importedServiceInstances.erase(instanceIt);
}
-void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_DEBUG("Adding service factory but no exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
auto existingFactory = _importedServiceFactories.find(interface);
-
if(existingFactory != end(_importedServiceFactories)) {
L_WARN("Adding imported factory but factory already exists");
return;
}
_importedServiceFactories.emplace(interface, factory);
- for(auto it = _toBeCreatedImportedEndpoints.begin(); it != _toBeCreatedImportedEndpoints.end();) {
- auto tbceInterface = it->getProperties().get(ENDPOINT_EXPORTS);
- if(tbceInterface.empty() || tbceInterface != interface) {
- it++;
- } else {
- addEndpointInternal(*it);
- _toBeCreatedImportedEndpoints.erase(it);
- }
- }
+ createImportServices();
}
-void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory([[maybe_unused]] celix::async_rsa::IImportedServiceFactory *factory, Properties &&properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeImportedServiceFactory(const std::shared_ptr<celix::async_rsa::IImportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Removing service factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_importedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the imported services also needed when a ImportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *factory, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::addExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& factory, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
- if(interface.empty()) {
+ if (interface.empty()) {
L_WARN("Adding exported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
- auto factoryIt = _exportedServiceFactories.find(interface);
-
- if(factoryIt != end(_exportedServiceFactories)) {
- L_WARN("Adding exported factory but factory already exists");
- return;
- }
-
+ std::lock_guard<std::mutex> lock{_m};
_exportedServiceFactories.emplace(interface, factory);
- L_WARN("Looping over %i tbce for interface %s", _toBeCreatedExportedEndpoints.size(), interface.c_str());
-
- for(auto it = _toBeCreatedExportedEndpoints.begin(); it != _toBeCreatedExportedEndpoints.end(); ) {
- auto interfaceToBeCreated = it->second.get(ENDPOINT_EXPORTS);
- L_WARN("Checking tbce interface %s", interfaceToBeCreated.c_str());
-
- if(interfaceToBeCreated.empty() || interfaceToBeCreated != interface) {
- it++;
- } else {
- auto endpointId = it->second.get(ENDPOINT_IDENTIFIER);
- _exportedServiceInstances.emplace(endpointId, factory->create(it->first, endpointId));
- it = _toBeCreatedExportedEndpoints.erase(it);
- }
- }
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(celix::async_rsa::IExportedServiceFactory *, Properties&& properties) {
- auto interface = properties.get(ENDPOINT_EXPORTS);
+void celix::async_rsa::AsyncAdmin::removeExportedServiceFactory(const std::shared_ptr<celix::async_rsa::IExportedServiceFactory>& /*factory*/, const std::shared_ptr<const celix::Properties>& properties) {
+ auto interface = properties->get(celix::rsa::Endpoint::EXPORTS);
if(interface.empty()) {
L_WARN("Removing imported factory but missing exported interfaces");
return;
}
- std::unique_lock l(_m);
+ std::lock_guard l(_m);
_exportedServiceFactories.erase(interface);
+
+ //TODO TBD are the removal of the exported services also needed when a ExportedServiceFactory is removed (and maybe the bundle is uninstalled?)
}
-void celix::async_rsa::AsyncAdmin::addService(void *svc, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto endpointId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
+void celix::async_rsa::AsyncAdmin::addService(const std::shared_ptr<void>& svc, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(objectClass == nullptr) {
+ auto serviceName = props->get(celix::SERVICE_NAME, "");
+ if (serviceName.empty()) {
L_WARN("Adding service to be exported but missing objectclass");
return;
}
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
- return;
- } else {
- L_WARN("found remote service %s", objectClass);
- }
-
- if(endpointId == nullptr) {
- L_WARN("Adding service to be exported but missing endpoint.id");
- return;
- }
-
- std::unique_lock l(_m);
- auto factory = _exportedServiceFactories.find(objectClass);
-
- if(factory == end(_exportedServiceFactories)) {
- L_WARN("Adding service to be exported but no factory available yet, delaying creation");
-
- Properties cppProps;
-
- auto it = celix_propertiesIterator_construct(props);
- const char* key;
- while(key = celix_propertiesIterator_nextKey(&it), key != nullptr) {
- cppProps.set(key, celix_properties_get(props, key, nullptr));
- }
-
- _toBeCreatedExportedEndpoints.emplace_back(std::make_pair(svc, std::move(cppProps)));
- return;
- }
-
- _exportedServiceInstances.emplace(endpointId, factory->second->create(svc, endpointId));
+ std::lock_guard<std::mutex> lock{_m};
+ _toBeExportedServices.emplace_back(svc, props);
+ createExportServices();
}
-void celix::async_rsa::AsyncAdmin::removeService(void *, const celix_properties_t *props) {
- auto *objectClass = celix_properties_get(props, OSGI_FRAMEWORK_OBJECTCLASS, nullptr);
- auto *remote = celix_properties_get(props, "remote", nullptr);
- auto svcId = celix_properties_get(props, ENDPOINT_IDENTIFIER, nullptr);
-
- if(objectClass == nullptr) {
- L_WARN("Removing service to be exported but missing objectclass");
- return;
- }
+void celix::async_rsa::AsyncAdmin::removeService(const std::shared_ptr<void>& /*svc*/, const std::shared_ptr<const celix::Properties>& props) {
+ assert(props->getAsBool(celix::rsa::REMOTE_SERVICE_PROPERTY_NAME, false)); //"only remote services should be tracker by the service tracker");
- if(remote == nullptr) {
- // not every service is remote, this is fine but not something the RSA admin is interested in.
+ long svcId = props->getAsLong(celix::SERVICE_ID, -1);
+ if (svcId < 0) {
+ L_WARN("Removing service to be exported but missing service.id");
return;
}
- if(svcId == nullptr) {
- L_WARN("Removing service to be exported but missing endpoint.id");
- return;
- }
+ std::lock_guard l(_m);
- std::unique_lock l(_m);
+ //remove export interface (if present)
auto instanceIt = _exportedServiceInstances.find(svcId);
-
- if(instanceIt == end(_exportedServiceInstances)) {
- return;
+ if (instanceIt != end(_exportedServiceInstances)) {
+ //note cannot remove async, because the svc lifetime needs to extend the lifetime of the component.
+ //TODO improve by creating export service based on svc id instead of svc pointer, so that export service can track its own dependencies.
+ _ctx->getDependencyManager()->removeComponent(instanceIt->second);
+ _exportedServiceInstances.erase(instanceIt);
}
- _mng->destroyComponent(instanceIt->second);
- _exportedServiceInstances.erase(instanceIt);
-}
-
-void celix::async_rsa::AsyncAdmin::addEndpointInternal(celix::rsa::Endpoint& endpoint) {
-
- const auto& properties = endpoint.getProperties();
- auto interface = properties.get(ENDPOINT_EXPORTS);
-
- if(interface.empty()) {
- L_WARN("Adding endpoint but missing exported interfaces");
- return;
+ //remove to be exported endpoint (if present)
+ for (auto it = _toBeExportedServices.begin(); it != _toBeExportedServices.end(); ++it) {
+ if (it->second->getAsBool(celix::SERVICE_ID, -1) == svcId) {
+ _toBeExportedServices.erase(it);
+ break;
+ }
}
- auto endpointId = properties.get(ENDPOINT_IDENTIFIER);
+}
- if(endpointId.empty()) {
- L_WARN("Adding endpoint but missing service id");
- return;
+void celix::async_rsa::AsyncAdmin::createImportServices() {
+ for (auto it = _toBeImportedServices.begin(); it != _toBeImportedServices.end(); ++it) {
+ auto interface = (*it)->getExportedInterfaces();
+ auto existingFactory = _importedServiceFactories.find(interface);
+ if (existingFactory == end(_importedServiceFactories)) {
+ L_DEBUG("Adding endpoint to be imported but no factory available yet, delaying import");
Review comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org