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 fa2ae18e1d490ef5e7619cc03a91e744211c5221 Author: Amichai Rothman <[email protected]> AuthorDate: Fri Jun 5 08:12:22 2026 +0300 ARIES-2233 Extract common InterestManager implementation for all discovery providers --- .../aries/rsa/discovery/config/Activator.java | 12 +- .../rsa/discovery/config/ConfigDiscovery.java | 119 ++--------------- .../aries/rsa/discovery/local/LocalDiscovery.java | 116 ++--------------- .../rsa/discovery/local/LocalDiscoveryTest.java | 11 +- .../apache/aries/rsa/discovery/mdns/Interest.java | 143 --------------------- .../aries/rsa/discovery/mdns/InterestManager.java | 65 +--------- .../aries/rsa/discovery/mdns/MdnsDiscovery.java | 13 +- .../rsa/discovery/tcp/TcpConnectionManager.java | 5 +- .../aries/rsa/discovery/tcp/TcpDiscovery.java | 5 +- .../rsa/discovery/zookeeper/InterestManager.java | 47 +------ .../client/ZookeeperEndpointListener.java | 35 +---- .../client/ZookeeperEndpointRepository.java | 6 +- .../discovery/zookeeper/InterestManagerTest.java | 3 - .../ZookeeperEndpointRepositoryPathTest.java | 1 - .../apache/aries/rsa/spi/discovery/Interest.java | 41 +++++- .../aries/rsa/spi/discovery}/InterestManager.java | 66 +++++----- 16 files changed, 136 insertions(+), 552 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 04fed288..117d71c1 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 @@ -67,26 +67,20 @@ public class Activator implements BundleActivator { @Override public EndpointEventListener addingService(ServiceReference<EndpointEventListener> reference) { EndpointEventListener service = super.addingService(reference); - configDiscovery.addListener(reference, service); + configDiscovery.getInterestManager().addListener(reference, service); return service; } @Override public void modifiedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { super.modifiedService(reference, service); - configDiscovery.removeListener(service); - - // This may cause duplicate registrations of remote services, - // but that's fine and should be filtered out on another level. - // See Remote Service Admin spec section 122.6.3 - configDiscovery.addListener(reference, service); + configDiscovery.getInterestManager().updateListener(reference, service); } @Override public void removedService(ServiceReference<EndpointEventListener> reference, EndpointEventListener service) { super.removedService(reference, service); - configDiscovery.removeListener(service); + configDiscovery.getInterestManager().removeListener(reference); } } - } 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 75932755..d783fb7c 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 @@ -18,15 +18,9 @@ */ 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.apache.aries.rsa.spi.discovery.InterestManager; import org.osgi.service.cm.ManagedServiceFactory; import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEvent; -import org.osgi.service.remoteserviceadmin.EndpointEventListener; import java.util.*; @@ -36,8 +30,11 @@ import java.util.*; * 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<>(); + private final InterestManager interestManager; + + public ConfigDiscovery() { + this.interestManager = new InterestManager(); + } @Override public String getName() { @@ -47,111 +44,15 @@ class ConfigDiscovery implements ManagedServiceFactory { @Override 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); + interestManager.addEndpoint(pid, endpoint); } @Override public void deleted(String 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 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) { - try { - filters.add(FrameworkUtil.createFilter(value)); - } catch (InvalidSyntaxException ignore) { // bad filter never matches - } - } - return filters; - } - - void addListener(ServiceReference<EndpointEventListener> ref, EndpointEventListener listener) { - Collection<Filter> filters = createFilters(ref); - if (!filters.isEmpty()) { - List<Map.Entry<Filter, EndpointDescription>> matched; - synchronized (this) { - listenerToFilters.put(listener, filters); - matched = findMatches(filters); - } - triggerEvents(listener, matched); - } - } - - void removeListener(EndpointEventListener listener) { - synchronized (this) { - listenerToFilters.remove(listener); - } - } - - 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)); - } - } - } - 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 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()); - } + interestManager.removeSource(pid); } - 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()); - listener.endpointChanged(event, entry.getKey().toString()); - } + public InterestManager getInterestManager() { + return interestManager; } } 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 5b0bdd4f..81aaf957 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 @@ -18,23 +18,13 @@ */ package org.apache.aries.rsa.discovery.local; -import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; -import java.util.Hashtable; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.aries.rsa.util.StringPlus; + +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.osgi.framework.Bundle; import org.osgi.framework.BundleContext; import org.osgi.framework.BundleEvent; import org.osgi.framework.BundleListener; -import org.osgi.framework.Filter; -import org.osgi.framework.FrameworkUtil; -import org.osgi.framework.InvalidSyntaxException; import org.osgi.framework.ServiceReference; import org.osgi.service.component.annotations.Activate; import org.osgi.service.component.annotations.Component; @@ -43,7 +33,6 @@ 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.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -53,27 +42,13 @@ public class LocalDiscovery implements BundleListener { private static final Logger LOG = LoggerFactory.getLogger(LocalDiscovery.class); - final Map<Bundle, Collection<EndpointDescription>> endpoints = new ConcurrentHashMap<>(); - final Map<EndpointEventListener, Collection<Filter>> listenerToFilters = new HashMap<>(); + final InterestManager interestManager; EndpointDescriptionBundleParser parser; public LocalDiscovery() { this.parser = new EndpointDescriptionBundleParser(); - } - - 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 ise) { - // invalid filter syntax means it can't match anything anyway, so we just ignore it - LOG.trace("ignoring invalid filter '{}'", value); - } - } - return filters; + this.interestManager = new InterestManager(); } @Activate @@ -97,58 +72,15 @@ public class LocalDiscovery implements BundleListener { @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC) void bindListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - List<Filter> filters = createFilters(sref); - if (filters.isEmpty()) { - return; - } - - synchronized (listenerToFilters) { - listenerToFilters.put(listener, filters); - } - - publishAllToListener(filters, listener); + interestManager.addListener(sref, listener); } - /** - * If the tracker was removed or the scope was changed this doesn't require - * additional callbacks on the tracker. It's the responsibility of the tracker - * itself to clean up any orphans. See Remote Service Admin spec 122.6.3 - * - * @param listener the listener being removed - */ - void unbindListener(EndpointEventListener listener) { - synchronized (listenerToFilters) { - listenerToFilters.remove(listener); - } + void unbindListener(ServiceReference<EndpointEventListener> sref) { + interestManager.removeListener(sref); } void updatedListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - // if service properties have been updated, the filter (scope) - // might have changed so we remove and re-add the listener - // TODO fix this so that we don't: - // 1. remove and add when there is no change - // 2. remove and add instead of modifying - // 3. remove instead of modified end match - synchronized (listenerToFilters) { - unbindListener(listener); - bindListener(sref, listener); - } - } - - private Map<Filter, Collection<EndpointEventListener>> getMatchingListeners(EndpointDescription endpoint) { - // return a copy of matched filters/listeners so that caller doesn't need to hold locks while triggering events - Hashtable<String, Object> props = new Hashtable<>(endpoint.getProperties()); - Map<Filter, Collection<EndpointEventListener>> matched = new HashMap<>(); - synchronized (listenerToFilters) { - for (Entry<EndpointEventListener, Collection<Filter>> entry : listenerToFilters.entrySet()) { - for (Filter filter : entry.getValue()) { - if (filter.match(props)) { // don't use matches() which is case-sensitive - matched.computeIfAbsent(filter, f -> new ArrayList<>()).add(entry.getKey()); - } - } - } - } - return matched; + interestManager.updateListener(sref, listener); } // BundleListener method @@ -167,38 +99,12 @@ public class LocalDiscovery implements BundleListener { private void addEndpoints(Bundle bundle) { Collection<EndpointDescription> endpoints = parser.getAllEndpointDescriptions(bundle); - if (!endpoints.isEmpty()) { - this.endpoints.put(bundle, endpoints); - endpoints.forEach(endpoint -> publishToAllListeners(new EndpointEvent(EndpointEvent.ADDED, endpoint))); + for (EndpointDescription endpoint : endpoints) { + interestManager.addEndpoint(String.valueOf(bundle.getBundleId()), endpoint); } } private void removeEndpoints(Bundle bundle) { - Collection<EndpointDescription> endpoints = this.endpoints.remove(bundle); - if (endpoints != null) { - endpoints.forEach(endpoint -> publishToAllListeners(new EndpointEvent(EndpointEvent.REMOVED, endpoint))); - } - } - - private void publishToAllListeners(EndpointEvent event) { - Map<Filter, Collection<EndpointEventListener>> matched = getMatchingListeners(event.getEndpoint()); - for (Map.Entry<Filter, Collection<EndpointEventListener>> entry : matched.entrySet()) { - for (EndpointEventListener listener : entry.getValue()) { - listener.endpointChanged(event, entry.getKey().toString()); - } - } - } - - private void publishAllToListener(Collection<Filter> filters, EndpointEventListener listener) { - endpoints.values().stream().flatMap(Collection::stream) - .forEach(endpoint -> { - EndpointEvent event = new EndpointEvent(EndpointEvent.ADDED, endpoint); - Hashtable<String, Object> props = new Hashtable<>(event.getEndpoint().getProperties()); - filters.forEach(filter -> { - if (filter.match(props)) { - listener.endpointChanged(event, filter.toString()); - } - }); - }); + interestManager.removeSource(String.valueOf(bundle.getBundleId())); } } 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 845ab279..7931fe99 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 @@ -169,7 +169,7 @@ public class LocalDiscoveryTest { assertEquals(2, listener.getEndpoints().size()); // remove listener - ld.unbindListener(listener); + ld.unbindListener(listener.getServiceReference()); // add new listener with scope A listener = new Listener("(objectClass=org.example.ClassA)"); @@ -181,7 +181,7 @@ 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); + ld.unbindListener(listener.getServiceReference()); listener.clear(); ld.bindListener(listener.getServiceReference(), listener); @@ -206,7 +206,7 @@ public class LocalDiscoveryTest { assertEquals(1, listener.getEndpoints().size()); // remove the EndpointListener service - ld.unbindListener(listener); + ld.unbindListener(listener.getServiceReference()); } @Test @@ -246,7 +246,7 @@ public class LocalDiscoveryTest { ld.bindListener(listener2.getServiceReference(), listener2); // remove first listener and start bundle - ld.unbindListener(listener); + ld.unbindListener(listener.getServiceReference()); Bundle bundle = mockBundle(); ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); @@ -262,7 +262,7 @@ public class LocalDiscoveryTest { assertEquals(0, listener2.getEndpoints().size()); // remove the second bundle and start the bundle - ld.unbindListener(listener2); + ld.unbindListener(listener2.getServiceReference()); bundle = mockBundle(); ld.bundleChanged(new BundleEvent(BundleEvent.STARTED, bundle)); @@ -291,6 +291,7 @@ public class LocalDiscoveryTest { .map(r -> LocalDiscoveryTest.class.getResource("/" + r)) .collect(Collectors.toList()); Bundle bundle = createMock(Bundle.class); + expect(bundle.getBundleId()).andReturn(Math.round(Math.random())).anyTimes(); expect(bundle.getState()).andReturn(Bundle.ACTIVE).anyTimes(); Dictionary<String, String> headers = new Hashtable<>(); headers.put("Remote-Service", "OSGI-INF/rsa/"); diff --git a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/Interest.java b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/Interest.java deleted file mode 100644 index 1fa851c5..00000000 --- a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/Interest.java +++ /dev/null @@ -1,143 +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.discovery.mdns; - -import static org.osgi.service.remoteserviceadmin.EndpointEvent.ADDED; -import static org.osgi.service.remoteserviceadmin.EndpointEvent.MODIFIED; -import static org.osgi.service.remoteserviceadmin.EndpointEvent.MODIFIED_ENDMATCH; -import static org.osgi.service.remoteserviceadmin.EndpointEvent.REMOVED; -import static org.osgi.service.remoteserviceadmin.EndpointEventListener.ENDPOINT_LISTENER_SCOPE; - -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.atomic.AtomicReference; - -import org.apache.aries.rsa.util.StringPlus; -import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEvent; -import org.osgi.service.remoteserviceadmin.EndpointEventListener; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -@SuppressWarnings("deprecation") -public class Interest { - private static final Logger LOG = LoggerFactory.getLogger(Interest.class); - - private final Long id; - private final ConcurrentMap<String, EndpointDescription> added = new ConcurrentHashMap<>(); - private final AtomicReference<List<String>> scopes = new AtomicReference<>(); - private final EndpointEventListener listener; - - public Interest(Long id, EndpointEventListener listener, Map<String, Object> props) { - this.id = id; - this.scopes.set(StringPlus.normalize(props.get(ENDPOINT_LISTENER_SCOPE))); - this.listener = listener; - } - - public void update(Map<String, Object> props) { - List<String> newScopes = StringPlus.normalize(props.get(ENDPOINT_LISTENER_SCOPE)); - List<String> oldScopes = this.scopes.getAndSet(newScopes); - - added.values().removeIf(ed -> { - Optional<String> newScope = getFirstMatch(ed, newScopes); - Optional<String> oldScope = getFirstMatch(ed, oldScopes); - EndpointEvent event; - boolean remove; - String filter; - if (newScope.isPresent()) { - remove = false; - filter = newScope.get(); - if (oldScope.isPresent() && oldScope.get().equals(filter)) { - event = null; - } else { - event = new EndpointEvent(MODIFIED, ed); - } - } else { - remove = true; - event = new EndpointEvent(REMOVED, ed); - filter = oldScope.orElse(null); - } - - if (event != null) - notifyListener(event, filter); - - return remove; - }); - } - - public void endpointChanged(EndpointDescription ed) { - List<String> scopes = this.scopes.get(); - Optional<String> currentScope = getFirstMatch(ed, scopes); - boolean alreadyAdded = added.containsKey(ed.getId()); - EndpointEvent event; - String filter; - if (currentScope.isPresent()) { - if (LOG.isDebugEnabled()) { - LOG.debug("Listener {} is interested in endpoint {}. It will be {}", id, ed, alreadyAdded ? "MODIFIED" : "ADDED"); - } - added.put(ed.getId(), ed); - event = new EndpointEvent(alreadyAdded ? MODIFIED : ADDED, ed); - filter = currentScope.get(); - } else if (alreadyAdded) { - if (LOG.isDebugEnabled()) { - LOG.debug("Listener {} is no longer interested in endpoint {}. It will be {}", id, ed, "MODIFIED"); - } - EndpointDescription previous = added.remove(ed.getId()); - event = new EndpointEvent(MODIFIED_ENDMATCH, ed); - filter = getFirstMatch(previous, scopes).orElse(null); - } else { - if (LOG.isDebugEnabled()) { - LOG.debug("Listener {} not interested in endpoint {}", id, ed); - } - return; - } - - notifyListener(event, filter); - } - - public void endpointRemoved(String id) { - EndpointDescription previous = added.remove(id); - if (previous != null) { - if (LOG.isDebugEnabled()) { - LOG.debug("Endpoint {} is no longer available for listener {}", id, this.id); - } - notifyListener(new EndpointEvent(REMOVED, previous), getFirstMatch(previous, scopes.get()).orElse(null)); - } - } - - private void notifyListener(EndpointEvent event, String filter) { - EndpointDescription endpoint = event.getEndpoint(); - LOG.info("Calling endpointChanged on class {} for filter {}, type {}, endpoint {} ", - listener, filter, event.getType(), endpoint); - listener.endpointChanged(event, filter); - } - - private Optional<String> getFirstMatch(EndpointDescription endpoint, List<String> scopes) { - return scopes.stream().filter(endpoint::matches).findFirst(); - } - - @Override - public String toString() { - return "Interest [scopes=" + scopes + ", listener=" + listener.getClass() + "]"; - } - -} 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 549054e4..f808f506 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 @@ -18,15 +18,10 @@ */ package org.apache.aries.rsa.discovery.mdns; -import static java.util.Collections.singleton; -import static java.util.stream.Collectors.toSet; import static org.apache.aries.rsa.discovery.mdns.PublishingEndpointListener.Subscription.ENDPOINT_REVOKED; import static org.apache.aries.rsa.discovery.mdns.PublishingEndpointListener.Subscription.ENDPOINT_UPDATED; -import static org.apache.aries.rsa.util.CollectionUtils.union; import java.io.InputStream; -import java.util.Map; -import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -35,7 +30,6 @@ import javax.ws.rs.sse.InboundSseEvent; import javax.ws.rs.sse.SseEventSource; import org.apache.aries.rsa.spi.EndpointDescriptionParser; -import org.osgi.framework.Constants; import org.osgi.service.jaxrs.client.SseEventSourceFactory; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEventListener; @@ -47,19 +41,15 @@ import org.slf4j.LoggerFactory; * Establishes SSE event sources to be called back on all changes in the remote targets. * Events are then forwarded to all interested {@link EndpointEventListener}s. */ -public class InterestManager { +public class InterestManager extends org.apache.aries.rsa.spi.discovery.InterestManager { private static final Logger LOG = LoggerFactory.getLogger(InterestManager.class); - private final ConcurrentMap<Long, Interest> interests = new ConcurrentHashMap<>(); - private final SseEventSourceFactory eventSourceFactory; private final EndpointDescriptionParser parser; private final Client client; - private final ConcurrentMap<String, Set<EndpointDescription>> endpointsBySource = new ConcurrentHashMap<>(); - private final ConcurrentMap<String, SseEventSource> streams = new ConcurrentHashMap<>(); public InterestManager(SseEventSourceFactory factory, EndpointDescriptionParser parser, Client client) { @@ -71,7 +61,6 @@ public class InterestManager { public void deactivate() { streams.values().forEach(SseEventSource::close); streams.clear(); - interests.clear(); } public void remoteAdded(String uri) { @@ -98,6 +87,7 @@ public class InterestManager { if (sseEventSource != null) { sseEventSource.close(); } + removeSource(uri); } private void onEndpointEvent(String source, InboundSseEvent event) { @@ -109,19 +99,10 @@ public class InterestManager { if (ENDPOINT_UPDATED.equals(name)) { EndpointDescription ed = parser.readEndpoint(event.readData(InputStream.class)); - endpointsBySource.compute(source, (a,b) -> b == null ? singleton(ed) : union(b, singleton(ed))); - interests.values().forEach(i -> i.endpointChanged(ed)); + addEndpoint(source, ed); } else if (ENDPOINT_REVOKED.equals(name)) { String id = event.readData(); - endpointsBySource.compute(source, (a,b) -> { - if (b == null) { - return null; - } else { - Set<EndpointDescription> set = b.stream().filter(ed -> !ed.getId().equals(id)).collect(toSet()); - return set.isEmpty() ? null : set; - } - }); - interests.values().forEach(i -> i.endpointRemoved(id)); + removeEndpoint(source, id); } } @@ -136,42 +117,6 @@ public class InterestManager { } } - Set<EndpointDescription> remove = endpointsBySource.remove(source); - if (remove != null) { - remove.forEach(ed -> interests.values().forEach(i -> i.endpointRemoved(ed.getId()))); - } - } - - private Long getServiceId(Map<String, Object> props) { - return (Long) props.get(Constants.SERVICE_ID); - } - - public void addInterest(EndpointEventListener epListener, Map<String, Object> props) { - Long id = getServiceId(props); - - if (LOG.isInfoEnabled()) { - LOG.info("Service {} has registered an interest in endpoint events", id); - } - - Interest interest = new Interest(id, epListener, props); - - interests.put(id, interest); - endpointsBySource.values().stream() - .flatMap(Set::stream) - .forEach(interest::endpointChanged); - } - - public void updateInterest(Map<String, Object> props) { - Long id = getServiceId(props); - - if (LOG.isInfoEnabled()) { - LOG.info("Service {} has changed its interest in endpoint events", id); - } - - interests.get(id).update(props); - } - - public void removeInterest(Map<String, Object> props) { - interests.remove(getServiceId(props)); + removeSource(source); } } 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 2327ee7b..74ddcdc8 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,6 +41,7 @@ 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; @@ -85,16 +86,16 @@ public class MdnsDiscovery { } @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC) - public void bindEndpointEventListener(EndpointEventListener epListener, Map<String, Object> props) { - interestManager.addInterest(epListener, props); + public void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { + interestManager.addListener(sref, listener); } - public void updatedEndpointEventListener(Map<String, Object> props) { - interestManager.updateInterest(props); + public void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { + interestManager.updateListener(sref, listener); } - public void unbindEndpointEventListener(Map<String, Object> props) { - interestManager.removeInterest(props); + public void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { + interestManager.removeListener(sref); } @Reference(policy = ReferencePolicy.DYNAMIC) diff --git a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpConnectionManager.java b/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpConnectionManager.java index ce928a26..d5623cb1 100644 --- a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpConnectionManager.java +++ b/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/TcpConnectionManager.java @@ -19,6 +19,7 @@ package org.apache.aries.rsa.discovery.tcp; import org.apache.aries.rsa.discovery.tcp.TcpMessage.*; +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; @@ -169,7 +170,7 @@ public class TcpConnectionManager implements EndpointEventListener { if (peerUuid != null) { // passed the handshake boolean removed = connectionsByUuid.remove(peerUuid, conn); // remove if this is the active connection if (removed && !closing) { - interestManager.removePeer(peerUuid); // no active connections with peer + interestManager.removeSource(peerUuid); // no active connections with peer } } // re-start the connect retry thread if necessary: @@ -217,7 +218,7 @@ public class TcpConnectionManager implements EndpointEventListener { } else if (message instanceof UpdateMessage) { UpdateMessage u = (UpdateMessage) message; EndpointDescription endpoint = new EndpointDescription(u.getProperties()); - interestManager.addEndpoint(endpoint); + interestManager.addEndpoint(endpoint.getFrameworkUUID(), endpoint); } else if (message instanceof RemoveMessage) { RemoveMessage r = (RemoveMessage) message; interestManager.removeEndpoint(conn.getPeerUuid(), r.getEndpointId()); 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 9efdd7f9..ff1ee771 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 @@ -19,6 +19,7 @@ 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; @@ -192,11 +193,11 @@ public class TcpDiscovery { @Reference(cardinality = MULTIPLE, policy = DYNAMIC, target = "(!(" + OWN_LISTENER_PROP + "=*))") void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.addListener(sref, listener, true); + interestManager.addListener(sref, listener); } void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - interestManager.addListener(sref, listener, false); + interestManager.updateListener(sref, listener); } void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { 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 0dba15e5..747aa034 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 @@ -18,10 +18,6 @@ */ package org.apache.aries.rsa.discovery.zookeeper; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.aries.rsa.spi.discovery.Interest; 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; @@ -31,7 +27,6 @@ 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.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,62 +37,34 @@ import org.slf4j.LoggerFactory; * Events from repository are then forwarded to all interested {@link EndpointEventListener}s. */ @Component(immediate = true) -public class InterestManager { +public class InterestManager extends org.apache.aries.rsa.spi.discovery.InterestManager { private static final Logger LOG = LoggerFactory.getLogger(InterestManager.class); - private final Map<ServiceReference<EndpointEventListener>, Interest> interests = new ConcurrentHashMap<>(); - private ZookeeperEndpointListener zkListener; - public InterestManager() { - } - // Using ARepository name to make sure it is injected first @Reference public void bindARepository(ZookeeperEndpointRepository repository) { - zkListener = repository.createListener(this::onEndpointEvent); + zkListener = repository.createListener(this); } @Deactivate public void deactivate() { zkListener.close(); - interests.clear(); - } - - private void onEndpointEvent(EndpointEvent event) { - interests.values().forEach(interest -> interest.notifyListener(event)); } - @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC) + @Reference(cardinality = ReferenceCardinality.MULTIPLE, policy = ReferencePolicy.DYNAMIC, + target = "(!(" + ClientManager.DISCOVERY_ZOOKEEPER_ID + "=*))") public void bindEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - addInterest(sref, listener); + addListener(sref, listener); } public void updatedEndpointEventListener(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - addInterest(sref, listener); + updateListener(sref, listener); } public void unbindEndpointEventListener(ServiceReference<EndpointEventListener> sref) { - interests.remove(sref); - } - - private void addInterest(ServiceReference<EndpointEventListener> sref, EndpointEventListener listener) { - if (isOurOwnEndpointEventListener(sref)) { - LOG.debug("Skipping our own EndpointEventListener"); - return; - } - Interest interest = new Interest(sref, listener); - boolean exists = interests.put(sref, interest) != null; - LOG.debug("{} Interest: {}", exists ? "Updating" : "Adding", interest); - if (zkListener != null) { - zkListener.getEndpoints().stream() - .map(endpoint -> new EndpointEvent(EndpointEvent.ADDED, endpoint)) - .forEach(interest::notifyListener); - } - } - - private static boolean isOurOwnEndpointEventListener(ServiceReference<EndpointEventListener> sref) { - return Boolean.parseBoolean(String.valueOf(sref.getProperty(ClientManager.DISCOVERY_ZOOKEEPER_ID))); + removeListener(sref); } int size() { diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointListener.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointListener.java index 81aedf0a..29532707 100644 --- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointListener.java +++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointListener.java @@ -20,13 +20,10 @@ package org.apache.aries.rsa.discovery.zookeeper.client; import java.io.ByteArrayInputStream; import java.io.Closeable; -import java.util.Collection; import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.function.Consumer; import org.apache.aries.rsa.spi.EndpointDescriptionParser; +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.ConnectionLossException; import org.apache.zookeeper.KeeperException.NoNodeException; @@ -35,39 +32,30 @@ import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.apache.aries.rsa.util.CollectionUtils.getChangedKeys; - /** * Listens to endpoint changes in Zookeeper and forwards changes in Endpoints to InterestManager. */ public class ZookeeperEndpointListener implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(ZookeeperEndpointListener.class); - private Map<String, EndpointDescription> endpoints = new ConcurrentHashMap<>(); private ZooKeeper zk; private EndpointDescriptionParser parser; - private Consumer<EndpointEvent> listener; + private InterestManager interestManager; private volatile boolean closed; - ZookeeperEndpointListener(ZooKeeper zk, EndpointDescriptionParser parser, Consumer<EndpointEvent> listener) { + ZookeeperEndpointListener(ZooKeeper zk, EndpointDescriptionParser parser, InterestManager interestManager) { this.zk = zk; this.parser = parser; - this.listener = listener; + this.interestManager = interestManager; watchRecursive(ZookeeperEndpointRepository.PATH_PREFIX); } - public Collection<EndpointDescription> getEndpoints() { - return endpoints.values(); - } - @Override public void close() { closed = true; - endpoints.clear(); } private void process(WatchedEvent event) { @@ -114,22 +102,11 @@ public class ZookeeperEndpointListener implements Closeable { } private void onChanged(String path, EndpointDescription endpoint) { - EndpointDescription old = endpoints.put(path, endpoint); - if (old != null && getChangedKeys(old.getProperties(), endpoint.getProperties()).isEmpty()) { - LOG.trace("ignoring endpoint that hasn't changed: {}", endpoint); - return; - } - int type = old == null ? EndpointEvent.ADDED : EndpointEvent.MODIFIED; - EndpointEvent event = new EndpointEvent(type, endpoint); - listener.accept(event); + interestManager.addEndpoint(path, endpoint); } private void onRemoved(String path) { - EndpointDescription endpoint = endpoints.remove(path); - if (endpoint != null) { - EndpointEvent event = new EndpointEvent(EndpointEvent.REMOVED, endpoint); - listener.accept(event); - } + interestManager.removeSource(path); } private EndpointDescription read(String path) throws KeeperException, InterruptedException { diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepository.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepository.java index 4c01d0ff..7349ece1 100644 --- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepository.java +++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepository.java @@ -25,9 +25,9 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.function.Consumer; import org.apache.aries.rsa.spi.EndpointDescriptionParser; +import org.apache.aries.rsa.spi.discovery.InterestManager; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.KeeperException.NoNodeException; @@ -79,8 +79,8 @@ public class ZookeeperEndpointRepository { } } - public ZookeeperEndpointListener createListener(Consumer<EndpointEvent> listener) { - return new ZookeeperEndpointListener(zk, parser, listener); + public ZookeeperEndpointListener createListener(InterestManager interestManager) { + return new ZookeeperEndpointListener(zk, parser, interestManager); } public void endpointChanged(EndpointEvent event) { 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 793bb35d..0a09d8b9 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 @@ -21,8 +21,6 @@ package org.apache.aries.rsa.discovery.zookeeper; import static org.junit.Assert.assertEquals; import static org.mockito.Mockito.when; -import java.util.Collections; - import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointListener; import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointRepository; import org.junit.Test; @@ -53,7 +51,6 @@ public class InterestManagerTest { @Test public void testEndpointListenerTrackerCustomizer() { when(repository.createListener(Mockito.any())).thenReturn(listener); - when(listener.getEndpoints()).thenReturn(Collections.emptyList()); im = new InterestManager(); im.bindARepository(repository); ServiceReference<EndpointEventListener> sref = createService("(objectClass=mine)"); diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepositoryPathTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepositoryPathTest.java index 900c2e22..b4bba6dd 100644 --- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepositoryPathTest.java +++ b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/client/ZookeeperEndpointRepositoryPathTest.java @@ -20,7 +20,6 @@ package org.apache.aries.rsa.discovery.zookeeper.client; import static org.junit.Assert.assertEquals; -import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointRepository; import org.junit.Test; public class ZookeeperEndpointRepositoryPathTest { diff --git a/spi/src/main/java/org/apache/aries/rsa/spi/discovery/Interest.java b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/Interest.java index 6900dab2..84fd2127 100644 --- a/spi/src/main/java/org/apache/aries/rsa/spi/discovery/Interest.java +++ b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/Interest.java @@ -27,6 +27,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; +import java.util.Optional; import static org.osgi.service.remoteserviceadmin.EndpointEventListener.ENDPOINT_LISTENER_SCOPE; @@ -49,13 +50,41 @@ public class Interest { this.listener = listener; } - public void notifyListener(EndpointEvent event) { - EndpointDescription endpoint = event.getEndpoint(); - scopes.stream().filter(endpoint::matches).findFirst().ifPresent(scope -> { // notify with first scope + /** + * Notify the listener about an endpoint change (added, removed, modified + * properties or end of match), according to the listener's interest scope + * and its previous endpoint data. + * + * @param prev the previous endpoint data (before this update), + * or null if this is a newly added endpoint + * @param endpoint the new endpoint data, or null if the previously + * known endpoint is being removed + */ + public void notifyListener(EndpointDescription prev, EndpointDescription endpoint) { + Optional<String> oldScope = prev == null ? Optional.empty() : scopes.stream().filter(prev::matches).findFirst(); + Optional<String> scope = endpoint == null ? Optional.empty() : scopes.stream().filter(endpoint::matches).findFirst(); + EndpointEvent event = null; + if (oldScope.isEmpty()) { // new endpoint + if (scope.isPresent()) { // new endpoint matched + event = new EndpointEvent(EndpointEvent.ADDED, endpoint); + } + } else if (scope.isPresent()) { // previously matched and currently matched endpoint + event = new EndpointEvent(EndpointEvent.MODIFIED, endpoint); + } else if (endpoint != null) { // previously matched and currently unmatched endpoint + event = new EndpointEvent(EndpointEvent.MODIFIED_ENDMATCH, endpoint); + scope = oldScope; + } else { // previously matched and now removed endpoint + event = new EndpointEvent(EndpointEvent.REMOVED, prev); + scope = oldScope; + } + + if (event != null) { LOG.info("Calling endpointChanged on {} for filter {}, type {}, endpoint {}", - listener, scope, event.getType(), endpoint); - listener.endpointChanged(event, scope); - }); + listener, scope, event.getType(), endpoint); + listener.endpointChanged(event, scope.get()); // notify with first matched scope + } else { // new unmatched endpoint - ignore + LOG.trace("interest {} ignoring unmatched endpoint {}", this, endpoint); + } } @Override diff --git a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/InterestManager.java b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java similarity index 53% rename from discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/InterestManager.java rename to spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java index 4929fcf9..af1b3538 100644 --- a/discovery/tcp/src/main/java/org/apache/aries/rsa/discovery/tcp/InterestManager.java +++ b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/InterestManager.java @@ -16,12 +16,10 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.aries.rsa.discovery.tcp; +package org.apache.aries.rsa.spi.discovery; -import org.apache.aries.rsa.spi.discovery.Interest; import org.osgi.framework.ServiceReference; import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -29,65 +27,75 @@ import org.slf4j.LoggerFactory; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import static org.osgi.service.remoteserviceadmin.EndpointEvent.*; +import static org.apache.aries.rsa.util.CollectionUtils.getChangedKeys; /** - * Manages bookkeeping of all known local {@link EndpointEventListener}s along with - * their interests (scopes), as well as all known remote endpoints, and notifies - * the former about the latter. + * Manages the bookkeeping of all known local {@link EndpointEventListener}s + * along with their interests (scopes) so that they can be notified of + * the endpoints they are interested in, and all known remote endpoints + * so that new listeners can be notified of all of them. */ public class InterestManager { private static final Logger LOG = LoggerFactory.getLogger(InterestManager.class); // listener to its interest - private final Map<ServiceReference<EndpointEventListener>, Interest> interests = new ConcurrentHashMap<>(); - // peer framework UUID to endpointId to endpoint - private final Map<String , Map<String, EndpointDescription>> remoteEndpoints = new ConcurrentHashMap<>(); + protected final Map<ServiceReference<EndpointEventListener>, Interest> interests = new ConcurrentHashMap<>(); + // source to endpointId to endpoint + protected final Map<String , Map<String, EndpointDescription>> remoteEndpoints = new ConcurrentHashMap<>(); public void addListener(ServiceReference<EndpointEventListener> sref, - EndpointEventListener listener, boolean isNew) { + EndpointEventListener listener) { // a new listener must be notified immediately of all previously known remote endpoints. // an existing listener is not notified, but we do need to update its scopes (interest) Interest interest = new Interest(sref, listener); - interests.put(sref, interest); // if it already exists - replace it with the new interest (scopes) - LOG.trace("Updated interest {} (new={})", isNew, interest); + boolean exists = interests.put(sref, interest) != null; + LOG.debug("{} interest {}, notifying listener of endpoints {}", + exists ? "Updated" : "Added", interest, remoteEndpoints.values()); // notify new or updated listener of all known remote endpoints // (according to the spec, the listener is idempotent so we don't worry about ADDED duplicates) - LOG.debug("Notifying new interest {} of all known remote endpoints {}", interest, remoteEndpoints.values()); remoteEndpoints.values().stream() .flatMap(endpoints -> endpoints.values().stream()) - .forEach(endpoint -> interest.notifyListener(new EndpointEvent(ADDED, endpoint))); + .forEach(endpoint -> interest.notifyListener(null, endpoint)); + } + + public void updateListener(ServiceReference<EndpointEventListener> sref, + EndpointEventListener listener) { + addListener(sref, listener); } public void removeListener(ServiceReference<EndpointEventListener> sref) { interests.remove(sref); } - private void notifyAllListeners(int type, EndpointDescription endpoint) { - EndpointEvent event = new EndpointEvent(type, endpoint); - interests.values().forEach(interest -> interest.notifyListener(event)); + private void notifyAllListeners(EndpointDescription old, EndpointDescription endpoint) { + // filter out duplicate endpoints (which may be received according to the spec) + if (old != null && endpoint != null && getChangedKeys(old.getProperties(), endpoint.getProperties()).isEmpty()) { + LOG.trace("ignoring unmodified endpoint: {}", endpoint); + return; + } + interests.values().forEach(interest -> interest.notifyListener(old, endpoint)); } - public void addEndpoint(EndpointDescription endpoint) { - Map<String, EndpointDescription> endpoints = remoteEndpoints - .computeIfAbsent(endpoint.getFrameworkUUID(), s -> new ConcurrentHashMap<>()); - boolean exists = endpoints.put(endpoint.getId(), endpoint) != null; - notifyAllListeners(exists ? MODIFIED : ADDED, endpoint); + public void addEndpoint(String source, EndpointDescription endpoint) { + Map<String, EndpointDescription> endpoints = + remoteEndpoints.computeIfAbsent(source, s -> new ConcurrentHashMap<>()); + EndpointDescription old = endpoints.put(endpoint.getId(), endpoint); + notifyAllListeners(old, endpoint); } - public void removeEndpoint(String peerUuid, String endpointId) { - Map<String, EndpointDescription> endpoints = remoteEndpoints.get(peerUuid); + public void removeEndpoint(String source, String endpointId) { + Map<String, EndpointDescription> endpoints = remoteEndpoints.get(source); if (endpoints != null) { EndpointDescription endpoint = endpoints.remove(endpointId); if (endpoint != null) { - notifyAllListeners(REMOVED, endpoint); + notifyAllListeners(endpoint, null); } } } - public void removePeer(String peerUuid) { - Map<String, EndpointDescription> endpoints = remoteEndpoints.remove(peerUuid); + public void removeSource(String source) { + Map<String, EndpointDescription> endpoints = remoteEndpoints.remove(source); if (endpoints != null) { - endpoints.values().forEach(endpoint -> notifyAllListeners(REMOVED, endpoint)); + endpoints.values().forEach(endpoint -> notifyAllListeners(endpoint, null)); } } }
