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 d2afc5a70944fd7ce4e27fac07f00f742c5f41e3 Author: Amichai Rothman <[email protected]> AuthorDate: Sat Jun 6 14:37:15 2026 +0300 ARIES-2233 Extract common LocalEndpointManager to track local endpoints --- .../aries/rsa/discovery/mdns/MdnsDiscovery.java | 2 +- .../discovery/mdns/PublishingEndpointListener.java | 112 ++----------- .../rsa/discovery/tcp/TcpConnectionManager.java | 14 +- .../aries/rsa/discovery/tcp/TcpDiscovery.java | 26 +-- .../zookeeper/PublishingEndpointListener.java | 37 +---- .../zookeeper/PublishingEndpointListenerTest.java | 65 -------- .../rsa/spi/discovery/LocalEndpointManager.java | 181 +++++++++++++++++++++ 7 files changed, 219 insertions(+), 218 deletions(-) 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 bee88eb2..b97141d0 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 @@ -80,7 +80,7 @@ public class MdnsDiscovery { interestManager = new InterestManager(eventSourceFactory, parser, client); interestManager.start(ctx, null); fwUuid = ctx.getProperty(FRAMEWORK_UUID); - publishingListener = new PublishingEndpointListener(parser, ctx, fwUuid); + publishingListener = new PublishingEndpointListener(parser, ctx); } @Reference(policy = ReferencePolicy.DYNAMIC) diff --git a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/PublishingEndpointListener.java b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/PublishingEndpointListener.java index bd46f37a..08fae8c7 100644 --- a/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/PublishingEndpointListener.java +++ b/discovery/mdns/src/main/java/org/apache/aries/rsa/discovery/mdns/PublishingEndpointListener.java @@ -18,18 +18,13 @@ */ package org.apache.aries.rsa.discovery.mdns; -import static java.util.Collections.singleton; -import static java.util.stream.Collectors.toSet; import static javax.ws.rs.core.MediaType.SERVER_SENT_EVENTS; -import static org.apache.aries.rsa.util.CollectionUtils.union; import java.io.ByteArrayOutputStream; import java.nio.charset.StandardCharsets; -import java.util.Dictionary; import java.util.Hashtable; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import javax.ws.rs.GET; import javax.ws.rs.Path; @@ -39,17 +34,14 @@ import javax.ws.rs.sse.Sse; import javax.ws.rs.sse.SseEventSink; import org.apache.aries.rsa.spi.EndpointDescriptionParser; -import org.osgi.framework.Bundle; +import org.apache.aries.rsa.spi.discovery.LocalEndpointManager; import org.osgi.framework.BundleContext; -import org.osgi.framework.Constants; -import org.osgi.framework.ServiceFactory; import org.osgi.framework.ServiceRegistration; import org.osgi.service.component.annotations.Deactivate; import org.osgi.service.jaxrs.whiteboard.annotations.RequireJaxrsWhiteboard; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; -import org.osgi.service.remoteserviceadmin.RemoteConstants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -62,80 +54,48 @@ public class PublishingEndpointListener { private static final Logger LOG = LoggerFactory.getLogger(PublishingEndpointListener.class); - private final String uuid; - private final EndpointDescriptionParser parser; - private final ServiceRegistration<?> listenerReg; - private final ServiceRegistration<?> resourceReg; + private final LocalEndpointManager localEndpointManager; - private final ConcurrentMap<String, SponsoredEndpoint> localEndpoints = new ConcurrentHashMap<>(); + private final ServiceRegistration<?> resourceReg; private final Set<Subscription> listeners = ConcurrentHashMap.newKeySet(); @SuppressWarnings("serial") - public PublishingEndpointListener(EndpointDescriptionParser parser, BundleContext bctx, String uuid) { + public PublishingEndpointListener(EndpointDescriptionParser parser, BundleContext context) { this.parser = parser; - this.uuid = uuid; - String[] ifAr = { EndpointEventListener.class.getName() }; - Dictionary<String, Object> props = serviceProperties(uuid); - listenerReg = bctx.registerService(ifAr, new ListenerFactory(), props); - resourceReg = bctx.registerService(PublishingEndpointListener.class, this, - new Hashtable<>() {{put("osgi.jaxrs.resource", Boolean.TRUE);}}); + this.localEndpointManager = new LocalEndpointManager(); + localEndpointManager.setListener((event, filter) -> endpointUpdate(event.getEndpoint(), event.getType())); + localEndpointManager.start(context, getClass().getName()); + resourceReg = context.registerService(PublishingEndpointListener.class, this, + new Hashtable<>() {{put("osgi.jaxrs.resource", Boolean.TRUE);}}); } @Deactivate public void stop() { - listenerReg.unregister(); + localEndpointManager.stop(); listeners.forEach(Subscription::close); resourceReg.unregister(); } - private void endpointUpdate(Long bundleId, EndpointDescription ed, int type) { - String edFwUuid = ed.getFrameworkUUID(); - if (edFwUuid == null || !edFwUuid.equals(uuid)) { - LOG.warn("This listener has been called with an endpoint {} for a remote framework {}", ed.getId(), edFwUuid); - return; - } + private void endpointUpdate(EndpointDescription ed, int type) { String id = ed.getId(); switch(type) { case EndpointEvent.ADDED: case EndpointEvent.MODIFIED: - localEndpoints.compute(id, (k,v) -> new SponsoredEndpoint(ed, - v == null - ? singleton(bundleId) - : union(v.sponsors, singleton(bundleId)))); String data = toEndpointData(ed); listeners.forEach(s -> s.update(data)); break; case EndpointEvent.MODIFIED_ENDMATCH: case EndpointEvent.REMOVED: - boolean act = localEndpoints.compute(id, (k,v) -> { - if (v == null) { - return null; - } else { - Set<Long> updated = v.sponsors.stream().filter(l -> !bundleId.equals(l)).collect(toSet()); - return updated.isEmpty() ? null : new SponsoredEndpoint(v.ed, updated); - } - }) == null; - - if (act) { - listeners.forEach(s -> s.revoke(id)); - } + listeners.forEach(s -> s.revoke(id)); break; default: LOG.error("Unknown event type {} for endpoint {}", type, ed); } } - private Dictionary<String, Object> serviceProperties(String uuid) { - String scope = String.format("(&(%s=*)(%s=%s))", Constants.OBJECTCLASS, - RemoteConstants.ENDPOINT_FRAMEWORK_UUID, uuid); - Dictionary<String, Object> props = new Hashtable<>(); - props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, scope); - return props; - } - private String toEndpointData(EndpointDescription ed) { try { ByteArrayOutputStream baos = new ByteArrayOutputStream(); @@ -154,45 +114,11 @@ public class PublishingEndpointListener { Subscription subscription = new Subscription(sse, sink); listeners.add(subscription); - localEndpoints.values().stream() - .map(s -> toEndpointData(s.ed)) + localEndpointManager.getEndpoints().stream() + .map(this::toEndpointData) .forEach(subscription::update); } - private class ListenerFactory implements ServiceFactory<PerClientEndpointEventListener> { - - @Override - public PerClientEndpointEventListener getService(Bundle bundle, - ServiceRegistration<PerClientEndpointEventListener> registration) { - return new PerClientEndpointEventListener(bundle.getBundleId()); - } - - @Override - public void ungetService(Bundle bundle, ServiceRegistration<PerClientEndpointEventListener> registration, - PerClientEndpointEventListener service) { - Long bundleId = service.bundleId; - localEndpoints.values().stream() - .filter(s -> s.sponsors.contains(bundleId)) - .forEach(s -> endpointUpdate(bundleId, s.ed, EndpointEvent.REMOVED)); - } - - } - - private class PerClientEndpointEventListener implements EndpointEventListener { - - private final Long bundleId; - - public PerClientEndpointEventListener(Long bundleId) { - super(); - this.bundleId = bundleId; - } - - @Override - public void endpointChanged(EndpointEvent event, String filter) { - endpointUpdate(bundleId, event.getEndpoint(), event.getType()); - } - } - class Subscription { static final String ENDPOINT_UPDATED = "UPDATED"; @@ -229,14 +155,4 @@ public class PublishingEndpointListener { } } } - - private static class SponsoredEndpoint { - private final EndpointDescription ed; - private final Set<Long> sponsors; - - public SponsoredEndpoint(EndpointDescription ed, Set<Long> sponsors) { - this.ed = ed; - this.sponsors = sponsors; - } - } } 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 d5623cb1..4962a16c 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 @@ -20,6 +20,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.apache.aries.rsa.spi.discovery.LocalEndpointManager; import org.osgi.service.remoteserviceadmin.EndpointDescription; import org.osgi.service.remoteserviceadmin.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; @@ -51,6 +52,7 @@ public class TcpConnectionManager implements EndpointEventListener { private static final Logger LOG = LoggerFactory.getLogger(TcpConnectionManager.class); private final InterestManager interestManager; + private final LocalEndpointManager localEndpointManager; private final String localAddress; private final String localUuid; private final long reconnectDelay; @@ -59,15 +61,16 @@ public class TcpConnectionManager implements EndpointEventListener { private final ExecutorService executor = Executors.newCachedThreadPool(); // for connect/accept threads private final Set<TcpConnection> connections = ConcurrentHashMap.newKeySet(); // all connections, including before handshake private final Map<String, TcpConnection> connectionsByUuid = new ConcurrentHashMap<>(); // connections after handshake (known uuid) - private final Map<String, EndpointDescription> localEndpoints = new ConcurrentHashMap<>(); private final Set<String> peers = ConcurrentHashMap.newKeySet(); // all configured and discovered (gossip) peer addresses private ServerSocket serverSocket; private volatile boolean closing; - public TcpConnectionManager(InterestManager interestManager, String localAddress, - String localUuid, long reconnectDelay, boolean gossip) { + public TcpConnectionManager(InterestManager interestManager, LocalEndpointManager localEndpointManager, + String localAddress, String localUuid, long reconnectDelay, boolean gossip) { this.interestManager = interestManager; + this.localEndpointManager = localEndpointManager; + localEndpointManager.setListener(this); this.localAddress = localAddress; this.localUuid = localUuid; this.reconnectDelay = reconnectDelay; @@ -214,7 +217,8 @@ public class TcpConnectionManager implements EndpointEventListener { return; } // send all of our known local endpoints to the new peer - localEndpoints.values().forEach(endpoint -> conn.send(new UpdateMessage(endpoint.getProperties()))); + localEndpointManager.getEndpoints().forEach( + endpoint -> conn.send(new UpdateMessage(endpoint.getProperties()))); } else if (message instanceof UpdateMessage) { UpdateMessage u = (UpdateMessage) message; EndpointDescription endpoint = new EndpointDescription(u.getProperties()); @@ -236,12 +240,10 @@ public class TcpConnectionManager implements EndpointEventListener { switch (event.getType()) { case EndpointEvent.ADDED: case EndpointEvent.MODIFIED: - localEndpoints.put(endpointId, endpoint); message = new UpdateMessage(endpoint.getProperties()); break; case EndpointEvent.MODIFIED_ENDMATCH: case EndpointEvent.REMOVED: - localEndpoints.remove(endpointId); message = new RemoveMessage(endpointId); break; default: throw new RuntimeException("Unknown event type: " + event.getType()); 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 e717bec5..49606523 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 @@ -20,8 +20,8 @@ package org.apache.aries.rsa.discovery.tcp; import org.apache.aries.rsa.annotations.RSADiscoveryProvider; import org.apache.aries.rsa.spi.discovery.InterestManager; +import org.apache.aries.rsa.spi.discovery.LocalEndpointManager; import org.osgi.framework.BundleContext; -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; @@ -36,14 +36,10 @@ import java.lang.reflect.Proxy; import java.net.URI; import java.net.URISyntaxException; import java.util.Arrays; -import java.util.Dictionary; -import java.util.Hashtable; import java.util.Objects; import java.util.stream.Collectors; import static org.osgi.framework.Constants.FRAMEWORK_UUID; -import static org.osgi.service.remoteserviceadmin.EndpointEventListener.ENDPOINT_LISTENER_SCOPE; -import static org.osgi.service.remoteserviceadmin.RemoteConstants.ENDPOINT_FRAMEWORK_UUID; /** * The main TCP Discovery provider component. @@ -72,11 +68,12 @@ public class TcpDiscovery { private InterestManager interestManager; private TcpConnectionManager connectionManager; - private ServiceRegistration<?> listenerRegistration; + private LocalEndpointManager localEndpointManager; public TcpDiscovery() { // initialize in constructor before we start getting reference bind events interestManager = new InterestManager(); + localEndpointManager = new LocalEndpointManager(); } public static URI toURI(String address) { @@ -151,18 +148,10 @@ public class TcpDiscovery { URI uri = toURI(address); bindAddress = bindAddress == null ? uri.getHost() : toURI(bindAddress).getHost(); // just the host connectionManager = new TcpConnectionManager( - interestManager, address, uuid, config.reconnectDelay(), config.gossip()); + interestManager, localEndpointManager, address, uuid, config.reconnectDelay(), config.gossip()); connectionManager.open(bindAddress, uri.getPort(), Arrays.asList(peers)); } - private void registerListener(BundleContext context, String uuid) { - Dictionary<String, Object> props = new Hashtable<>(); - props.put(OWN_LISTENER_PROP, Boolean.TRUE); // mark our own listener for exclusion - String scope = "(&(objectClass=*)(" + ENDPOINT_FRAMEWORK_UUID + "=" + uuid + "))"; - props.put(ENDPOINT_LISTENER_SCOPE, scope); - listenerRegistration = context.registerService(EndpointEventListener.class, connectionManager, props); - } - @Activate void start(BundleContext context, Config config) { String uuid = context.getProperty(FRAMEWORK_UUID); @@ -170,18 +159,17 @@ public class TcpDiscovery { LOG.info("Starting TCP discovery for framework {} with config {}", uuid, config); try { initConnectionManager(config, uuid); - // register ourselves to capture local endpoint exports - registerListener(context, uuid); } catch (IOException ioe) { LOG.error("failed to start TCP connection manager", ioe); } interestManager.start(context, OWN_LISTENER_PROP); + localEndpointManager.start(context, OWN_LISTENER_PROP); } @Deactivate void stop() throws IOException { - if (listenerRegistration != null) { - listenerRegistration.unregister(); + if (localEndpointManager != null) { + localEndpointManager.stop(); } if (connectionManager != null) { connectionManager.close(); diff --git a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java index db6d35a0..239a9ea5 100644 --- a/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java +++ b/discovery/zookeeper/src/main/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListener.java @@ -18,21 +18,16 @@ */ package org.apache.aries.rsa.discovery.zookeeper; -import java.util.Dictionary; -import java.util.Hashtable; - import org.apache.aries.rsa.discovery.zookeeper.client.ClientManager; import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointRepository; +import org.apache.aries.rsa.spi.discovery.LocalEndpointManager; import org.osgi.framework.BundleContext; -import org.osgi.framework.Constants; -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.EndpointEvent; import org.osgi.service.remoteserviceadmin.EndpointEventListener; -import org.osgi.service.remoteserviceadmin.RemoteConstants; /** * Listens for local {@link EndpointEvent}s using {@link EndpointEventListener} @@ -40,38 +35,22 @@ import org.osgi.service.remoteserviceadmin.RemoteConstants; */ @SuppressWarnings("deprecation") @Component(service = {}, immediate = true) -public class PublishingEndpointListener implements EndpointEventListener { +public class PublishingEndpointListener { - private ServiceRegistration<?> listenerReg; + protected LocalEndpointManager localEndpointManager; @Reference private ZookeeperEndpointRepository repository; @Activate - public void start(BundleContext bctx) { - String uuid = bctx.getProperty(Constants.FRAMEWORK_UUID); - String[] ifAr = { EndpointEventListener.class.getName() }; - Dictionary<String, String> props = serviceProperties(uuid); - listenerReg = bctx.registerService(ifAr, this, props); + public void start(BundleContext context) { + localEndpointManager = new LocalEndpointManager(); + localEndpointManager.setListener((event, filter) -> repository.endpointChanged(event)); + localEndpointManager.start(context, ClientManager.DISCOVERY_ZOOKEEPER_ID); } @Deactivate public void stop() { - listenerReg.unregister(); - } - - @Override - public void endpointChanged(EndpointEvent event, String filter) { - repository.endpointChanged(event); + localEndpointManager.stop(); } - - private Dictionary<String, String> serviceProperties(String uuid) { - String scope = String.format("(&(%s=*)(%s=%s))", Constants.OBJECTCLASS, - RemoteConstants.ENDPOINT_FRAMEWORK_UUID, uuid); - Dictionary<String, String> props = new Hashtable<>(); - props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, scope); - props.put(ClientManager.DISCOVERY_ZOOKEEPER_ID, "true"); - return props; - } - } diff --git a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java b/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java deleted file mode 100644 index 89987b22..00000000 --- a/discovery/zookeeper/src/test/java/org/apache/aries/rsa/discovery/zookeeper/PublishingEndpointListenerTest.java +++ /dev/null @@ -1,65 +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.zookeeper; - -import static org.mockito.Mockito.verify; - -import java.util.HashMap; -import java.util.Map; - -import org.apache.aries.rsa.discovery.zookeeper.client.ZookeeperEndpointRepository; -import org.apache.zookeeper.KeeperException; -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.InjectMocks; -import org.mockito.Mock; -import org.mockito.junit.MockitoJUnitRunner; -import org.osgi.framework.Constants; -import org.osgi.service.remoteserviceadmin.EndpointDescription; -import org.osgi.service.remoteserviceadmin.EndpointEvent; -import org.osgi.service.remoteserviceadmin.RemoteConstants; - -@RunWith(MockitoJUnitRunner.class) -public class PublishingEndpointListenerTest { - @Mock - ZookeeperEndpointRepository repository; - - @InjectMocks - PublishingEndpointListener eli; - - @Test - public void testEndpointRemovalAdding() throws KeeperException, InterruptedException { - EndpointDescription endpoint = createEndpoint(); - EndpointEvent event1 = new EndpointEvent(EndpointEvent.ADDED, endpoint); - eli.endpointChanged(event1, null); - EndpointEvent event2 = new EndpointEvent(EndpointEvent.REMOVED, endpoint); - eli.endpointChanged(event2, null); - - verify(repository).endpointChanged(event1); - verify(repository).endpointChanged(event2); - } - - private EndpointDescription createEndpoint() { - Map<String, Object> props = new HashMap<>(); - props.put(Constants.OBJECTCLASS, new String[] {"myClass"}); - props.put(RemoteConstants.ENDPOINT_ID, "http://google.de:80/test/sub";); - props.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, "myConfig"); - return new EndpointDescription(props); - } -} diff --git a/spi/src/main/java/org/apache/aries/rsa/spi/discovery/LocalEndpointManager.java b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/LocalEndpointManager.java new file mode 100644 index 00000000..b9f4d6ae --- /dev/null +++ b/spi/src/main/java/org/apache/aries/rsa/spi/discovery/LocalEndpointManager.java @@ -0,0 +1,181 @@ +/* + * 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.spi.discovery; + +import org.osgi.framework.*; +import org.osgi.service.remoteserviceadmin.EndpointDescription; +import org.osgi.service.remoteserviceadmin.EndpointEvent; +import org.osgi.service.remoteserviceadmin.EndpointEventListener; +import org.osgi.service.remoteserviceadmin.RemoteConstants; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +import static java.util.Collections.singleton; +import static org.apache.aries.rsa.util.CollectionUtils.union; + +/** + * Keeps track of all local endpoints so that updates about them can be sent to + * remote systems. + * <p> + * This includes registering an EndpointEventListener with a scope that excludes all + * non-local endpoints, a special service property that uniquely identifies this + * instance so that the interest manager can exclude it from its tracking, and a + * ServiceFactory that keeps track of which bundles notified us about which endpoints + * so that when a bundle is stopped all of its endpoints will be removed, as the spec + * requires. + * <p> + * After being processed here, endpoint events are forwarded to a supplied listener + * for implementation-specific handling (e.g. transmitting updates to remote hosts). + */ +public class LocalEndpointManager { + + private static final Logger LOG = LoggerFactory.getLogger(LocalEndpointManager.class); + + protected static class EndpointBundles { + EndpointDescription endpoint; + Set<Long> bundleIds; + + EndpointBundles(EndpointDescription endpoint, Set<Long> bundleIds) { + this.endpoint = endpoint; + this.bundleIds = bundleIds; + } + } + + protected ServiceRegistration<EndpointEventListener> listenerRegistration; + // local endpoint ids and their respective endpoints and the bundles that notified us about them + protected Map<String, EndpointBundles> localEndpoints = new ConcurrentHashMap<>(); + // the listener to which endpoint events are delegated after being processed + protected EndpointEventListener listener; + + public Collection<EndpointDescription> getEndpoints() { + return localEndpoints.values().stream().map(eb -> eb.endpoint).collect(Collectors.toList()); + } + + public void setListener(EndpointEventListener listener) { + this.listener = listener; + } + + public void start(BundleContext context, String excludeProperty) { + listenerRegistration = registerListener(context, excludeProperty); + } + + public void stop() { + if (listenerRegistration != null) { + listenerRegistration.unregister(); + } + } + + /** + * Registers an EndpointEventListener so we get notified of all local endpoints, + * with a few special features: + * <ul> + * <li>It is registered with a special given property that uniquely identifies + * this instance in order for the interest manager to exclude it in its scope + * when listening for interests</li> + * <li>It is registered with a scope that excludes all non-local endpoints + * by requiring the framework ID to be equal to the given context's framework ID</li> + * <li>It is registered as a {@link ServiceFactory} in order to keep track of + * which bundle(s) notifies us of each endpoint, so that when the bundle is + * stopped we can remove those endpoints (as required by the spec)</li> + * </ul> + * + * @param context the registering bundle context + * @param excludeProperty the name of a service property that uniquely identifies + * this instance + * @return the listener service registration + */ + protected ServiceRegistration<EndpointEventListener> registerListener( + BundleContext context, String excludeProperty) { + Dictionary<String, Object> props = new Hashtable<>(); + props.put(excludeProperty, Boolean.TRUE); // mark our own listener for exclusion + String uuid = context.getProperty(Constants.FRAMEWORK_UUID); + String scope = String.format("(&(%s=*)(%s=%s))", Constants.OBJECTCLASS, + RemoteConstants.ENDPOINT_FRAMEWORK_UUID, uuid); + props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, scope); + LOG.debug("registering EndpointEventListener factory with properties {}", props); + return context.registerService(EndpointEventListener.class, new ListenerServiceFactory(), props); + } + + /** + * A {@link ServiceFactory} registered as an {@link EndpointEventListener} so that we can + * keep track of which bundle(s) notified us about every endpoint. + */ + protected class ListenerServiceFactory implements ServiceFactory<EndpointEventListener> { + + @Override + public EndpointEventListener getService(Bundle bundle, ServiceRegistration<EndpointEventListener> registration) { + long bundleId = bundle.getBundleId(); + LOG.debug("service factory getting EndpointEventListener for bundle {} ({})", + bundleId, bundle.getSymbolicName()); + return ((event, filter) -> endpointChanged(event, filter, bundleId)); + } + + @Override + public void ungetService(Bundle bundle, ServiceRegistration<EndpointEventListener> registration, EndpointEventListener service) { + long bundleId = bundle.getBundleId(); + LOG.debug("service factory ungetting EndpointEventListener for bundle {} ({})", + bundleId, bundle.getSymbolicName()); + localEndpoints.values().stream() + .filter(eb -> eb.bundleIds.contains(bundleId)) + .forEach(eb -> endpointChanged(new EndpointEvent(EndpointEvent.REMOVED, eb.endpoint), "", bundleId)); + } + } + + /** + * Notification that an endpoint has changed. + * <p> + * This method is similar to {@link EndpointEventListener#endpointChanged}, + * with the added bundle ID of the bundle invoking the listener (which + * we provide by using a ServiceFactory). + * + * @param event The event containing the details about the change. + * @param filter The filter from the {@link EndpointEventListener#ENDPOINT_LISTENER_SCOPE} + * that matches (or for {@link EndpointEvent#MODIFIED_ENDMATCH} and + * {@link EndpointEvent#REMOVED} used to match) the endpoint, or an empty + * string when removing the endpoint because the notifying bundle was stopped + */ + public void endpointChanged(EndpointEvent event, String filter, Long bundleId) { + EndpointDescription endpoint = event.getEndpoint(); + LOG.debug("got endpointChanged event of type {} from bundle {}: {}", + event.getType(), bundleId, endpoint); + switch (event.getType()) { + case EndpointEvent.ADDED: + case EndpointEvent.MODIFIED: + localEndpoints.compute(endpoint.getId(), + (e, eb) -> new EndpointBundles(endpoint, + eb == null ? singleton(bundleId) : union(eb.bundleIds, singleton(bundleId)))); + break; + case EndpointEvent.REMOVED: + case EndpointEvent.MODIFIED_ENDMATCH: + localEndpoints.computeIfPresent(endpoint.getId(), + (e, eb) -> { + Set<Long> bundleIds = eb.bundleIds.stream() + .filter(id -> !bundleId.equals(id)) + .collect(Collectors.toSet()); + return bundleIds.isEmpty() ? null : new EndpointBundles(endpoint, bundleIds); + }); + break; + } + listener.endpointChanged(event, filter); + } +}
