You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aries.apache.org by am...@apache.org on 2026/03/17 17:41:48 UTC
(aries-rsa) branch master updated (db27a1e2 -> 2261a1dc)
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
from db27a1e2 Standardize logger fields
new 2460eba3 Refactor and clarify TopologyManagerImport updates calculation
new 25cbbded Fix potential concurrency-related NPE when updating exported service properties
new 2a8d1c40 Refactor ExportRegistrationImpl.update()
new 424ab659 Fix ServiceExportsRepository retaining services with no exports
new b6eeaf55 Refactor ServiceExportsRepository
new 60bc1eab Fix ImportRegistrationImpl.update not returning updated endpoint
new 0ccbde72 Add TopologyManagerImport tests
new 0003efa3 Cleanup RemoteServiceAdminCore
new 14b6de83 Cleanup TopologyManagerExport
new 32ac095e Move and cleanup EndpointEventListenerTracker
new 140bf2de Fix EndpointEventListenerTracker not being closed
new 6a2f3f65 Cleanup EndpointListenerNotifier
new 9f049b5a Remove unnecessary reverse map in ConfigDiscovery
new 6d59fdf7 Simplify ConfigDiscovery by reversing endpoints map
new f1dbc52d Optimize ConfigDiscovery by creating Filters only once
new e44e28c0 Fix ConfigDiscovery sending ADDED event instead of MODIFIED
new 302c58a6 Fix race condition in ConfigDiscovery that can trigger duplicate events
new 6b90f390 Fix ConfigDiscovery not sending MODIFIED_ENDMATCH events
new 69ab66ef ConfigDiscovery cleanup
new 291b2c04 Add MultiMap.replace method
new 5155c7a0 Fix TopologyManagerImport not closing registration when import fails
new e5cf1237 Fix TopologyManagerImport not unregistering listener when stopped
new 2261a1dc Fix thread pools effectively having fixed size (because queue is unbounded)
The 23 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rsa/discovery/config/ConfigDiscovery.java | 180 ++++++++++-----------
.../aries/rsa/core/ExportRegistrationImpl.java | 23 ++-
.../aries/rsa/core/ImportRegistrationImpl.java | 3 +-
.../aries/rsa/core/RemoteServiceAdminCore.java | 95 +++++------
.../aries/rsa/topologymanager/Activator.java | 17 +-
.../EndpointEventListenerTracker.java | 35 ++--
.../exporter/EndpointListenerNotifier.java | 44 +++--
.../exporter/ServiceExportsRepository.java | 113 +++++--------
.../exporter/TopologyManagerExport.java | 54 +++----
.../rsa/topologymanager/importer/ImportDiff.java | 72 ---------
.../rsa/topologymanager/importer/MultiMap.java | 11 +-
.../importer/TopologyManagerImport.java | 83 ++++++----
.../exporter/EndpointListenerNotifierTest.java | 6 +-
.../exporter/TopologyManagerExportTest.java | 28 +++-
.../importer/TopologyManagerImportTest.java | 158 +++++++++++++-----
15 files changed, 455 insertions(+), 467 deletions(-)
rename topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/{ => exporter}/EndpointEventListenerTracker.java (53%)
delete mode 100644 topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/ImportDiff.java
(aries-rsa) 07/23: Add TopologyManagerImport tests
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 0ccbde72bb1f21c4e2757d1c66913c8665e45fc6
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 16:34:26 2026 +0200
Add TopologyManagerImport tests
---
.../importer/TopologyManagerImportTest.java | 157 +++++++++++++++------
1 file changed, 113 insertions(+), 44 deletions(-)
diff --git a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
index ea600e62..621c2b63 100644
--- a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
+++ b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
@@ -18,14 +18,17 @@
*/
package org.apache.aries.rsa.topologymanager.importer;
-import static org.junit.Assert.assertTrue;
+import static org.easymock.EasyMock.*;
+import static org.easymock.EasyMock.expect;
+import static org.junit.Assert.assertEquals;
-import java.util.concurrent.Semaphore;
-import java.util.concurrent.TimeUnit;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
-import org.easymock.EasyMock;
-import org.easymock.IAnswer;
import org.easymock.IMocksControl;
+import org.junit.After;
+import org.junit.Before;
import org.junit.Test;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceRegistration;
@@ -38,54 +41,120 @@ import org.osgi.service.remoteserviceadmin.RemoteServiceAdminListener;
public class TopologyManagerImportTest {
- @Test
- public void testImportForNewlyAddedRSA() throws InterruptedException {
- IMocksControl c = EasyMock.createControl();
- c.makeThreadSafe(true);
- final Semaphore sema = new Semaphore(0);
- BundleContext bc = getBundleContext(c);
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private BundleContext mockBundleContext(IMocksControl c) {
+ ServiceRegistration sreg = c.createMock(ServiceRegistration.class);
+ BundleContext bc = c.createMock(BundleContext.class);
+ expect(bc.registerService(eq(RemoteServiceAdminListener.class),
+ anyObject(RemoteServiceAdminListener.class),
+ anyObject())).andReturn(sreg).anyTimes();
+ return bc;
+ }
+ private ImportRegistration mockImportRegistration(IMocksControl c, EndpointDescription endpoint) {
+ final ImportRegistration ireg = c.createMock(ImportRegistration.class);
+ expect(ireg.getException()).andReturn(null).anyTimes();
+ expect(ireg.update(anyObject())).andReturn(true).anyTimes();
+ ImportReference iref = c.createMock(ImportReference.class);
+ expect(ireg.getImportReference()).andReturn(iref).anyTimes();
+ expect(iref.getImportedEndpoint()).andReturn(endpoint).anyTimes();
+ return ireg;
+ }
+
+ private EndpointDescription createEndpoint() {
EndpointDescription endpoint = c.createMock(EndpointDescription.class);
- RemoteServiceAdmin rsa = c.createMock(RemoteServiceAdmin.class);
- final ImportRegistration ir = getRegistration(c, endpoint);
- EasyMock.expect(rsa.importService(EasyMock.eq(endpoint))).andAnswer(new IAnswer<ImportRegistration>() {
- public ImportRegistration answer() throws Throwable {
- sema.release();
- return ir;
- }
- }).once();
-
- ir.close();
- EasyMock.expectLastCall();
- c.replay();
+ final ImportRegistration ir = mockImportRegistration(c, endpoint);
+ ir.close(); // must be closed
+ expectLastCall().andAnswer(() -> {
+ endpoints.get(endpoint).decrementAndGet();
+ return null;
+ });
+ expect(rsa.importService(eq(endpoint))).andAnswer(() -> {
+ endpoints.get(endpoint).incrementAndGet();
+ return ir;
+ });
+ endpoints.put(endpoint, new AtomicInteger());
+ return endpoint;
+ }
+
+ IMocksControl c;
+ BundleContext bc;
+ RemoteServiceAdmin rsa;
+ TopologyManagerImport tm;
+ Map<EndpointDescription, AtomicInteger> endpoints = new ConcurrentHashMap<>();
- TopologyManagerImport tm = new TopologyManagerImport(bc);
+ @Before
+ public void setUp() {
+ c = createControl();
+ c.makeThreadSafe(true);
+ bc = mockBundleContext(c);
+ rsa = c.createMock(RemoteServiceAdmin.class);
+ tm = new TopologyManagerImport(bc);
+ }
+
+ public void start() {
+ c.replay();
tm.start();
- EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
- tm.endpointChanged(event, "myFilter");
- tm.add(rsa);
- assertTrue("rsa.ImportService should have been called",
- sema.tryAcquire(100, TimeUnit.SECONDS));
+ }
+
+ @After
+ public void tearDown() {
tm.stop();
c.verify();
}
- private ImportRegistration getRegistration(IMocksControl c, EndpointDescription endpoint) {
- final ImportRegistration ireg = c.createMock(ImportRegistration.class);
- EasyMock.expect(ireg.getException()).andReturn(null).anyTimes();
- ImportReference iref = c.createMock(ImportReference.class);
- EasyMock.expect(ireg.getImportReference()).andReturn(iref).anyTimes();
- EasyMock.expect(iref.getImportedEndpoint()).andReturn(endpoint).anyTimes();
- return ireg;
+ private void assertImports(EndpointDescription endpoint, int registrations) throws InterruptedException {
+ long end = System.currentTimeMillis() + 1000;
+ while (System.currentTimeMillis() < end) {
+ if (endpoints.get(endpoint).get() == registrations)
+ return;
+ Thread.sleep(10);
+ }
+ assertEquals("wrong number of open import registrations", registrations, endpoints.get(endpoint).get());
}
- @SuppressWarnings({ "rawtypes", "unchecked" })
- private BundleContext getBundleContext(IMocksControl c) {
- ServiceRegistration sreg = c.createMock(ServiceRegistration.class);
- BundleContext bc = c.createMock(BundleContext.class);
- EasyMock.expect(bc.registerService(EasyMock.eq(RemoteServiceAdminListener.class),
- EasyMock.anyObject(RemoteServiceAdminListener.class),
- EasyMock.anyObject())).andReturn(sreg).anyTimes();
- return bc;
+ @Test
+ public void testAddEndpointBeforeRsa() throws InterruptedException {
+ EndpointDescription endpoint = createEndpoint();
+ start();
+
+ tm.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), "myFilter");
+ tm.add(rsa);
+ assertImports(endpoint, 1);
+ }
+
+ @Test
+ public void testAddEndpointAfterRsa() throws InterruptedException {
+ EndpointDescription endpoint = createEndpoint();
+ start();
+
+ tm.add(rsa);
+ tm.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), "myFilter");
+ assertImports(endpoint, 1);
+ }
+
+ @Test
+ public void testAddEndpointTwice() throws InterruptedException {
+ EndpointDescription endpoint = createEndpoint();
+ start();
+
+ tm.add(rsa);
+ EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
+ tm.endpointChanged(event, "myFilter");
+ assertImports(endpoint, 1);
+ tm.endpointChanged(event, "myFilter");
+ assertImports(endpoint, 1); // still one one import
+ }
+
+ @Test
+ public void testRemoveEndpoint() throws InterruptedException {
+ EndpointDescription endpoint = createEndpoint();
+ start();
+
+ tm.add(rsa);
+ tm.endpointChanged(new EndpointEvent(EndpointEvent.ADDED, endpoint), "myFilter");
+ assertImports(endpoint, 1);
+ tm.endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, endpoint), "myFilter");
+ assertImports(endpoint, 0);
}
}
(aries-rsa) 23/23: Fix thread pools effectively having fixed size (because queue is unbounded)
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 2261a1dc759da4854c7a020398788d7c49ec0510
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 17:33:28 2026 +0200
Fix thread pools effectively having fixed size (because queue is unbounded)
---
.../java/org/apache/aries/rsa/topologymanager/Activator.java | 9 ++++-----
.../rsa/topologymanager/importer/TopologyManagerImport.java | 5 ++---
2 files changed, 6 insertions(+), 8 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
index daef5a33..080b1696 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
@@ -20,9 +20,8 @@ package org.apache.aries.rsa.topologymanager;
import java.util.Dictionary;
import java.util.Hashtable;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ExecutorService;
import org.apache.aries.rsa.spi.ExportPolicy;
import org.apache.aries.rsa.topologymanager.exporter.DefaultExportPolicy;
@@ -76,7 +75,7 @@ public class Activator implements BundleActivator {
private TopologyManagerImport importManager;
EndpointListenerNotifier notifier;
private ServiceTracker<RemoteServiceAdmin, RemoteServiceAdmin> rsaTracker;
- private ThreadPoolExecutor exportExecutor;
+ private ExecutorService exportExecutor;
private ServiceTracker<ExportPolicy, ExportPolicy> policyTracker;
private EndpointListenerManager endpointListenerManager;
@@ -121,7 +120,7 @@ public class Activator implements BundleActivator {
public void doStart(final BundleContext bc, ExportPolicy policy) {
LOG.debug("TopologyManager: start()");
notifier = new EndpointListenerNotifier();
- exportExecutor = new ThreadPoolExecutor(5, 10, 50, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(TopologyManagerExport.class));
+ exportExecutor = Executors.newCachedThreadPool(new NamedThreadFactory(TopologyManagerExport.class));
exportManager = new TopologyManagerExport(notifier, exportExecutor, policy);
endpointEventListenerTracker = new EndpointEventListenerTracker(bc, exportManager);
importManager = new TopologyManagerImport(bc);
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
index e48ef198..b57d494b 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
@@ -22,9 +22,8 @@ import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.osgi.framework.BundleContext;
@@ -68,7 +67,7 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
public TopologyManagerImport(BundleContext bc) {
rsaSet = new CopyOnWriteArraySet<>();
bctx = bc;
- execService = new ThreadPoolExecutor(5, 10, 50, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getClass()));
+ execService = Executors.newCachedThreadPool(new NamedThreadFactory(getClass()));
}
public void start() {
(aries-rsa) 21/23: Fix TopologyManagerImport not closing registration when import fails
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 5155c7a02b3c71eebf143cc919fd82487e30d771
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 16:44:46 2026 +0200
Fix TopologyManagerImport not closing registration when import fails
---
.../aries/rsa/topologymanager/importer/TopologyManagerImport.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
index 3f5de4a4..d962a816 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
@@ -173,7 +173,8 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
importedServices.put(filter, reg);
return;
} else {
- LOG.info("Error importing service {}", endpoint, reg.getException());
+ LOG.warn("Error importing service {}", endpoint, reg.getException());
+ reg.close();
}
}
}
(aries-rsa) 15/23: Optimize ConfigDiscovery by creating Filters only once
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit f1dbc52d4b03b70d16f7cb40d503136f4fe1c43c
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 22:50:56 2026 +0200
Optimize ConfigDiscovery by creating Filters only once
---
.../rsa/discovery/config/ConfigDiscovery.java | 73 +++++++++++-----------
1 file changed, 35 insertions(+), 38 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index 0223a45b..00532048 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -21,6 +21,7 @@ package org.apache.aries.rsa.discovery.config;
import org.apache.aries.rsa.util.StringPlus;
import org.osgi.framework.Filter;
import org.osgi.framework.FrameworkUtil;
+import org.osgi.framework.InvalidSyntaxException;
import org.osgi.framework.ServiceReference;
import org.osgi.service.cm.ConfigurationException;
import org.osgi.service.cm.ManagedServiceFactory;
@@ -33,7 +34,7 @@ import java.util.concurrent.ConcurrentHashMap;
class ConfigDiscovery implements ManagedServiceFactory {
private final Map<String, EndpointDescription> endpoints = new ConcurrentHashMap<>();
- private final Map<EndpointEventListener, Collection<String>> listenerToFilters = new HashMap<>();
+ private final Map<EndpointEventListener, Collection<Filter>> listenerToFilters = new HashMap<>();
@Override
public String getName() {
@@ -50,22 +51,35 @@ class ConfigDiscovery implements ManagedServiceFactory {
removeServiceDeclaredInConfig(pid);
}
- void addListener(ServiceReference<EndpointEventListener> endpointListenerRef, EndpointEventListener endpointListener) {
- List<String> filters = StringPlus.normalize(endpointListenerRef.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE));
- if (filters.isEmpty()) {
- return;
+ private static List<Filter> createFilters(ServiceReference<EndpointEventListener> ref) {
+ List<String> values = StringPlus.normalize(ref.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE));
+ List<Filter> filters = new ArrayList<>(values.size());
+ for (String value : values) {
+ try {
+ filters.add(FrameworkUtil.createFilter(value));
+ } catch (InvalidSyntaxException ignore) { // bad filter never matches
+ }
}
+ return filters;
+ }
- synchronized (listenerToFilters) {
- listenerToFilters.put(endpointListener, filters);
- }
+ private static boolean matches(Filter filter, EndpointDescription endpoint) {
+ return filter.match(new Hashtable<>(endpoint.getProperties())); // don't use matches() which is case-sensitive
+ }
- triggerCallbacks(filters, endpointListener);
+ void addListener(ServiceReference<EndpointEventListener> ref, EndpointEventListener listener) {
+ List<Filter> filters = createFilters(ref);
+ if (!filters.isEmpty()) {
+ synchronized (listenerToFilters) {
+ listenerToFilters.put(listener, filters);
+ }
+ triggerCallbacks(filters, listener);
+ }
}
- void removeListener(EndpointEventListener endpointListener) {
+ void removeListener(EndpointEventListener listener) {
synchronized (listenerToFilters) {
- listenerToFilters.remove(endpointListener);
+ listenerToFilters.remove(listener);
}
}
@@ -86,52 +100,35 @@ class ConfigDiscovery implements ManagedServiceFactory {
private void triggerCallbacks(EndpointEvent event) {
EndpointDescription endpoint = event.getEndpoint();
// make a copy of matched filters/listeners so that caller doesn't need to hold locks while triggering events
- List<Map.Entry<EndpointEventListener, String>> matched = new ArrayList<>();
+ List<Map.Entry<EndpointEventListener, Filter>> matched = new ArrayList<>();
synchronized (listenerToFilters) {
- for (Map.Entry<EndpointEventListener, Collection<String>> entry : listenerToFilters.entrySet()) {
+ for (Map.Entry<EndpointEventListener, Collection<Filter>> entry : listenerToFilters.entrySet()) {
EndpointEventListener listener = entry.getKey();
- for (String filter : entry.getValue()) {
- if (matchFilter(filter, endpoint)) {
+ for (Filter filter : entry.getValue()) {
+ if (matches(filter, endpoint)) {
matched.add(Map.entry(listener, filter));
}
}
}
}
// then trigger events without a lock
- for (Map.Entry<EndpointEventListener, String> entry : matched) {
+ for (Map.Entry<EndpointEventListener, Filter> entry : matched) {
entry.getKey().endpointChanged(event, entry.getValue().toString());
}
}
- private void triggerCallbacks(EndpointEventListener endpointListener, String filter,
- EndpointEvent event) {
- if (!matchFilter(filter, event.getEndpoint())) {
- return;
+ private void triggerCallbacks(EndpointEventListener endpointListener, Filter filter, EndpointEvent event) {
+ if (matches(filter, event.getEndpoint())) {
+ endpointListener.endpointChanged(event, filter.toString());
}
-
- endpointListener.endpointChanged(event, filter);
}
- private void triggerCallbacks(Collection<String> filters, EndpointEventListener endpointListener) {
- for (String filter : filters) {
+ private void triggerCallbacks(Collection<Filter> filters, EndpointEventListener endpointListener) {
+ for (Filter filter : filters) {
for (EndpointDescription endpoint : endpoints.values()) {
EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
triggerCallbacks(endpointListener, filter, event);
}
}
}
-
- private static boolean matchFilter(String filter, EndpointDescription endpoint) {
- if (filter == null) {
- return false;
- }
-
- try {
- Filter f = FrameworkUtil.createFilter(filter);
- Dictionary<String, Object> dict = new Hashtable<>(endpoint.getProperties());
- return f.match(dict);
- } catch (Exception e) {
- return false;
- }
- }
}
(aries-rsa) 02/23: Fix potential concurrency-related NPE when updating exported service properties
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 25cbbded4faeed396176343b4c7fe02603aab7ab
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 12:40:25 2026 +0200
Fix potential concurrency-related NPE when updating exported service properties
---
.../java/org/apache/aries/rsa/core/ExportRegistrationImpl.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
index 8b66c7ce..9a722885 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
@@ -182,19 +182,20 @@ public class ExportRegistrationImpl implements ExportRegistration {
@Override
public EndpointDescription update(Map<String, ?> properties) {
- if (getExportReference() == null) {
+ ExportReference ref = getExportReference();
+ if (ref == null) {
return null;
}
- ServiceReference<?> sref = getExportReference().getExportedService();
+ ServiceReference<?> sref = ref.getExportedService();
HashMap<String, Object> props = new HashMap<>(properties);
- EndpointDescription oldEpd = getExportReference().getExportedEndpoint();
+ EndpointDescription oldEpd = ref.getExportedEndpoint();
copyIfNull(props, oldEpd, RemoteConstants.ENDPOINT_ID);
copyIfNull(props, oldEpd, RemoteConstants.SERVICE_IMPORTED_CONFIGS);
EndpointDescription epd = new EndpointDescription(sref, props);
exportReference = new ExportReferenceImpl(sref, epd);
- this.sender.notifyUpdate(this.getExportReference());
+ this.sender.notifyUpdate(ref);
return epd;
}
(aries-rsa) 13/23: Remove unnecessary reverse map in ConfigDiscovery
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 9f049b5a7db1a95d460babb917d252f929bbf220
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 22:03:42 2026 +0200
Remove unnecessary reverse map in ConfigDiscovery
---
.../rsa/discovery/config/ConfigDiscovery.java | 52 +++++++---------------
1 file changed, 15 insertions(+), 37 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index 84cdfc40..a19f3db6 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -34,7 +34,6 @@ import java.util.concurrent.ConcurrentHashMap;
class ConfigDiscovery implements ManagedServiceFactory {
private final Map<EndpointDescription, String> endpointDescriptions = new ConcurrentHashMap<>();
private final Map<EndpointEventListener, Collection<String>> listenerToFilters = new HashMap<>();
- private final Map<String, Collection<EndpointEventListener>> filterToListeners = new HashMap<>();
@Override
public String getName() {
@@ -59,10 +58,6 @@ class ConfigDiscovery implements ManagedServiceFactory {
synchronized (listenerToFilters) {
listenerToFilters.put(endpointListener, filters);
- for (String filter : filters) {
- Collection<EndpointEventListener> listeners = filterToListeners.computeIfAbsent(filter, k -> new ArrayList<>());
- listeners.add(endpointListener);
- }
}
triggerCallbacks(filters, endpointListener);
@@ -70,35 +65,8 @@ class ConfigDiscovery implements ManagedServiceFactory {
void removeListener(EndpointEventListener endpointListener) {
synchronized (listenerToFilters) {
- Collection<String> filters = listenerToFilters.remove(endpointListener);
- if (filters == null) {
- return;
- }
-
- for (String filter : filters) {
- Collection<EndpointEventListener> listeners = filterToListeners.get(filter);
- if (listeners != null) {
- listeners.remove(endpointListener);
- if (listeners.isEmpty()) {
- filterToListeners.remove(filter);
- }
- }
- }
- }
- }
-
- private Map<String, Collection<EndpointEventListener>> getMatchingListeners(EndpointDescription endpoint) {
- // return a copy of matched filters/listeners so that caller doesn't need to hold locks while triggering events
- Map<String, Collection<EndpointEventListener>> matched = new HashMap<>();
- synchronized (listenerToFilters) {
- for (Map.Entry<String, Collection<EndpointEventListener>> entry : filterToListeners.entrySet()) {
- String filter = entry.getKey();
- if (matchFilter(filter, endpoint)) {
- matched.put(filter, new ArrayList<>(entry.getValue()));
- }
- }
+ listenerToFilters.remove(endpointListener);
}
- return matched;
}
@SuppressWarnings("rawtypes")
@@ -123,12 +91,22 @@ class ConfigDiscovery implements ManagedServiceFactory {
private void triggerCallbacks(EndpointEvent event) {
EndpointDescription endpoint = event.getEndpoint();
- for (Map.Entry<String, Collection<EndpointEventListener>> entry : getMatchingListeners(endpoint).entrySet()) {
- String filter = entry.getKey();
- for (EndpointEventListener listener : entry.getValue()) {
- triggerCallbacks(listener, filter, event);
+ // make a copy of matched filters/listeners so that caller doesn't need to hold locks while triggering events
+ List<Map.Entry<EndpointEventListener, String>> matched = new ArrayList<>();
+ synchronized (listenerToFilters) {
+ for (Map.Entry<EndpointEventListener, Collection<String>> entry : listenerToFilters.entrySet()) {
+ EndpointEventListener listener = entry.getKey();
+ for (String filter : entry.getValue()) {
+ if (matchFilter(filter, endpoint)) {
+ matched.add(Map.entry(listener, filter));
+ }
+ }
}
}
+ // then trigger events without a lock
+ for (Map.Entry<EndpointEventListener, String> entry : matched) {
+ entry.getKey().endpointChanged(event, entry.getValue().toString());
+ }
}
private void triggerCallbacks(EndpointEventListener endpointListener, String filter,
(aries-rsa) 16/23: Fix ConfigDiscovery sending ADDED event instead of MODIFIED
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit e44e28c0a802cee0a5556f27652d75b584341fe5
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 22:59:27 2026 +0200
Fix ConfigDiscovery sending ADDED event instead of MODIFIED
---
.../java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index 00532048..9f843488 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -86,8 +86,8 @@ class ConfigDiscovery implements ManagedServiceFactory {
@SuppressWarnings("rawtypes")
private void addDeclaredRemoteService(String pid, Dictionary config) {
EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(config));
- endpoints.put(pid, endpoint);
- triggerCallbacks(new EndpointEvent(EndpointEvent.ADDED, endpoint));
+ boolean isNew = endpoints.put(pid, endpoint) == null;
+ triggerCallbacks(new EndpointEvent(isNew ? EndpointEvent.ADDED : EndpointEvent.MODIFIED, endpoint));
}
private void removeServiceDeclaredInConfig(String pid) {
(aries-rsa) 03/23: Refactor ExportRegistrationImpl.update()
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 2a8d1c405bd4c5f3102b60a8349aa9aac8954cd6
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 12:48:18 2026 +0200
Refactor ExportRegistrationImpl.update()
---
.../apache/aries/rsa/core/ExportRegistrationImpl.java | 18 ++++++------------
1 file changed, 6 insertions(+), 12 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
index 9a722885..08149d3e 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
@@ -49,7 +49,7 @@ public class ExportRegistrationImpl implements ExportRegistration {
private int instanceCount;
private volatile boolean closed;
- private EventProducer sender;
+ private final EventProducer sender;
private ExportRegistrationImpl(ExportRegistrationImpl parent,
CloseHandler rsaCore,
@@ -186,22 +186,16 @@ public class ExportRegistrationImpl implements ExportRegistration {
if (ref == null) {
return null;
}
- ServiceReference<?> sref = ref.getExportedService();
- HashMap<String, Object> props = new HashMap<>(properties);
- EndpointDescription oldEpd = ref.getExportedEndpoint();
- copyIfNull(props, oldEpd, RemoteConstants.ENDPOINT_ID);
- copyIfNull(props, oldEpd, RemoteConstants.SERVICE_IMPORTED_CONFIGS);
+ Map<String, Object> oldProps = ref.getExportedEndpoint().getProperties();
+ Map<String, Object> props = new HashMap<>(properties);
+ props.putIfAbsent(RemoteConstants.ENDPOINT_ID, oldProps.get(RemoteConstants.ENDPOINT_ID));
+ props.putIfAbsent(RemoteConstants.SERVICE_IMPORTED_CONFIGS, oldProps.get(RemoteConstants.SERVICE_IMPORTED_CONFIGS));
+ ServiceReference<?> sref = ref.getExportedService();
EndpointDescription epd = new EndpointDescription(sref, props);
exportReference = new ExportReferenceImpl(sref, epd);
this.sender.notifyUpdate(ref);
return epd;
}
-
- private void copyIfNull(HashMap<String, Object> props, EndpointDescription oldEpd, String key) {
- if (props.get(key) == null) {
- props.put(key, oldEpd.getProperties().get(key));
- }
- }
}
(aries-rsa) 09/23: Cleanup TopologyManagerExport
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 14b6de837d5ca5aa5504176a91cb2f6dabaaf4a1
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 02:45:36 2026 +0200
Cleanup TopologyManagerExport
---
.../exporter/TopologyManagerExport.java | 50 +++++++++-------------
.../exporter/TopologyManagerExportTest.java | 28 +++++++++++-
2 files changed, 47 insertions(+), 31 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
index d2176588..5423ce87 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
@@ -61,22 +61,22 @@ public class TopologyManagerExport implements ServiceListener {
private final Executor executor;
private final ExportPolicy policy;
private final Map<RemoteServiceAdmin, ServiceExportsRepository> endpointRepo;
- private final Set<ServiceReference<?>> toBeExported;
+ private final Set<ServiceReference<?>> exportable;
public TopologyManagerExport(EndpointListenerNotifier notifier, Executor executor, ExportPolicy policy) {
this.notifier = notifier;
this.executor = executor;
this.policy = policy;
this.endpointRepo = new ConcurrentHashMap<>();
- this.toBeExported = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ this.exportable = Collections.newSetFromMap(new ConcurrentHashMap<>());
}
private String getTypeName(ServiceEvent event) {
switch (event.getType()) {
- case ServiceEvent.MODIFIED: return "modified";
- case ServiceEvent.MODIFIED_ENDMATCH: return "modified endmatch";
- case ServiceEvent.REGISTERED: return "registered";
- case ServiceEvent.UNREGISTERING: return "unregistering";
+ case ServiceEvent.MODIFIED: return "MODIFIED";
+ case ServiceEvent.MODIFIED_ENDMATCH: return "MODIFIED_ENDMATCH";
+ case ServiceEvent.REGISTERED: return "REGISTERED";
+ case ServiceEvent.UNREGISTERING: return "UNREGISTERING";
default: return null;
}
}
@@ -86,13 +86,13 @@ public class TopologyManagerExport implements ServiceListener {
public void serviceChanged(ServiceEvent event) {
ServiceReference<?> sref = event.getServiceReference();
if (!shouldExport(sref)) {
- LOG.debug("Skipping service {}", sref);
+ LOG.trace("Skipping service {}", sref);
return;
}
LOG.info("Received ServiceEvent type: {}, sref: {}", getTypeName(event), sref);
switch (event.getType()) {
case ServiceEvent.REGISTERED:
- doExport(sref);
+ add(sref);
break;
case ServiceEvent.MODIFIED:
@@ -106,24 +106,23 @@ public class TopologyManagerExport implements ServiceListener {
}
}
+ private void add(ServiceReference<?> sref) {
+ exportable.add(sref);
+ export(sref);
+ }
+
private void modified(ServiceReference<?> sref) {
- for (ServiceExportsRepository repo : endpointRepo.values()) {
- repo.modifyService(sref);
- }
+ endpointRepo.values().forEach(repo -> repo.modifyService(sref));
}
private void remove(ServiceReference<?> sref) {
- toBeExported.remove(sref);
- for (ServiceExportsRepository repo : endpointRepo.values()) {
- repo.removeService(sref);
- }
+ exportable.remove(sref);
+ endpointRepo.values().forEach(repo -> repo.removeService(sref));
}
public void add(RemoteServiceAdmin rsa) {
endpointRepo.put(rsa, new ServiceExportsRepository(rsa, notifier));
- for (ServiceReference<?> sref : toBeExported) {
- exportInBackground(sref);
- }
+ exportable.forEach(sref -> executor.execute(() -> export(sref)));
}
public void remove(RemoteServiceAdmin rsa) {
@@ -133,25 +132,16 @@ public class TopologyManagerExport implements ServiceListener {
}
}
- private void exportInBackground(final ServiceReference<?> sref) {
- executor.execute(() -> doExport(sref));
- }
-
- private void doExport(final ServiceReference<?> sref) {
+ private void export(final ServiceReference<?> sref) {
LOG.debug("Exporting service {}", sref);
- toBeExported.add(sref);
if (endpointRepo.isEmpty()) {
Bundle bundle = sref.getBundle();
String bundleName = bundle == null ? null : bundle.getSymbolicName();
- LOG.error("Unable to export service from bundle {}, interfaces: {} as no RemoteServiceAdmin is available. Marked for later export.",
+ LOG.warn("Unable to export service from bundle {}, interfaces {}, as no RemoteServiceAdmin is available. Marked for later export.",
bundleName, sref.getProperty(org.osgi.framework.Constants.OBJECTCLASS));
return;
}
-
- for (Map.Entry<RemoteServiceAdmin, ServiceExportsRepository> entry : endpointRepo.entrySet()) {
- Collection<ExportRegistration> regs = exportService(entry.getKey(), sref);
- entry.getValue().addService(sref, regs);
- }
+ endpointRepo.forEach((rsa, repo) -> repo.addService(sref, exportService(rsa, sref)));
}
private Collection<ExportRegistration> exportService(final RemoteServiceAdmin rsa, final ServiceReference<?> sref) {
diff --git a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExportTest.java b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExportTest.java
index f8a47c76..f34300b7 100644
--- a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExportTest.java
+++ b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExportTest.java
@@ -89,7 +89,7 @@ public class TopologyManagerExportTest {
expectServiceExported(sref2, epd2);
c.replay();
- exportManager.add(rsa);
+ exportManager.add(rsa); // rsa before services
exportManager.serviceChanged(new ServiceEvent(ServiceEvent.REGISTERED, sref));
exportManager.serviceChanged(new ServiceEvent(ServiceEvent.REGISTERED, sref2));
exportManager.serviceChanged(new ServiceEvent(ServiceEvent.MODIFIED, sref));
@@ -106,6 +106,32 @@ public class TopologyManagerExportTest {
MatcherAssert.assertThat(events.getValues(), sameBeanAs(expectedEvents));
}
+ @Test
+ public void testServiceExportUnexportWithRSAAddedAfterService() throws Exception {
+ EndpointDescription epd = createEndpoint();
+ ServiceReference sref = createUserService("*");
+ expectServiceExported(sref, epd);
+ EndpointDescription epd2 = createEndpoint();
+ ServiceReference sref2 = createUserService("*");
+ expectServiceExported(sref2, epd2);
+
+ c.replay();
+ exportManager.serviceChanged(new ServiceEvent(ServiceEvent.REGISTERED, sref));
+ exportManager.serviceChanged(new ServiceEvent(ServiceEvent.REGISTERED, sref2));
+ exportManager.add(rsa); // rsa after services
+ exportManager.serviceChanged(new ServiceEvent(ServiceEvent.MODIFIED, sref));
+ exportManager.remove(rsa);
+ c.verify();
+
+ List<EndpointEvent> expectedEvents = Arrays.asList(
+ new EndpointEvent(EndpointEvent.ADDED, epd),
+ new EndpointEvent(EndpointEvent.ADDED, epd2),
+ new EndpointEvent(EndpointEvent.MODIFIED, epd),
+ new EndpointEvent(EndpointEvent.REMOVED, epd),
+ new EndpointEvent(EndpointEvent.REMOVED, epd2));
+ MatcherAssert.assertThat(events.getValues(), sameBeanAs(expectedEvents));
+ }
+
@Test
public void testExportExistingMultipleInterfaces() throws Exception {
List<String> exportedInterfaces = Arrays.asList("a.b.C", "foo.Bar");
(aries-rsa) 20/23: Add MultiMap.replace method
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 291b2c04c6811417d0813154f1c9998d74ebb6b8
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 16:23:02 2026 +0200
Add MultiMap.replace method
---
.../apache/aries/rsa/topologymanager/importer/MultiMap.java | 11 +++++++++--
.../rsa/topologymanager/importer/TopologyManagerImport.java | 4 ++--
2 files changed, 11 insertions(+), 4 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/MultiMap.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/MultiMap.java
index e6b47f8d..69d28428 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/MultiMap.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/MultiMap.java
@@ -34,18 +34,25 @@ import java.util.stream.Collectors;
*/
public class MultiMap<K, V> {
- private Map<K, Set<V>> map;
+ private final Map<K, Set<V>> map;
public MultiMap() {
map = new ConcurrentHashMap<>();
}
public void put(K key, V value) {
+ replace(key, null, value);
+ }
+
+ public void replace(K key, V oldValue, V newValue) {
map.compute(key, (k, v) -> {
if (v == null) {
v = new CopyOnWriteArraySet<>();
}
- v.add(value);
+ if (oldValue != null) {
+ v.remove(oldValue);
+ }
+ v.add(newValue);
return v;
});
}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
index 247d334c..3f5de4a4 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
@@ -200,8 +200,8 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
importPossibilities.remove(filter, endpoint);
break;
case EndpointEvent.MODIFIED:
- importPossibilities.remove(filter, endpoint);
- importPossibilities.put(filter, endpoint);
+ // new endpoint has same endpoint.id and equals old endpoint, but has updated properties
+ importPossibilities.replace(filter, endpoint, endpoint);
break;
}
synchronizeImportsAsync(filter);
(aries-rsa) 18/23: Fix ConfigDiscovery not sending MODIFIED_ENDMATCH events
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 6b90f390697093b3e9221f4a3013df4b2160ba7e
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 00:57:16 2026 +0200
Fix ConfigDiscovery not sending MODIFIED_ENDMATCH events
---
.../aries/rsa/discovery/config/ConfigDiscovery.java | 21 ++++++++++++++++-----
1 file changed, 16 insertions(+), 5 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index d7d35b5f..052dd0f0 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -82,13 +82,24 @@ class ConfigDiscovery implements ManagedServiceFactory {
private void addDeclaredRemoteService(String pid, Dictionary<String, ?> config) {
EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(config));
- List<Map.Entry<Filter, EndpointEventListener>> matched;
- boolean isNew;
+ EndpointDescription old;
+ Set<Map.Entry<Filter, EndpointEventListener>> oldMatches, newMatches;
synchronized (this) {
- isNew = endpoints.put(pid, endpoint) == null;
- matched = findMatches(endpoint);
+ old = endpoints.put(pid, endpoint);
+ oldMatches = old == null ? null : new HashSet<>(findMatches(old));
+ newMatches = new HashSet<>(findMatches(endpoint));
+ }
+ List<Map.Entry<Filter, EndpointEventListener>> added = new ArrayList<>(newMatches);
+ if (old != null) {
+ added.removeAll(oldMatches);
+ List<Map.Entry<Filter, EndpointEventListener>> endmatch = new ArrayList<>(oldMatches);
+ endmatch.removeAll(newMatches);
+ List<Map.Entry<Filter, EndpointEventListener>> modified = new ArrayList<>(oldMatches);
+ modified.removeAll(endmatch);
+ triggerEvent(new EndpointEvent(EndpointEvent.MODIFIED, endpoint), modified);
+ triggerEvent(new EndpointEvent(EndpointEvent.MODIFIED_ENDMATCH, old), endmatch);
}
- triggerEvent(new EndpointEvent(isNew ? EndpointEvent.ADDED : EndpointEvent.MODIFIED, endpoint), matched);
+ triggerEvent(new EndpointEvent(EndpointEvent.ADDED, endpoint), added);
}
private void removeServiceDeclaredInConfig(String pid) {
(aries-rsa) 06/23: Fix ImportRegistrationImpl.update not returning updated endpoint
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 60bc1eab86e3eda82e2e0c71b5a6320db0ae9abd
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 16:30:30 2026 +0200
Fix ImportRegistrationImpl.update not returning updated endpoint
---
.../main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
index 86373143..f7151cce 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
@@ -228,7 +228,8 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
@SuppressWarnings("unchecked")
@Override
public boolean update(EndpointDescription endpoint) {
- importedService.setProperties(new Hashtable<>(endpoint.getProperties()));
+ this.endpoint = endpoint;
+ importedService.setProperties(new Hashtable(endpoint.getProperties()));
eventProducer.notifyUpdate(this);
return true;
}
(aries-rsa) 08/23: Cleanup RemoteServiceAdminCore
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 0003efa3dce5ac520c292f9513816c42c17281f8
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 20:28:47 2026 +0200
Cleanup RemoteServiceAdminCore
---
.../aries/rsa/core/RemoteServiceAdminCore.java | 95 ++++++++++------------
1 file changed, 41 insertions(+), 54 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
index 2c7489e0..a347d356 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
@@ -63,8 +63,8 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
private final Map<Map<String, Object>, Collection<ExportRegistration>> exportedServices = new LinkedHashMap<>();
private final Map<EndpointDescription, Collection<ImportRegistration>> importedServices = new LinkedHashMap<>();
- // Is stored in exportedServices while the export is in progress as a marker
- private final List<ExportRegistration> exportInProgress = Collections.emptyList();
+ // marker stored in exportedServices while the export is in progress
+ private final static List<ExportRegistration> EXPORT_IN_PROGRESS = Collections.emptyList();
private final BundleContext bctx;
private final EventProducer eventProducer;
@@ -101,25 +101,22 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
// listen for exported services being unregistered, so we can close the export
protected void createServiceListener() {
- this.exportedServiceListener = new ServiceListener() {
- public void serviceChanged(ServiceEvent event) {
- if (event.getType() == ServiceEvent.UNREGISTERING) {
- removeServiceExports(event.getServiceReference());
- }
+ this.exportedServiceListener = event -> {
+ if (event.getType() == ServiceEvent.UNREGISTERING) {
+ removeServiceExports(event.getServiceReference());
}
};
this.bctx.addServiceListener(exportedServiceListener);
}
@Override
- @SuppressWarnings({ "rawtypes", "unchecked" })
+ @SuppressWarnings("unchecked")
public List<ExportRegistration> exportService(ServiceReference serviceReference, Map additionalProperties)
throws IllegalArgumentException, UnsupportedOperationException {
Map<String, Object> serviceProperties = getProperties(serviceReference);
if (additionalProperties != null) {
overlayProperties(serviceProperties, additionalProperties);
}
- Map<String, Object> key = makeKey(serviceProperties);
List<String> interfaceNames = getInterfaceNames(serviceProperties);
@@ -127,22 +124,21 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
return Collections.emptyList();
}
- List<ExportRegistration> exportRegs = getExistingAndLock(key, interfaceNames);
- if (exportRegs != null) {
- return exportRegs;
- }
-
- try {
- ExportRegistration exportReg = exportService(interfaceNames, serviceReference, serviceProperties);
- exportRegs = new ArrayList<>();
- if (exportReg != null) {
- exportRegs.add(exportReg);
+ Map<String, Object> key = makeKey(serviceProperties);
+ List<ExportRegistration> regs = getExistingAndLock(key, interfaceNames);
+ if (regs == null) {
+ regs = new ArrayList<>();
+ try {
+ ExportRegistration reg = exportService(interfaceNames, serviceReference, serviceProperties);
+ if (reg != null) {
+ regs.add(reg);
+ store(key, regs);
+ }
+ } finally {
+ unlock(key);
}
- store(key, exportRegs);
- return exportRegs;
- } finally {
- unlock(key);
}
+ return regs;
}
private boolean isExportConfigSupported(Map<String, Object> serviceProperties) {
@@ -166,21 +162,17 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
return false;
}
- private void store(Map<String, Object> key, List<ExportRegistration> exportRegs) {
- if (!exportRegs.isEmpty()) {
- // enlist initial export registrations in global list of exportRegistrations
- synchronized (exportedServices) {
- exportedServices.put(key, new ArrayList<>(exportRegs));
- }
- eventProducer.publishNotification(exportRegs);
+ private void store(Map<String, Object> key, List<ExportRegistration> regs) {
+ // enlist initial export registrations in global list of exportRegistrations
+ synchronized (exportedServices) {
+ exportedServices.put(key, new ArrayList<>(regs));
}
+ eventProducer.publishNotification(regs);
}
private void unlock(Map<String, Object> key) {
synchronized (exportedServices) {
- if (exportedServices.get(key) == exportInProgress) {
- exportedServices.remove(key);
- }
+ exportedServices.remove(key, EXPORT_IN_PROGRESS);
exportedServices.notifyAll(); // in any case, always notify waiting threads
}
}
@@ -188,13 +180,13 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
private List<ExportRegistration> getExistingAndLock(Map<String, Object> key, List<String> interfaces) {
synchronized (exportedServices) {
// check if it is already exported...
- Collection<ExportRegistration> existingRegs = exportedServices.get(key);
+ Collection<ExportRegistration> regs = exportedServices.get(key);
// if the export is already in progress, wait for it to be complete
- while (existingRegs == exportInProgress) {
+ while (regs == EXPORT_IN_PROGRESS) {
try {
exportedServices.wait();
- existingRegs = exportedServices.get(key);
+ regs = exportedServices.get(key);
} catch (InterruptedException ie) {
LOG.debug("interrupted while waiting for export in progress");
return Collections.emptyList();
@@ -202,13 +194,13 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
// if the export is complete, return a copy of existing export
- if (existingRegs != null) {
- LOG.debug("already exported this service. Returning existing exportRegs {} ", interfaces);
- return copyExportRegistration(existingRegs);
+ if (regs != null) {
+ LOG.debug("already exported this service. Returning existing registrations {} ", interfaces);
+ return copyExportRegistration(regs);
}
// mark export as being in progress
- exportedServices.put(key, exportInProgress);
+ exportedServices.put(key, EXPORT_IN_PROGRESS);
}
return null;
}
@@ -237,11 +229,8 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
final Class<?>[] interfaces = getInterfaces(serviceO, interfaceNames);
final Map<String, Object> eprops = createEndpointProps(serviceProperties, interfaces);
- Endpoint endpoint = AccessController.doPrivileged(new PrivilegedAction<Endpoint>() {
- public Endpoint run() {
- return provider.exportService(serviceO, serviceContext, eprops, interfaces);
- }
- });
+ Endpoint endpoint = AccessController.doPrivileged(
+ (PrivilegedAction<Endpoint>) () -> provider.exportService(serviceO, serviceContext, eprops, interfaces));
if (endpoint == null) {
return null;
}
@@ -426,7 +415,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
return ir;
}
- if (determineConfigTypesForImport(endpoint).size() == 0) {
+ if (determineConfigTypesForImport(endpoint).isEmpty()) {
LOG.info("No matching handler can be found for remote endpoint {}.", endpoint.getId());
return null;
}
@@ -434,7 +423,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
// TODO: somehow select the interfaces that should be imported ---> job of the TopologyManager?
List<String> matchingInterfaces = endpoint.getInterfaces();
- if (matchingInterfaces.size() == 0) {
+ if (matchingInterfaces.isEmpty()) {
LOG.info("No matching interfaces found for remote endpoint {}.", endpoint.getId());
return null;
}
@@ -442,7 +431,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
LOG.info("Importing service {} with interfaces {} using handler {}.",
endpoint.getId(), endpoint.getInterfaces(), provider.getClass());
- ImportRegistrationImpl imReg = exposeServiceFactory(matchingInterfaces.toArray(new String[matchingInterfaces.size()]), endpoint, provider);
+ ImportRegistrationImpl imReg = exposeServiceFactory(matchingInterfaces.toArray(new String[0]), endpoint, provider);
if (imRegs == null) {
imRegs = new ArrayList<>();
importedServices.put(endpoint, imRegs);
@@ -463,7 +452,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
}
- if (usableConfigurationTypes.size() == 0) {
+ if (usableConfigurationTypes.isEmpty()) {
LOG.info("Ignoring endpoint {} as it has no compatible configuration types: {}.",
endpoint.getId(), remoteConfigurationTypes);
}
@@ -483,11 +472,9 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
ClientServiceFactory csf = new ClientServiceFactory(endpoint, handler, imReg);
imReg.setClientServiceFactory(csf);
- /**
- * Export the factory using the api context as it has very few imports.
- * If the bundle publishing the factory does not import the service interface
- * package then the factory is visible for all consumers which we want.
- */
+ // Export the factory using the api context as it has very few imports.
+ // If the bundle publishing the factory does not import the service interface
+ // package then the factory is visible for all consumers which we want.
ServiceRegistration<?> csfReg = apictx.registerService(interfaceNames, csf, serviceProps);
imReg.setImportedServiceRegistration(csfReg);
} catch (Exception ex) {
(aries-rsa) 19/23: ConfigDiscovery cleanup
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 69ab66efe88a79f6b5910f1f4ebf63d819b0d3fb
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 01:21:41 2026 +0200
ConfigDiscovery cleanup
---
.../rsa/discovery/config/ConfigDiscovery.java | 84 +++++++++++-----------
1 file changed, 40 insertions(+), 44 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index 052dd0f0..75932755 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -23,7 +23,6 @@ import org.osgi.framework.Filter;
import org.osgi.framework.FrameworkUtil;
import org.osgi.framework.InvalidSyntaxException;
import org.osgi.framework.ServiceReference;
-import org.osgi.service.cm.ConfigurationException;
import org.osgi.service.cm.ManagedServiceFactory;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
import org.osgi.service.remoteserviceadmin.EndpointEvent;
@@ -31,6 +30,11 @@ import org.osgi.service.remoteserviceadmin.EndpointEventListener;
import java.util.*;
+/**
+ * Monitors adding/updating/removing of configuration instances under the factory pid,
+ * as well as adding/removing of EndpointEventListeners, and triggers EndpointEvents
+ * accordingly as endpoints are created/updated/removed (via the configuration changes).
+ */
class ConfigDiscovery implements ManagedServiceFactory {
private final Map<String, EndpointDescription> endpoints = new HashMap<>();
private final Map<EndpointEventListener, Collection<Filter>> listenerToFilters = new HashMap<>();
@@ -41,16 +45,43 @@ class ConfigDiscovery implements ManagedServiceFactory {
}
@Override
- public void updated(String pid, Dictionary<String, ?> properties) throws ConfigurationException {
- addDeclaredRemoteService(pid, properties);
+ public void updated(String pid, Dictionary<String, ?> properties) {
+ EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(properties));
+ EndpointDescription old;
+ Set<Map.Entry<Filter, EndpointEventListener>> oldMatches, newMatches;
+ synchronized (this) {
+ old = endpoints.put(pid, endpoint);
+ oldMatches = old == null ? null : new HashSet<>(findMatches(old));
+ newMatches = new HashSet<>(findMatches(endpoint));
+ }
+ List<Map.Entry<Filter, EndpointEventListener>> added = new ArrayList<>(newMatches);
+ if (old != null) {
+ added.removeAll(oldMatches);
+ List<Map.Entry<Filter, EndpointEventListener>> endmatch = new ArrayList<>(oldMatches);
+ endmatch.removeAll(newMatches);
+ List<Map.Entry<Filter, EndpointEventListener>> modified = new ArrayList<>(oldMatches);
+ modified.removeAll(endmatch);
+ triggerEvents(new EndpointEvent(EndpointEvent.MODIFIED, endpoint), modified);
+ triggerEvents(new EndpointEvent(EndpointEvent.MODIFIED_ENDMATCH, old), endmatch);
+ }
+ triggerEvents(new EndpointEvent(EndpointEvent.ADDED, endpoint), added);
}
@Override
public void deleted(String pid) {
- removeServiceDeclaredInConfig(pid);
+ EndpointDescription endpoint;
+ List<Map.Entry<Filter, EndpointEventListener>> matched;
+ synchronized (this) {
+ endpoint = endpoints.remove(pid);
+ if (endpoint == null) {
+ return;
+ }
+ matched = findMatches(endpoint);
+ }
+ triggerEvents(new EndpointEvent(EndpointEvent.REMOVED, endpoint), matched);
}
- private static List<Filter> createFilters(ServiceReference<EndpointEventListener> ref) {
+ private static Collection<Filter> createFilters(ServiceReference<EndpointEventListener> ref) {
List<String> values = StringPlus.normalize(ref.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE));
List<Filter> filters = new ArrayList<>(values.size());
for (String value : values) {
@@ -63,14 +94,14 @@ class ConfigDiscovery implements ManagedServiceFactory {
}
void addListener(ServiceReference<EndpointEventListener> ref, EndpointEventListener listener) {
- List<Filter> filters = createFilters(ref);
+ Collection<Filter> filters = createFilters(ref);
if (!filters.isEmpty()) {
List<Map.Entry<Filter, EndpointDescription>> matched;
synchronized (this) {
listenerToFilters.put(listener, filters);
matched = findMatches(filters);
}
- triggerEvent(listener, matched);
+ triggerEvents(listener, matched);
}
}
@@ -80,41 +111,6 @@ class ConfigDiscovery implements ManagedServiceFactory {
}
}
- private void addDeclaredRemoteService(String pid, Dictionary<String, ?> config) {
- EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(config));
- EndpointDescription old;
- Set<Map.Entry<Filter, EndpointEventListener>> oldMatches, newMatches;
- synchronized (this) {
- old = endpoints.put(pid, endpoint);
- oldMatches = old == null ? null : new HashSet<>(findMatches(old));
- newMatches = new HashSet<>(findMatches(endpoint));
- }
- List<Map.Entry<Filter, EndpointEventListener>> added = new ArrayList<>(newMatches);
- if (old != null) {
- added.removeAll(oldMatches);
- List<Map.Entry<Filter, EndpointEventListener>> endmatch = new ArrayList<>(oldMatches);
- endmatch.removeAll(newMatches);
- List<Map.Entry<Filter, EndpointEventListener>> modified = new ArrayList<>(oldMatches);
- modified.removeAll(endmatch);
- triggerEvent(new EndpointEvent(EndpointEvent.MODIFIED, endpoint), modified);
- triggerEvent(new EndpointEvent(EndpointEvent.MODIFIED_ENDMATCH, old), endmatch);
- }
- triggerEvent(new EndpointEvent(EndpointEvent.ADDED, endpoint), added);
- }
-
- private void removeServiceDeclaredInConfig(String pid) {
- EndpointDescription endpoint;
- List<Map.Entry<Filter, EndpointEventListener>> matched;
- synchronized (this) {
- endpoint = endpoints.remove(pid);
- if (endpoint == null) {
- return;
- }
- matched = findMatches(endpoint);
- }
- triggerEvent(new EndpointEvent(EndpointEvent.REMOVED, endpoint), matched);
- }
-
private List<Map.Entry<Filter, EndpointEventListener>> findMatches(EndpointDescription endpoint) {
// called with lock, makes a copy of matched filters/listeners so we can later trigger events without locks
List<Map.Entry<Filter, EndpointEventListener>> matched = new ArrayList<>();
@@ -144,14 +140,14 @@ class ConfigDiscovery implements ManagedServiceFactory {
return matched;
}
- private void triggerEvent(EndpointEvent event, List<Map.Entry<Filter, EndpointEventListener>> matched) {
+ private void triggerEvents(EndpointEvent event, List<Map.Entry<Filter, EndpointEventListener>> matched) {
// trigger events without holding a lock
for (Map.Entry<Filter, EndpointEventListener> entry : matched) {
entry.getValue().endpointChanged(event, entry.getKey().toString());
}
}
- private void triggerEvent(EndpointEventListener listener, List<Map.Entry<Filter, EndpointDescription>> matched) {
+ private void triggerEvents(EndpointEventListener listener, List<Map.Entry<Filter, EndpointDescription>> matched) {
// trigger events without holding a lock
for (Map.Entry<Filter, EndpointDescription> entry : matched) {
EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, entry.getValue());
(aries-rsa) 04/23: Fix ServiceExportsRepository retaining services with no exports
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 424ab659f64c03e1fb1300958ce02c80fcfac647
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 12:57:43 2026 +0200
Fix ServiceExportsRepository retaining services with no exports
---
.../rsa/topologymanager/exporter/ServiceExportsRepository.java | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
index 2840c397..5373cb78 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
@@ -91,20 +91,22 @@ public class ServiceExportsRepository implements Closeable {
@Override
public synchronized void close() {
LOG.debug("Closing registry for RemoteServiceAdmin {}", rsa.getClass().getName());
- for (ServiceReference<?> sref : exportsMap.keySet()) {
+ for (ServiceReference<?> sref : new ArrayList<>(exportsMap.keySet())) { // iterate over copy to avoid CME
removeService(sref);
}
}
public synchronized void addService(ServiceReference<?> sref, Collection<ExportRegistration> registrations) {
Collection<ExportRegistrationHolder> exports = new ArrayList<>(registrations.size());
- exportsMap.put(sref, exports);
for (ExportRegistration reg : registrations) {
ExportReference reference = reg.getExportReference();
if (reference != null) {
exports.add(new ExportRegistrationHolder(reg, reference.getExportedEndpoint()));
}
}
+ if (!exports.isEmpty()) {
+ exportsMap.put(sref, exports);
+ }
}
public synchronized void modifyService(ServiceReference<?> sref) {
@@ -117,12 +119,11 @@ public class ServiceExportsRepository implements Closeable {
}
public synchronized void removeService(ServiceReference<?> sref) {
- Collection<ExportRegistrationHolder> exports = exportsMap.get(sref);
+ Collection<ExportRegistrationHolder> exports = exportsMap.remove(sref);
if (exports != null) {
for (ExportRegistrationHolder reg : exports) {
reg.close();
}
- exports.clear();
}
}
(aries-rsa) 10/23: Move and cleanup EndpointEventListenerTracker
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 32ac095e09f1b2fbe472aaa3cdbce26be5855a71
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 03:02:20 2026 +0200
Move and cleanup EndpointEventListenerTracker
---
.../aries/rsa/topologymanager/Activator.java | 7 +++--
.../EndpointEventListenerTracker.java | 35 ++++++++++------------
.../exporter/TopologyManagerExport.java | 4 +--
3 files changed, 22 insertions(+), 24 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
index c80be7a2..bb07f761 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.aries.rsa.spi.ExportPolicy;
import org.apache.aries.rsa.topologymanager.exporter.DefaultExportPolicy;
+import org.apache.aries.rsa.topologymanager.exporter.EndpointEventListenerTracker;
import org.apache.aries.rsa.topologymanager.exporter.EndpointListenerNotifier;
import org.apache.aries.rsa.topologymanager.exporter.TopologyManagerExport;
import org.apache.aries.rsa.topologymanager.importer.NamedThreadFactory;
@@ -79,7 +80,7 @@ public class Activator implements BundleActivator {
private ServiceTracker<ExportPolicy, ExportPolicy> policyTracker;
private EndpointListenerManager endpointListenerManager;
- private EndpointEventListenerTracker epeListenerTracker;
+ private EndpointEventListenerTracker endpointEventListenerTracker;
public void start(final BundleContext bc) throws Exception {
Dictionary<String, String> props = new Hashtable<>();
@@ -122,14 +123,14 @@ public class Activator implements BundleActivator {
notifier = new EndpointListenerNotifier();
exportExecutor = new ThreadPoolExecutor(5, 10, 50, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(TopologyManagerExport.class));
exportManager = new TopologyManagerExport(notifier, exportExecutor, policy);
- epeListenerTracker = new EndpointEventListenerTracker(bc, exportManager);
+ endpointEventListenerTracker = new EndpointEventListenerTracker(bc, exportManager);
importManager = new TopologyManagerImport(bc);
endpointListenerManager = new EndpointListenerManager(bc, importManager);
endpointListenerManager.start();
rsaTracker = new RSATracker(bc, RemoteServiceAdmin.class, null);
bc.addServiceListener(exportManager);
rsaTracker.open();
- epeListenerTracker.open();
+ endpointEventListenerTracker.open();
exportExistingServices(bc);
importManager.start();
}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/EndpointEventListenerTracker.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
similarity index 53%
rename from topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/EndpointEventListenerTracker.java
rename to topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
index d0d8b8ed..40f744db 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/EndpointEventListenerTracker.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
@@ -16,10 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.aries.rsa.topologymanager;
+package org.apache.aries.rsa.topologymanager.exporter;
-import org.apache.aries.rsa.topologymanager.exporter.EndpointListenerNotifier;
-import org.apache.aries.rsa.topologymanager.exporter.TopologyManagerExport;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Filter;
import org.osgi.framework.FrameworkUtil;
@@ -28,41 +26,40 @@ import org.osgi.framework.ServiceReference;
import org.osgi.service.remoteserviceadmin.EndpointEventListener;
import org.osgi.util.tracker.ServiceTracker;
-@SuppressWarnings({ "deprecation", "rawtypes", "unchecked" })
-final class EndpointEventListenerTracker extends ServiceTracker<EndpointEventListener, EndpointEventListener> {
- private TopologyManagerExport tmExport;
+public final class EndpointEventListenerTracker extends ServiceTracker<EndpointEventListener, EndpointEventListener> {
+ private final TopologyManagerExport tme;
- EndpointEventListenerTracker(BundleContext context, TopologyManagerExport tmExport) {
+ public EndpointEventListenerTracker(BundleContext context, TopologyManagerExport tme) {
super(context, getFilter(), null);
- this.tmExport = tmExport;
+ this.tme = tme;
}
private static Filter getFilter() {
- String filterSt = String.format("(objectClass=%s)", EndpointEventListener.class.getName());
+ String filter = String.format("(objectClass=%s)", EndpointEventListener.class.getName());
try {
- return FrameworkUtil.createFilter(filterSt);
+ return FrameworkUtil.createFilter(filter);
} catch (InvalidSyntaxException e) {
throw new IllegalArgumentException(e.getMessage(), e);
}
}
@Override
- public EndpointEventListener addingService(ServiceReference reference) {
- EndpointEventListener listener = super.addingService(reference);
- this.tmExport.addEPListener(listener, EndpointListenerNotifier.filtersFromEEL(reference));
+ public EndpointEventListener addingService(ServiceReference sref) {
+ EndpointEventListener listener = super.addingService(sref);
+ tme.addEndpointEventListener(listener, EndpointListenerNotifier.filtersFromEEL(sref));
return listener;
}
@Override
- public void modifiedService(ServiceReference reference, EndpointEventListener listener) {
- this.tmExport.addEPListener(listener, EndpointListenerNotifier.filtersFromEEL(reference));
- super.modifiedService(reference, listener);
+ public void modifiedService(ServiceReference sref, EndpointEventListener listener) {
+ tme.addEndpointEventListener(listener, EndpointListenerNotifier.filtersFromEEL(sref));
+ super.modifiedService(sref, listener);
}
@Override
- public void removedService(ServiceReference reference, EndpointEventListener listener) {
- this.tmExport.removeEPListener(listener);
- super.removedService(reference, listener);
+ public void removedService(ServiceReference sref, EndpointEventListener listener) {
+ tme.removeEndpointEventListener(listener);
+ super.removedService(sref, listener);
}
}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
index 5423ce87..4e3b16e5 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/TopologyManagerExport.java
@@ -186,7 +186,7 @@ public class TopologyManagerExport implements ServiceListener {
return exported != null && !exported.isEmpty() || addExported != null && !addExported.isEmpty();
}
- public void addEPListener(EndpointEventListener listener, Set<Filter> filters) {
+ public void addEndpointEventListener(EndpointEventListener listener, Set<Filter> filters) {
Collection<EndpointDescription> endpoints = new ArrayList<>();
for (ServiceExportsRepository repo : endpointRepo.values()) {
endpoints.addAll(repo.getAllEndpoints());
@@ -194,7 +194,7 @@ public class TopologyManagerExport implements ServiceListener {
notifier.add(listener, filters, endpoints);
}
- public void removeEPListener(EndpointEventListener listener) {
+ public void removeEndpointEventListener(EndpointEventListener listener) {
notifier.remove(listener);
}
}
(aries-rsa) 12/23: Cleanup EndpointListenerNotifier
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 6a2f3f65cd205c983ab5a3c76edb8c6ce28dcf75
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 03:20:19 2026 +0200
Cleanup EndpointListenerNotifier
---
.../exporter/EndpointEventListenerTracker.java | 4 +-
.../exporter/EndpointListenerNotifier.java | 44 ++++++++++------------
.../exporter/EndpointListenerNotifierTest.java | 6 +--
3 files changed, 25 insertions(+), 29 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
index 40f744db..471eb22a 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointEventListenerTracker.java
@@ -46,13 +46,13 @@ public final class EndpointEventListenerTracker extends ServiceTracker<EndpointE
@Override
public EndpointEventListener addingService(ServiceReference sref) {
EndpointEventListener listener = super.addingService(sref);
- tme.addEndpointEventListener(listener, EndpointListenerNotifier.filtersFromEEL(sref));
+ tme.addEndpointEventListener(listener, EndpointListenerNotifier.createScopeFilters(sref));
return listener;
}
@Override
public void modifiedService(ServiceReference sref, EndpointEventListener listener) {
- tme.addEndpointEventListener(listener, EndpointListenerNotifier.filtersFromEEL(sref));
+ tme.addEndpointEventListener(listener, EndpointListenerNotifier.createScopeFilters(sref));
super.modifiedService(sref, listener);
}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifier.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifier.java
index c9e53f9e..ce2813dc 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifier.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifier.java
@@ -39,23 +39,22 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Tracks EndpointListeners and allows to notify them of endpoints.
+ * Tracks EndpointEventListeners and allows to notify them of endpoints.
*/
-@SuppressWarnings("deprecation")
public class EndpointListenerNotifier {
private static final Logger LOG = LoggerFactory.getLogger(EndpointListenerNotifier.class);
- private Map<EndpointEventListener, Set<Filter>> listeners;
+ private final Map<EndpointEventListener, Set<Filter>> listeners;
public EndpointListenerNotifier() {
this.listeners = new ConcurrentHashMap<>();
}
- public static Set<Filter> filtersFromEEL(ServiceReference<EndpointEventListener> sref) {
+ public static Set<Filter> createScopeFilters(ServiceReference<EndpointEventListener> sref) {
List<String> scopes = StringPlus.normalize(sref.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE));
- return getFilterSet(scopes);
+ return createScopeFilters(scopes);
}
- private static Set<Filter> getFilterSet(List<String> scopes) {
+ private static Set<Filter> createScopeFilters(List<String> scopes) {
Set<Filter> filters = new HashSet<>();
for (String scope : scopes) {
try {
@@ -67,33 +66,30 @@ public class EndpointListenerNotifier {
return filters;
}
- public void add(EndpointEventListener ep, Set<Filter> filters, Collection<EndpointDescription> endpoints) {
- LOG.debug("EndpointListener added");
- listeners.put(ep, filters);
+ public void add(EndpointEventListener listener, Set<Filter> filters, Collection<EndpointDescription> endpoints) {
+ LOG.debug("EndpointEventListener added");
+ listeners.put(listener, filters);
for (EndpointDescription endpoint : endpoints) {
EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
- notifyListener(event, ep, filters);
+ notifyListener(event, listener, filters);
}
}
- public void remove(EndpointEventListener ep) {
- LOG.debug("EndpointListener removed");
- listeners.remove(ep);
+ public void remove(EndpointEventListener listener) {
+ LOG.debug("EndpointEventListener removed");
+ listeners.remove(listener);
}
public void sendEvent(EndpointEvent event) {
- for (EndpointEventListener listener : listeners.keySet()) {
- Set<Filter> filters = listeners.get(listener);
- notifyListener(event, listener, filters);
- }
+ listeners.forEach((listener, filters) -> notifyListener(event, listener, filters));
}
/**
* Notifies an endpoint listener about endpoints being added or removed.
*
- * @param type specifies whether endpoints were added (true) or removed (false)
- * @param endpointListenerRef the ServiceReference of an EndpointListener to notify
- * @param endpoints the endpoints the listener should be notified about
+ * @param event the event to send
+ * @param listener the listener to send the event to
+ * @param filters the listener filters
*/
private void notifyListener(EndpointEvent event, EndpointEventListener listener, Set<Filter> filters) {
Set<Filter> matchingFilters = getMatchingFilters(filters, event.getEndpoint());
@@ -103,19 +99,19 @@ public class EndpointListenerNotifier {
}
private static Set<Filter> getMatchingFilters(Set<Filter> filters, EndpointDescription endpoint) {
- Set<Filter> matchingFilters = new HashSet<>();
+ Set<Filter> matches = new HashSet<>();
if (endpoint == null) {
- return matchingFilters;
+ return matches;
}
Dictionary<String, Object> dict = new Hashtable<>(endpoint.getProperties());
for (Filter filter : filters) {
if (filter.match(dict)) {
LOG.debug("Filter {} matches endpoint {}", filter, dict);
- matchingFilters.add(filter);
+ matches.add(filter);
} else {
LOG.trace("Filter {} does not match endpoint {}", filter, dict);
}
}
- return matchingFilters;
+ return matches;
}
}
diff --git a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifierTest.java b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifierTest.java
index b045b077..c97d6890 100644
--- a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifierTest.java
+++ b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/exporter/EndpointListenerNotifierTest.java
@@ -106,7 +106,7 @@ public class EndpointListenerNotifierTest {
@Test
public void testNormalizeScopeForSingleString() {
ServiceReference sr = createListenerServiceWithFilter("(myProp=A)");
- Set<Filter> res = EndpointListenerNotifier.filtersFromEEL(sr);
+ Set<Filter> res = EndpointListenerNotifier.createScopeFilters(sr);
assertEquals(1, res.size());
Filter filter = res.iterator().next();
filterMatches(filter);
@@ -116,7 +116,7 @@ public class EndpointListenerNotifierTest {
public void testNormalizeScopeForStringArray() {
String[] filters = {"(myProp=A)", "(otherProp=B)"};
ServiceReference sr = createListenerServiceWithFilter(filters);
- Set<Filter> res = EndpointListenerNotifier.filtersFromEEL(sr);
+ Set<Filter> res = EndpointListenerNotifier.createScopeFilters(sr);
assertEquals(filters.length, res.size());
Iterator<Filter> it = res.iterator();
Filter filter1 = it.next();
@@ -130,7 +130,7 @@ public class EndpointListenerNotifierTest {
public void testNormalizeScopeForCollection() {
Collection<String> collection = Arrays.asList("(myProp=A)", "(otherProp=B)");
ServiceReference sr = createListenerServiceWithFilter(collection);
- Set<Filter> res = EndpointListenerNotifier.filtersFromEEL(sr);
+ Set<Filter> res = EndpointListenerNotifier.createScopeFilters(sr);
Iterator<Filter> it = res.iterator();
Filter filter1 = it.next();
Filter filter2 = it.next();
(aries-rsa) 22/23: Fix TopologyManagerImport not unregistering listener when stopped
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit e5cf123780712cc28e39fbefa61cab2c95db2c36
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 16:50:16 2026 +0200
Fix TopologyManagerImport not unregistering listener when stopped
---
.../aries/rsa/topologymanager/importer/TopologyManagerImport.java | 7 ++++++-
.../rsa/topologymanager/importer/TopologyManagerImportTest.java | 3 ++-
2 files changed, 8 insertions(+), 2 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
index d962a816..e48ef198 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceRegistration;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
import org.osgi.service.remoteserviceadmin.EndpointEvent;
import org.osgi.service.remoteserviceadmin.EndpointEventListener;
@@ -51,6 +52,7 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
private final ExecutorService execService;
private final BundleContext bctx;
private final Set<RemoteServiceAdmin> rsaSet;
+ private ServiceRegistration<RemoteServiceAdminListener> rsaListenerRegistration;
private volatile boolean stopped;
/**
@@ -71,11 +73,14 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
public void start() {
stopped = false;
- bctx.registerService(RemoteServiceAdminListener.class, this, null);
+ rsaListenerRegistration = bctx.registerService(RemoteServiceAdminListener.class, this, null);
}
public void stop() {
stopped = true;
+ if (rsaListenerRegistration != null) {
+ rsaListenerRegistration.unregister();
+ }
execService.shutdown();
try {
execService.awaitTermination(10, TimeUnit.SECONDS);
diff --git a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
index 621c2b63..857729a9 100644
--- a/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
+++ b/topology-manager/src/test/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImportTest.java
@@ -44,10 +44,11 @@ public class TopologyManagerImportTest {
@SuppressWarnings({ "rawtypes", "unchecked" })
private BundleContext mockBundleContext(IMocksControl c) {
ServiceRegistration sreg = c.createMock(ServiceRegistration.class);
+ sreg.unregister();
BundleContext bc = c.createMock(BundleContext.class);
expect(bc.registerService(eq(RemoteServiceAdminListener.class),
anyObject(RemoteServiceAdminListener.class),
- anyObject())).andReturn(sreg).anyTimes();
+ anyObject())).andReturn(sreg).once();
return bc;
}
(aries-rsa) 01/23: Refactor and clarify TopologyManagerImport updates calculation
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 2460eba3c93feac01ef58df6ba201fc2e1f8451f
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Fri Mar 13 11:43:52 2026 +0200
Refactor and clarify TopologyManagerImport updates calculation
---
.../rsa/topologymanager/importer/ImportDiff.java | 72 ----------------------
.../importer/TopologyManagerImport.java | 66 +++++++++++++-------
2 files changed, 43 insertions(+), 95 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/ImportDiff.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/ImportDiff.java
deleted file mode 100644
index 468d2350..00000000
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/ImportDiff.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.aries.rsa.topologymanager.importer;
-
-import java.util.Objects;
-import java.util.Set;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.ImportReference;
-import org.osgi.service.remoteserviceadmin.ImportRegistration;
-
-public class ImportDiff {
- private Set<EndpointDescription> possible;
- private Set<ImportRegistration> imported;
-
- public ImportDiff(Set<EndpointDescription> possible, Set<ImportRegistration> imported) {
- this.possible = possible;
- this.imported = imported;
- }
-
- public Stream<ImportRegistration> getRemoved() {
- return imported.stream()
- .filter(this::toRemove);
- }
-
- public Stream<EndpointDescription> getAdded() {
- Set<EndpointDescription> importedEndpoints = importedEndpoints();
- return possible.stream()
- .filter(not(importedEndpoints::contains));
- }
-
- /**
- * Checks if the import registration is not possible anymore or closed
- *
- * @param ireg registration to check
- * @return
- */
- private boolean toRemove(ImportRegistration ireg) {
- ImportReference iref = ireg != null ? ireg.getImportReference() : null;
- return iref == null || !possible.contains(iref.getImportedEndpoint());
- }
-
- private Set<EndpointDescription> importedEndpoints() {
- return imported.stream()
- .map(ImportRegistration::getImportReference).filter(Objects::nonNull)
- .map(ImportReference::getImportedEndpoint).filter(Objects::nonNull)
- .collect(Collectors.toSet());
- }
-
- private static <T> Predicate<T> not(Predicate<T> t) {
- return t.negate();
- }
-}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
index fe53007d..247d334c 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/importer/TopologyManagerImport.java
@@ -18,13 +18,14 @@
*/
package org.apache.aries.rsa.topologymanager.importer;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
import org.osgi.framework.BundleContext;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
@@ -63,7 +64,7 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
private final MultiMap<String, ImportRegistration> importedServices = new MultiMap<>();
public TopologyManagerImport(BundleContext bc) {
- this.rsaSet = new CopyOnWriteArraySet<>();
+ rsaSet = new CopyOnWriteArraySet<>();
bctx = bc;
execService = new ThreadPoolExecutor(5, 10, 50, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory(getClass()));
}
@@ -101,7 +102,7 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
ImportReference ref = event.getImportReference();
if (event.getType() == RemoteServiceAdminEvent.IMPORT_UNREGISTRATION && ref != null) {
importedServices.allValues().stream()
- .filter(ir -> ref.equals(ir.getImportReference()))
+ .filter(reg -> ref.equals(reg.getImportReference()))
.forEach(this::unimportRegistration);
}
}
@@ -115,48 +116,67 @@ public class TopologyManagerImport implements EndpointEventListener, RemoteServi
/**
* Synchronizes the actual imports with the possible imports for the given filter,
- * i.e. unimports previously imported endpoints that are no longer possible,
+ * i.e. un-imports previously imported endpoints that are no longer valid or possible,
* and imports new possible endpoints that are not already imported.
- *
+ * <p>
* TODO but optional: if the service is already imported and the endpoint is still
- * in the list of possible imports check if a "better" endpoint is now in the list
+ * in the list of possible imports check if a "better" endpoint is now in the list.
*
* @param filter the filter whose endpoints are synchronized
*/
private void synchronizeImports(final String filter) {
try {
- ImportDiff diff = new ImportDiff(importPossibilities.get(filter), importedServices.get(filter));
- diff.getRemoved()
- .forEach(this::unimportRegistration);
- diff.getAdded()
- .flatMap(this::importService)
- .forEach(ir -> importedServices.put(filter, ir));
+ // we have a set of all current imports, and a set of all possible imports (with overlap)
+ Set<ImportRegistration> imported = importedServices.get(filter);
+ Set<EndpointDescription> possible = importPossibilities.get(filter);
+ // first we iterate over all current imports, and split them into two groups:
+ // - still valid (no null references) and possible (in possible set)
+ // - invalid (contain null references) or no longer possible (not in possible set)
+ // note that this part should be concurrency-safe (get every reference only once and don't modify anything)
+ Set<EndpointDescription> valid = new HashSet<>(); // imports that are still valid and possible
+ Set<ImportRegistration> invalid = new LinkedHashSet<>(); // imports that are no longer valid/possible
+ for (ImportRegistration reg : imported) {
+ ImportReference ref = reg.getImportReference();
+ EndpointDescription endpoint = ref == null ? null : ref.getImportedEndpoint();
+ // check if the currently imported endpoint is still valid and possible
+ if (endpoint != null && possible.contains(endpoint)) {
+ valid.add(endpoint); // valid and possible
+ } else {
+ invalid.add(reg); // invalid (reg or ref or endpoint is null) or no longer possible
+ }
+ }
+ // now that we figured out what needs to be done, apply the changes
+ invalid.forEach(this::unimportRegistration); // remove invalid/non-possible imports
+ possible.forEach(endpoint -> { // import all possible endpoints that are not already imported
+ if (!valid.contains(endpoint)) {
+ importService(filter, endpoint);
+ }
+ });
} catch (Exception e) {
- LOG.error(e.getMessage(), e);
+ LOG.error("error synchronizing imports", e);
}
// Notify EndpointListeners? NO!
}
/**
- * Tries to import the service with each rsa until one import is successful.
+ * Tries to import the service with each RSA until one import is successful.
*
* @param filter the filter that matched the endpoint
* @param endpoint endpoint to import
- * @return
*/
- private Stream<ImportRegistration> importService(EndpointDescription endpoint) {
+ private void importService(String filter, EndpointDescription endpoint) {
for (RemoteServiceAdmin rsa : rsaSet) {
- ImportRegistration ir = rsa.importService(endpoint);
- if (ir != null) {
- if (ir.getException() == null) {
- LOG.debug("Service import was successful {}", ir);
- return Stream.of(ir);
+ ImportRegistration reg = rsa.importService(endpoint);
+ if (reg != null) {
+ if (reg.getException() == null) {
+ LOG.debug("Service import was successful {}", reg);
+ importedServices.put(filter, reg);
+ return;
} else {
- LOG.info("Error importing service {}", endpoint, ir.getException());
+ LOG.info("Error importing service {}", endpoint, reg.getException());
}
}
}
- return Stream.empty();
}
private void unimportRegistration(ImportRegistration reg) {
(aries-rsa) 14/23: Simplify ConfigDiscovery by reversing endpoints map
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 6d59fdf7f9b2ac2f07bff5caeeceff7f76a172bf
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 22:11:39 2026 +0200
Simplify ConfigDiscovery by reversing endpoints map
---
.../aries/rsa/discovery/config/ConfigDiscovery.java | 20 +++++++-------------
1 file changed, 7 insertions(+), 13 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index a19f3db6..0223a45b 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -32,7 +32,7 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
class ConfigDiscovery implements ManagedServiceFactory {
- private final Map<EndpointDescription, String> endpointDescriptions = new ConcurrentHashMap<>();
+ private final Map<String, EndpointDescription> endpoints = new ConcurrentHashMap<>();
private final Map<EndpointEventListener, Collection<String>> listenerToFilters = new HashMap<>();
@Override
@@ -72,20 +72,14 @@ class ConfigDiscovery implements ManagedServiceFactory {
@SuppressWarnings("rawtypes")
private void addDeclaredRemoteService(String pid, Dictionary config) {
EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(config));
- endpointDescriptions.put(endpoint, pid);
- EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
- triggerCallbacks(event);
+ endpoints.put(pid, endpoint);
+ triggerCallbacks(new EndpointEvent(EndpointEvent.ADDED, endpoint));
}
private void removeServiceDeclaredInConfig(String pid) {
- for (Iterator<Map.Entry<EndpointDescription, String>> i = endpointDescriptions.entrySet().iterator();
- i.hasNext(); ) {
- Map.Entry<EndpointDescription, String> entry = i.next();
- if (pid.equals(entry.getValue())) {
- EndpointEvent event = new EndpointEvent(EndpointEvent.REMOVED, entry.getKey());
- triggerCallbacks(event);
- i.remove();
- }
+ EndpointDescription endpoint = endpoints.remove(pid);
+ if (endpoint != null) {
+ triggerCallbacks(new EndpointEvent(EndpointEvent.REMOVED, endpoint));
}
}
@@ -120,7 +114,7 @@ class ConfigDiscovery implements ManagedServiceFactory {
private void triggerCallbacks(Collection<String> filters, EndpointEventListener endpointListener) {
for (String filter : filters) {
- for (EndpointDescription endpoint : endpointDescriptions.keySet()) {
+ for (EndpointDescription endpoint : endpoints.values()) {
EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
triggerCallbacks(endpointListener, filter, event);
}
(aries-rsa) 05/23: Refactor ServiceExportsRepository
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit b6eeaf559705b5f65d3f1fbdfb53ddd3a3c3d3da
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Sun Mar 15 16:28:41 2026 +0200
Refactor ServiceExportsRepository
---
.../exporter/ServiceExportsRepository.java | 112 ++++++++-------------
1 file changed, 42 insertions(+), 70 deletions(-)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
index 5373cb78..179df20c 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
@@ -20,6 +20,7 @@ package org.apache.aries.rsa.topologymanager.exporter;
import java.io.Closeable;
import java.util.*;
+import java.util.stream.Collectors;
import org.osgi.framework.ServiceReference;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
@@ -39,104 +40,75 @@ public class ServiceExportsRepository implements Closeable {
private final RemoteServiceAdmin rsa;
private final EndpointListenerNotifier notifier;
- private final Map<ServiceReference<?>, Collection<ExportRegistrationHolder>> exportsMap = new LinkedHashMap<>();
-
- /**
- * The holder allows us to work around that registration.getReference() is null when the registration is closed.
- */
- private class ExportRegistrationHolder {
- private final ExportRegistration registration;
- private final ExportReference reference;
- private EndpointDescription endpoint;
-
- ExportRegistrationHolder(ExportRegistration registration, EndpointDescription endpoint) {
- this.registration = registration;
- this.reference = registration.getExportReference();
- this.endpoint = endpoint;
- notifier.sendEvent(new EndpointEvent(EndpointEvent.ADDED, endpoint));
- }
-
- ExportRegistration getRegistration() {
- return registration;
- }
-
- void close() {
- if (reference != null) {
- notifier.sendEvent(new EndpointEvent(EndpointEvent.REMOVED, endpoint));
- registration.close();
- }
- }
-
- public void update(ServiceReference<?> sref) {
- if (reference != null) {
- endpoint = registration.update(getServiceProps(sref));
- notifier.sendEvent(new EndpointEvent(EndpointEvent.MODIFIED, endpoint));
- }
- }
-
- private Map<String, ?> getServiceProps(ServiceReference<?> sref) {
- HashMap<String, Object> props = new HashMap<>();
- for (String key : sref.getPropertyKeys()) {
- props.put(key, sref.getProperty(key));
- }
- return props;
- }
- }
+ // holds the set of exports for each service, and also caches the endpoint associated with each export,
+ // so we'll be able to access it even after it becomes unreachable via the export when it is closed
+ private final Map<ServiceReference<?>, Map<ExportRegistration, EndpointDescription>> exports = new LinkedHashMap<>();
public ServiceExportsRepository(RemoteServiceAdmin rsa, EndpointListenerNotifier notifier) {
this.rsa = rsa;
this.notifier = notifier;
}
+ private static Map<String, Object> getServiceProps(ServiceReference<?> sref) {
+ HashMap<String, Object> props = new HashMap<>();
+ for (String key : sref.getPropertyKeys()) {
+ props.put(key, sref.getProperty(key));
+ }
+ return props;
+ }
+
@Override
public synchronized void close() {
LOG.debug("Closing registry for RemoteServiceAdmin {}", rsa.getClass().getName());
- for (ServiceReference<?> sref : new ArrayList<>(exportsMap.keySet())) { // iterate over copy to avoid CME
- removeService(sref);
- }
+ new ArrayList<>(exports.keySet()).forEach(this::removeService); // iterate over copy to avoid CME
}
public synchronized void addService(ServiceReference<?> sref, Collection<ExportRegistration> registrations) {
- Collection<ExportRegistrationHolder> exports = new ArrayList<>(registrations.size());
+ Map<ExportRegistration, EndpointDescription> regs = new LinkedHashMap<>();
for (ExportRegistration reg : registrations) {
- ExportReference reference = reg.getExportReference();
- if (reference != null) {
- exports.add(new ExportRegistrationHolder(reg, reference.getExportedEndpoint()));
+ ExportReference ref = reg.getExportReference();
+ EndpointDescription endpoint = ref == null ? null : ref.getExportedEndpoint();
+ if (endpoint != null) {
+ regs.put(reg, endpoint);
+ notifier.sendEvent(new EndpointEvent(EndpointEvent.ADDED, endpoint));
}
}
- if (!exports.isEmpty()) {
- exportsMap.put(sref, exports);
+ if (!regs.isEmpty()) {
+ exports.put(sref, regs);
}
}
public synchronized void modifyService(ServiceReference<?> sref) {
- Collection<ExportRegistrationHolder> exports = exportsMap.get(sref);
- if (exports != null) {
- for (ExportRegistrationHolder reg : exports) {
- reg.update(sref);
+ Map<ExportRegistration, EndpointDescription> regs = exports.get(sref);
+ if (regs != null) {
+ Map<String, ?> props = getServiceProps(sref);
+ for (Map.Entry<ExportRegistration, EndpointDescription> entry: regs.entrySet()) {
+ EndpointDescription updated = entry.getKey().update(props);
+ if (updated != null) {
+ entry.setValue(updated);
+ notifier.sendEvent(new EndpointEvent(EndpointEvent.MODIFIED, updated));
+ }
}
}
}
public synchronized void removeService(ServiceReference<?> sref) {
- Collection<ExportRegistrationHolder> exports = exportsMap.remove(sref);
- if (exports != null) {
- for (ExportRegistrationHolder reg : exports) {
+ Map<ExportRegistration, EndpointDescription> regs = exports.remove(sref);
+ if (regs != null) {
+ regs.forEach((reg, endpoint) -> {
+ notifier.sendEvent(new EndpointEvent(EndpointEvent.REMOVED, endpoint));
reg.close();
- }
+ });
}
}
public synchronized List<EndpointDescription> getAllEndpoints() {
- List<EndpointDescription> endpoints = new ArrayList<>();
- for (Collection<ExportRegistrationHolder> exports : exportsMap.values()) {
- for (ExportRegistrationHolder reg : exports) {
- ExportReference reference = reg.getRegistration().getExportReference();
- if (reference != null) {
- endpoints.add(reference.getExportedEndpoint());
- }
- }
- }
- return endpoints;
+ return exports.values().stream()
+ .flatMap(regs -> regs.keySet().stream())
+ .map(ExportRegistration::getExportReference)
+ .filter(Objects::nonNull)
+ .map(ExportReference::getExportedEndpoint)
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
}
}
(aries-rsa) 17/23: Fix race condition in ConfigDiscovery that can trigger duplicate events
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 302c58a67129fa57a43113dc32ce08317cf88408
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Tue Mar 17 00:35:16 2026 +0200
Fix race condition in ConfigDiscovery that can trigger duplicate events
---
.../rsa/discovery/config/ConfigDiscovery.java | 94 +++++++++++++---------
1 file changed, 55 insertions(+), 39 deletions(-)
diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
index 9f843488..d7d35b5f 100644
--- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
+++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/ConfigDiscovery.java
@@ -30,10 +30,9 @@ import org.osgi.service.remoteserviceadmin.EndpointEvent;
import org.osgi.service.remoteserviceadmin.EndpointEventListener;
import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
class ConfigDiscovery implements ManagedServiceFactory {
- private final Map<String, EndpointDescription> endpoints = new ConcurrentHashMap<>();
+ private final Map<String, EndpointDescription> endpoints = new HashMap<>();
private final Map<EndpointEventListener, Collection<Filter>> listenerToFilters = new HashMap<>();
@Override
@@ -63,72 +62,89 @@ class ConfigDiscovery implements ManagedServiceFactory {
return filters;
}
- private static boolean matches(Filter filter, EndpointDescription endpoint) {
- return filter.match(new Hashtable<>(endpoint.getProperties())); // don't use matches() which is case-sensitive
- }
-
void addListener(ServiceReference<EndpointEventListener> ref, EndpointEventListener listener) {
List<Filter> filters = createFilters(ref);
if (!filters.isEmpty()) {
- synchronized (listenerToFilters) {
+ List<Map.Entry<Filter, EndpointDescription>> matched;
+ synchronized (this) {
listenerToFilters.put(listener, filters);
+ matched = findMatches(filters);
}
- triggerCallbacks(filters, listener);
+ triggerEvent(listener, matched);
}
}
void removeListener(EndpointEventListener listener) {
- synchronized (listenerToFilters) {
+ synchronized (this) {
listenerToFilters.remove(listener);
}
}
- @SuppressWarnings("rawtypes")
- private void addDeclaredRemoteService(String pid, Dictionary config) {
+ private void addDeclaredRemoteService(String pid, Dictionary<String, ?> config) {
EndpointDescription endpoint = new EndpointDescription(PropertyValidator.validate(config));
- boolean isNew = endpoints.put(pid, endpoint) == null;
- triggerCallbacks(new EndpointEvent(isNew ? EndpointEvent.ADDED : EndpointEvent.MODIFIED, endpoint));
+ List<Map.Entry<Filter, EndpointEventListener>> matched;
+ boolean isNew;
+ synchronized (this) {
+ isNew = endpoints.put(pid, endpoint) == null;
+ matched = findMatches(endpoint);
+ }
+ triggerEvent(new EndpointEvent(isNew ? EndpointEvent.ADDED : EndpointEvent.MODIFIED, endpoint), matched);
}
private void removeServiceDeclaredInConfig(String pid) {
- EndpointDescription endpoint = endpoints.remove(pid);
- if (endpoint != null) {
- triggerCallbacks(new EndpointEvent(EndpointEvent.REMOVED, endpoint));
+ EndpointDescription endpoint;
+ List<Map.Entry<Filter, EndpointEventListener>> matched;
+ synchronized (this) {
+ endpoint = endpoints.remove(pid);
+ if (endpoint == null) {
+ return;
+ }
+ matched = findMatches(endpoint);
}
+ triggerEvent(new EndpointEvent(EndpointEvent.REMOVED, endpoint), matched);
}
- private void triggerCallbacks(EndpointEvent event) {
- EndpointDescription endpoint = event.getEndpoint();
- // make a copy of matched filters/listeners so that caller doesn't need to hold locks while triggering events
- List<Map.Entry<EndpointEventListener, Filter>> matched = new ArrayList<>();
- synchronized (listenerToFilters) {
- for (Map.Entry<EndpointEventListener, Collection<Filter>> entry : listenerToFilters.entrySet()) {
- EndpointEventListener listener = entry.getKey();
- for (Filter filter : entry.getValue()) {
- if (matches(filter, endpoint)) {
- matched.add(Map.entry(listener, filter));
- }
+ private List<Map.Entry<Filter, EndpointEventListener>> findMatches(EndpointDescription endpoint) {
+ // called with lock, makes a copy of matched filters/listeners so we can later trigger events without locks
+ List<Map.Entry<Filter, EndpointEventListener>> matched = new ArrayList<>();
+ Dictionary<String, Object> props = new Hashtable<>(endpoint.getProperties());
+ for (Map.Entry<EndpointEventListener, Collection<Filter>> entry : listenerToFilters.entrySet()) {
+ EndpointEventListener listener = entry.getKey();
+ for (Filter filter : entry.getValue()) {
+ if (filter.match(props)) { // don't use matches() which is case-sensitive
+ matched.add(Map.entry(filter, listener));
}
}
}
- // then trigger events without a lock
- for (Map.Entry<EndpointEventListener, Filter> entry : matched) {
- entry.getKey().endpointChanged(event, entry.getValue().toString());
+ return matched;
+ }
+
+ private List<Map.Entry<Filter, EndpointDescription>> findMatches(Collection<Filter> filters) {
+ // called with lock, makes a copy of matched filters/endpoints so we can later trigger events without locks
+ List<Map.Entry<Filter, EndpointDescription>> matched = new ArrayList<>();
+ for (EndpointDescription endpoint : endpoints.values()) {
+ Dictionary<String, Object> props = new Hashtable<>(endpoint.getProperties());
+ for (Filter filter : filters) {
+ if (filter.match(props)) { // don't use matches() which is case-sensitive
+ matched.add(Map.entry(filter, endpoint));
+ }
+ }
}
+ return matched;
}
- private void triggerCallbacks(EndpointEventListener endpointListener, Filter filter, EndpointEvent event) {
- if (matches(filter, event.getEndpoint())) {
- endpointListener.endpointChanged(event, filter.toString());
+ private void triggerEvent(EndpointEvent event, List<Map.Entry<Filter, EndpointEventListener>> matched) {
+ // trigger events without holding a lock
+ for (Map.Entry<Filter, EndpointEventListener> entry : matched) {
+ entry.getValue().endpointChanged(event, entry.getKey().toString());
}
}
- private void triggerCallbacks(Collection<Filter> filters, EndpointEventListener endpointListener) {
- for (Filter filter : filters) {
- for (EndpointDescription endpoint : endpoints.values()) {
- EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint);
- triggerCallbacks(endpointListener, filter, event);
- }
+ private void triggerEvent(EndpointEventListener listener, List<Map.Entry<Filter, EndpointDescription>> matched) {
+ // trigger events without holding a lock
+ for (Map.Entry<Filter, EndpointDescription> entry : matched) {
+ EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, entry.getValue());
+ listener.endpointChanged(event, entry.getKey().toString());
}
}
}
(aries-rsa) 11/23: Fix EndpointEventListenerTracker not being closed
Posted by am...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
amichai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 140bf2de9c4be5e7ef6ac4ba0ee56063b1a52923
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 16 03:04:09 2026 +0200
Fix EndpointEventListenerTracker not being closed
---
.../src/main/java/org/apache/aries/rsa/topologymanager/Activator.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
index bb07f761..daef5a33 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/Activator.java
@@ -146,6 +146,7 @@ public class Activator implements BundleActivator {
importManager.stop();
endpointListenerManager.stop();
rsaTracker.close();
+ endpointEventListenerTracker.close();
exportManager = null;
}