This is an automated email from the ASF dual-hosted git repository. amichair pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
commit 608556f3ade1a898f2deb5759b8d7af27183dfbd Author: Amichai Rothman <[email protected]> AuthorDate: Fri Jun 5 11:14:54 2026 +0300 ARIES-2233 Extract common EndpointEventListener service tracking into InterestManager --- .../aries/rsa/discovery/config/Activator.java | 42 +++-------------- .../rsa/discovery/config/ConfigDiscovery.java | 4 +- .../aries/rsa/discovery/local/LocalDiscovery.java | 20 +------- .../rsa/discovery/local/LocalDiscoveryTest.java | 55 +++++++++++++++------- .../aries/rsa/discovery/mdns/InterestManager.java | 4 +- .../aries/rsa/discovery/mdns/MdnsDiscovery.java | 25 ++-------- .../aries/rsa/discovery/tcp/TcpDiscovery.java | 19 +------- .../rsa/discovery/zookeeper/InterestManager.java | 25 ++++------ .../discovery/zookeeper/InterestManagerTest.java | 12 ++--- .../zookeeper/ZookeeperDiscoveryTest.java | 2 +- .../aries/rsa/spi/discovery/InterestManager.java | 52 ++++++++++++++++++++ 11 files changed, 124 insertions(+), 136 deletions(-) diff --git a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/Activator.java b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/Activator.java index 117d71c1..f05b0f7d 100644 --- a/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/Activator.java +++ b/discovery/config/src/main/java/org/apache/aries/rsa/discovery/config/Activator.java @@ -20,15 +20,13 @@ package org.apache.aries.rsa.discovery.config; import java.util.Hashtable; +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.osgi.annotation.bundle.Header; import org.osgi.framework.BundleActivator; import org.osgi.framework.BundleContext; import org.osgi.framework.Constants; -import org.osgi.framework.ServiceReference; import org.osgi.framework.ServiceRegistration; import org.osgi.service.cm.ManagedServiceFactory; -import org.osgi.service.remoteserviceadmin.EndpointEventListener; -import org.osgi.util.tracker.ServiceTracker; @Header(name = Constants.BUNDLE_ACTIVATOR, value = "${@class}") @org.osgi.annotation.bundle.Capability( // @@ -39,13 +37,13 @@ import org.osgi.util.tracker.ServiceTracker; public class Activator implements BundleActivator { private static final String FACTORY_PID = "org.apache.aries.rsa.discovery.config"; - private ServiceTracker<EndpointEventListener, EndpointEventListener> listenerTracker; + private InterestManager interestManager; private ServiceRegistration<ManagedServiceFactory> registration; public void start(BundleContext context) { - ConfigDiscovery configDiscovery = new ConfigDiscovery(); - listenerTracker = new EPListenerTracker(context, configDiscovery); - listenerTracker.open(); + interestManager = new InterestManager(); + ConfigDiscovery configDiscovery = new ConfigDiscovery(interestManager); + interestManager.start(context, null); Hashtable<String, Object> props = new Hashtable<>(); props.put(Constants.SERVICE_PID, FACTORY_PID); registration = context.registerService(ManagedServiceFactory.class, configDiscovery, props); @@ -53,34 +51,6 @@ public class Activator implements BundleActivator { public void stop(BundleContext context) { registration.unregister(); - listenerTracker.close(); - } - - private final class EPListenerTracker extends ServiceTracker<EndpointEventListener, EndpointEventListener> { - private final ConfigDiscovery configDiscovery; - - private EPListenerTracker(BundleContext context, ConfigDiscovery configDiscovery) { - super(context, EndpointEventListener.class, null); - this.configDiscovery = configDiscovery; - } - - @Override - public EndpointEventListener addingService(ServiceReference<EndpointEventListener> reference) { - EndpointEventListener service = super.addingService(reference); - configDiscovery.getInterestManager().addListener(reference, service); - return service; - } - - @Override - public void modifiedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { - super.modifiedService(reference, service); - configDiscovery.getInterestManager().updateListener(reference, service); - } - - @Override - public void removedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { - super.removedService(reference, service); - configDiscovery.getInterestManager().removeListener(reference); - } + interestManager.stop(); } } 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 d783fb7c..1c2ac43f 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,8 +32,8 @@ import java.util.*; class ConfigDiscovery implements ManagedServiceFactory { private final InterestManager interestManager; - public ConfigDiscovery() { - this.interestManager = new InterestManager(); + public ConfigDiscovery(InterestManager interestManager) { + this.interestManager = interestManager; } @Override diff --git a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java index 81aaf957..07178f89 100644 --- a/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java +++ b/discovery/local/src/main/java/org/apache/aries/rsa/discovery/local/LocalDiscovery.java @@ -25,15 +25,10 @@ import org.osgi.framework.Bundle; import org.osgi.framework.BundleContext; import org.osgi.framework.BundleEvent; import org.osgi.framework.BundleListener; -import org.osgi.framework.ServiceReference; import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; import org.osgi.service.component.annotations.Deactivate; -import org.osgi.service.component.annotations.Reference; -import org.osgi.service.component.annotations.ReferenceCardinality; -import org.osgi.service.component.annotations.ReferencePolicy; import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -55,11 +50,13 @@ public class LocalDiscovery implements BundleListener { public void activate(BundleContext context) { context.addBundleListener(this); processExistingBundles(context.getBundles()); + interestManager.start(context, null); } @Deactivate public void deactivate(BundleContext context) { context.removeBundleListener(this); + interestManager.stop(); } protected void processExistingBundles(Bundle[] bundles) { @@ -70,19 +67,6 @@ public class LocalDiscovery implements BundleListener { } } - @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC) - void bindListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.addListener(sref, listener); - } - - void unbindListener(ServiceReference<EndpointEventListener> sref) { - interestManager.removeListener(sref); - } - - void updatedListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.updateListener(sref, listener); - } - // BundleListener method @Override public void bundleChanged(BundleEvent event) { diff --git a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java index 7931fe99..3d39bcd6 100644 --- a/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java +++ b/discovery/local/src/test/java/org/apache/aries/rsa/discovery/local/LocalDiscoveryTest.java @@ -19,8 +19,10 @@ package org.apache.aries.rsa.discovery.local; import static org.easymock.EasyMock.anyObject; +import static org.easymock.EasyMock.anyString; import static org.easymock.EasyMock.createMock; import static org.easymock.EasyMock.expect; +import static org.easymock.EasyMock.getCurrentArguments; import static org.easymock.EasyMock.replay; import static org.junit.Assert.assertEquals; @@ -38,12 +40,16 @@ import java.util.Set; import java.util.function.Function; import java.util.stream.Collectors; +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.easymock.EasyMock; import org.junit.Test; import org.osgi.framework.Bundle; import org.osgi.framework.BundleContext; import org.osgi.framework.BundleEvent; import org.osgi.framework.BundleListener; +import org.osgi.framework.FrameworkUtil; +import org.osgi.framework.InvalidSyntaxException; +import org.osgi.framework.ServiceListener; import org.osgi.framework.ServiceReference; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEvent; @@ -120,7 +126,8 @@ public class LocalDiscoveryTest { LocalDiscovery ld = new LocalDiscovery(); Listener listener = new Listener(); - ld.bindListener(listener.getServiceReference(), listener); + InterestManager im = ld.interestManager; + im.addListener(listener.getServiceReference(), listener); ld.activate(context); // should get all endpoints from existing bundles on activation @@ -137,7 +144,8 @@ public class LocalDiscoveryTest { LocalDiscovery ld = new LocalDiscovery(); Listener listener = new Listener(); - ld.bindListener(listener.getServiceReference(), listener); + InterestManager im = ld.interestManager; + im.addListener(listener.getServiceReference(), listener); // install the bundle ld.bundleChanged(new BundleEvent(BundleEvent.INSTALLED, bundle)); @@ -162,18 +170,19 @@ public class LocalDiscoveryTest { public void testEndpointListenerLifecycle() { LocalDiscovery ld = new LocalDiscovery(); Listener listener = new Listener(); - ld.bindListener(listener.getServiceReference(), listener); + InterestManager im = ld.interestManager; + im.addListener(listener.getServiceReference(), listener); Bundle bundle = mockBundle(); // created with two endpoint descriptions, ClassA and ClassB ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); assertEquals(2, listener.getEndpoints().size()); // remove listener - ld.unbindListener(listener.getServiceReference()); + im.removeListener(listener.getServiceReference()); // add new listener with scope A listener = new Listener("(objectClass=org.example.ClassA)"); - ld.bindListener(listener.getServiceReference(), listener); + im.addListener(listener.getServiceReference(), listener); Collection<String> filters = listener.getFilters(); assertEquals(1, filters.size()); @@ -181,9 +190,9 @@ public class LocalDiscoveryTest { // unbind the listener, modify its scope to A+B, and bind again listener.setFilter("(|(objectClass=org.example.ClassA)(objectClass=org.example.ClassB))"); - ld.unbindListener(listener.getServiceReference()); + im.removeListener(listener.getServiceReference()); listener.clear(); - ld.bindListener(listener.getServiceReference(), listener); + im.addListener(listener.getServiceReference(), listener); Set<String> expectedInterfaces = new HashSet<>(Arrays.asList("org.example.ClassA", "org.example.ClassB")); Set<String> interfaces = listener.getEndpointData(EndpointDescription::getInterfaces).stream() @@ -194,38 +203,39 @@ public class LocalDiscoveryTest { // now change the scope to C via updated listener service properties listener.setFilter("(objectClass=org.example.ClassC)"); listener.clear(); - ld.updatedListener(listener.getServiceReference(), listener); + im.updateListener(listener.getServiceReference(), listener); assertEquals(0, listener.getEndpoints().size()); // and update again to scope B listener.setFilter("(objectClass=org.example.ClassB)"); listener.clear(); - ld.updatedListener(listener.getServiceReference(), listener); + im.updateListener(listener.getServiceReference(), listener); assertEquals(1, listener.getEndpoints().size()); // remove the EndpointListener service - ld.unbindListener(listener.getServiceReference()); + im.removeListener(listener.getServiceReference()); } @Test public void testRegisterListener() { LocalDiscovery ld = new LocalDiscovery(); Listener listener = new Listener("(objectClass=org.example.ClassA)"); - ld.bindListener(listener.getServiceReference(), listener); + InterestManager im = ld.interestManager; + im.addListener(listener.getServiceReference(), listener); assertEquals("Precondition failed", 0, listener.getEndpoints().size()); - ld.bindListener(listener.getServiceReference(), listener); + im.addListener(listener.getServiceReference(), listener); // add another one with the same scope filter Listener listener2 = new Listener("(objectClass=org.example.ClassA)"); - ld.bindListener(listener2.getServiceReference(), listener2); + im.addListener(listener2.getServiceReference(), listener2); // add another listener with a multi-value scope Listener listener3 = new Listener("(objectClass=org.example.ClassA)", "(objectClass=org.example.ClassB)"); - ld.bindListener(listener3.getServiceReference(), listener3); + im.addListener(listener3.getServiceReference(), listener3); Bundle bundle = mockBundle(); ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); @@ -239,14 +249,15 @@ public class LocalDiscoveryTest { @Test public void testUnregisterListener() { LocalDiscovery ld = new LocalDiscovery(); + InterestManager im = ld.interestManager; // start with two listeners Listener listener = new Listener("(objectClass=org.example.ClassA)"); - ld.bindListener(listener.getServiceReference(), listener); + im.addListener(listener.getServiceReference(), listener); Listener listener2 = new Listener("(objectClass=org.example.ClassA)"); - ld.bindListener(listener2.getServiceReference(), listener2); + im.addListener(listener2.getServiceReference(), listener2); // remove first listener and start bundle - ld.unbindListener(listener.getServiceReference()); + im.removeListener(listener.getServiceReference()); Bundle bundle = mockBundle(); ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); @@ -262,7 +273,7 @@ public class LocalDiscoveryTest { assertEquals(0, listener2.getEndpoints().size()); // remove the second bundle and start the bundle - ld.unbindListener(listener2.getServiceReference()); + im.removeListener(listener2.getServiceReference()); bundle = mockBundle(); ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); @@ -311,6 +322,14 @@ public class LocalDiscoveryTest { expect(context.getBundles()).andReturn(bundles).anyTimes(); context.addBundleListener(anyObject(BundleListener.class)); context.removeBundleListener(anyObject(BundleListener.class)); + try { + expect(context.createFilter(anyString())) + .andAnswer(() -> FrameworkUtil.createFilter((String)getCurrentArguments()[0])); + context.addServiceListener(anyObject(ServiceListener.class), anyString()); + expect(context.getServiceReferences(anyString(), anyString())).andReturn(null); + } catch (InvalidSyntaxException e) { + throw new RuntimeException(e); + } replay(context); return context; } diff --git a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/InterestManager.java b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/InterestManager.java index f808f506..4b973217 100644 --- a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/InterestManager.java +++ b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/InterestManager.java @@ -58,7 +58,9 @@ public class InterestManager extends org.apache.aries.rsa.spi.discovery.Interest this.client = client; } - public void deactivate() { + @Override + public void stop() { + super.stop(); streams.values().forEach(SseEventSource::close); streams.clear(); } diff --git a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/MdnsDiscovery.java b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/MdnsDiscovery.java index 74ddcdc8..bee88eb2 100644 --- a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/MdnsDiscovery.java +++ b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/MdnsDiscovery.java @@ -41,17 +41,14 @@ import javax.ws.rs.client.ClientBuilder; import org.apache.aries.rsa.spi.EndpointDescriptionParser; import org.apache.aries.rsa.util.StringPlus; import org.osgi.framework.BundleContext; -import org.osgi.framework.ServiceReference; import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; import org.osgi.service.component.annotations.Deactivate; import org.osgi.service.component.annotations.Reference; -import org.osgi.service.component.annotations.ReferenceCardinality; import org.osgi.service.component.annotations.ReferencePolicy; import org.osgi.service.jaxrs.client.SseEventSourceFactory; import org.osgi.service.jaxrs.runtime.JaxrsServiceRuntime; import org.osgi.service.jaxrs.runtime.dto.RuntimeDTO; -import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -79,23 +76,11 @@ public class MdnsDiscovery { @Activate public MdnsDiscovery(BundleContext ctx, @Reference SseEventSourceFactory eventSourceFactory, @Reference ClientBuilder clientBuilder, @Reference EndpointDescriptionParser parser) { - this.client = clientBuilder.build(); - this.interestManager = new InterestManager(eventSourceFactory, parser, client); + client = clientBuilder.build(); + interestManager = new InterestManager(eventSourceFactory, parser, client); + interestManager.start(ctx, null); fwUuid = ctx.getProperty(FRAMEWORK_UUID); - this.publishingListener = new PublishingEndpointListener(parser, ctx, fwUuid); - } - - @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC) - public void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.addListener(sref, listener); - } - - public void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.updateListener(sref, listener); - } - - public void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { - interestManager.removeListener(sref); + publishingListener = new PublishingEndpointListener(parser, ctx, fwUuid); } @Reference(policy = ReferencePolicy.DYNAMIC) @@ -198,7 +183,7 @@ public class MdnsDiscovery { LOG.warn("An exception occurred closing the mdns discovery"); } - interestManager.deactivate(); + interestManager.stop(); publishingListener.stop(); } diff --git a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpDiscovery.java b/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpDiscovery.java index ff1ee771..e717bec5 100644 --- a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpDiscovery.java +++ b/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpDiscovery.java @@ -21,12 +21,10 @@ package org.apache.aries.rsa.discovery.tcp; import org.apache.aries.rsa.annotations.RSADiscoveryProvider; import org.apache.aries.rsa.spi.discovery.InterestManager; import org.osgi.framework.BundleContext; -import org.osgi.framework.ServiceReference; import org.osgi.framework.ServiceRegistration; import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; import org.osgi.service.component.annotations.Deactivate; -import org.osgi.service.component.annotations.Reference; import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -44,8 +42,6 @@ import java.util.Objects; import java.util.stream.Collectors; import static org.osgi.framework.Constants.FRAMEWORK_UUID; -import static org.osgi.service.component.annotations.ReferenceCardinality.MULTIPLE; -import static org.osgi.service.component.annotations.ReferencePolicy.DYNAMIC; import static org.osgi.service.remoteserviceadmin.EndpointEventListener.ENDPOINT_LISTENER_SCOPE; import static org.osgi.service.remoteserviceadmin.RemoteConstants.ENDPOINT_FRAMEWORK_UUID; @@ -179,6 +175,7 @@ public class TcpDiscovery { } catch (IOException ioe) { LOG.error("failed to start TCP connection manager", ioe); } + interestManager.start(context, OWN_LISTENER_PROP); } @Deactivate @@ -189,18 +186,6 @@ public class TcpDiscovery { if (connectionManager != null) { connectionManager.close(); } - } - - @Reference(cardinality = MULTIPLE, policy = DYNAMIC, target = "(!(" + OWN_LISTENER_PROP + "=*))") - void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.addListener(sref, listener); - } - - void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.updateListener(sref, listener); - } - - void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { - interestManager.removeListener(sref); + interestManager.stop(); } } diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java index 747aa034..a47dd0cb 100644 --- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java +++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/InterestManager.java @@ -21,12 +21,11 @@ package org.apache.aries.rsa.discovery.zookeeper; import org.apache.aries.rsa.discovery.zookeeper.client.ClientManager; import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointListener; import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointRepository; -import org.osgi.framework.ServiceReference; +import org.osgi.framework.BundleContext; +import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; import org.osgi.service.component.annotations.Deactivate; import org.osgi.service.component.annotations.Reference; -import org.osgi.service.component.annotations.ReferenceCardinality; -import org.osgi.service.component.annotations.ReferencePolicy; import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,25 +47,17 @@ public class InterestManager extends org.apache.aries.rsa.spi.discovery.Interest zkListener = repository.createListener(this); } + @Activate + public void activate(BundleContext context) { + start(context, ClientManager.DISCOVERY_ZOOKEEPER_ID); + } + @Deactivate public void deactivate() { + stop(); zkListener.close(); } - @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC, - target = "(!(" + ClientManager.DISCOVERY_ZOOKEEPER_ID + "=*))") - public void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - addListener(sref, listener); - } - - public void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - updateListener(sref, listener); - } - - public void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { - removeListener(sref); - } - int size() { return interests.size(); } diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java index 0a09d8b9..0757ee7f 100644 --- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java +++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/InterestManagerTest.java @@ -58,22 +58,22 @@ public class InterestManagerTest { // sref has no scope -> nothing should happen assertNumInterests(0); - im.bindEndpointEventListener(sref, epListener1); + im.addListener(sref, epListener1); assertNumInterests(1); - im.bindEndpointEventListener(sref, epListener1); + im.addListener(sref, epListener1); assertNumInterests(1); - im.bindEndpointEventListener(sref2, epListener2); + im.addListener(sref2, epListener2); assertNumInterests(2); - im.unbindEndpointEventListener(sref); + im.removeListener(sref); assertNumInterests(1); - im.unbindEndpointEventListener(sref); + im.removeListener(sref); assertNumInterests(1); - im.unbindEndpointEventListener(sref2); + im.removeListener(sref2); assertNumInterests(0); } diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java index ddac35a3..300a01b4 100644 --- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java +++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/ZookeeperDiscoveryTest.java @@ -99,7 +99,7 @@ public class ZookeeperDiscoveryTest { String scope = "("+ Constants.OBJECTCLASS +"=*)"; Mockito.when(sref.getProperty(Mockito.eq(EndpointEventListener.ENDPOINT_LISTENER_SCOPE))).thenReturn(scope); - im.bindEndpointEventListener(sref, this::onEndpointChanged); + im.addListener(sref, this::onEndpointChanged); assertThat(semConnected.tryAcquire(1, SECONDS), equalTo(true)); diff --git a/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java index af1b3538..d7d156b5 100644 --- a/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java +++ b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java @@ -18,9 +18,14 @@ */ package org.apache.aries.rsa.spi.discovery; +import org.osgi.framework.BundleContext; +import org.osgi.framework.Constants; +import org.osgi.framework.Filter; +import org.osgi.framework.InvalidSyntaxException; import org.osgi.framework.ServiceReference; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEventListener; +import org.osgi.util.tracker.ServiceTracker; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,6 +46,27 @@ public class InterestManager { protected final Map<ServiceReference<EndpointEventListener>, Interest> interests = new ConcurrentHashMap<>(); // source to endpointId to endpoint protected final Map<String , Map<String, EndpointDescription>> remoteEndpoints = new ConcurrentHashMap<>(); + // service tracker for finding interested EndpointEventListeners + protected ServiceTracker<EndpointEventListener, EndpointEventListener> tracker; + + public void start(BundleContext context, String excludeProperty) { + String classFilter = String.format("(%s=%s)", Constants.OBJECTCLASS, EndpointEventListener.class.getName()); + String filter = excludeProperty == null + ? classFilter + : String.format("(&%s(!(%s=*)))", classFilter, excludeProperty); + try { + tracker = new EndpointEventListenerServiceTracker(context, context.createFilter(filter)); + tracker.open(); + } catch (InvalidSyntaxException ise) { + throw new RuntimeException(ise); + } + } + + public void stop() { + if (tracker != null) { + tracker.close(); + } + } public void addListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { @@ -98,4 +124,30 @@ public class InterestManager { endpoints.values().forEach(endpoint -> notifyAllListeners(endpoint, null)); } } + + protected class EndpointEventListenerServiceTracker extends ServiceTracker<EndpointEventListener, EndpointEventListener> { + + public EndpointEventListenerServiceTracker(BundleContext context, Filter filter) { + super(context, filter, null); + } + + @Override + public EndpointEventListener addingService(ServiceReference<EndpointEventListener> sref) { + EndpointEventListener service = super.addingService(sref); + addListener(sref, service); + return service; + } + + @Override + public void modifiedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { + super.modifiedService(reference, service); + updateListener(reference, service); + } + + @Override + public void removedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { + super.removedService(reference, service); + removeListener(reference); + } + } }
