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 ceeabcbf905807bbaa64affc58a1f8286566a8dc Author: Amichai Rothman <[email protected]> AuthorDate: Thu Jun 4 17:11:13 2026 +0300 ARIES-2233 Extract common collection union utility method --- .../aries/rsa/discovery/mdns/InterestManager.java | 7 +--- .../discovery/mdns/PublishingEndpointListener.java | 5 +-- .../rsa/provider/fastbin/FastBinProvider.java | 17 ++------ .../apache/aries/rsa/provider/tcp/TcpProvider.java | 13 +----- .../apache/aries/rsa/core/EventListenerBridge.java | 5 ++- .../org/apache/aries/rsa/util/CollectionUtils.java | 48 ++++++++++++++++++++++ 6 files changed, 61 insertions(+), 34 deletions(-) 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 cc2143d7..549054e4 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 @@ -20,16 +20,15 @@ package org.apache.aries.rsa.discovery.mdns; import static java.util.Collections.singleton; import static java.util.stream.Collectors.toSet; -import static java.util.stream.Stream.concat; 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; -import java.util.stream.Stream; import javax.ws.rs.client.Client; import javax.ws.rs.sse.InboundSseEvent; @@ -110,9 +109,7 @@ public class InterestManager { if (ENDPOINT_UPDATED.equals(name)) { EndpointDescription ed = parser.readEndpoint(event.readData(InputStream.class)); - endpointsBySource.compute(source, (a,b) -> { - return b == null ? singleton(ed) : concat(b.stream(), Stream.of(ed)).collect(toSet()); - }); + endpointsBySource.compute(source, (a,b) -> b == null ? singleton(ed) : union(b, singleton(ed))); interests.values().forEach(i -> i.endpointChanged(ed)); } else if (ENDPOINT_REVOKED.equals(name)) { String id = event.readData(); 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 0f7049fe..bd46f37a 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 @@ -20,8 +20,8 @@ package org.apache.aries.rsa.discovery.mdns; import static java.util.Collections.singleton; import static java.util.stream.Collectors.toSet; -import static java.util.stream.Stream.concat; 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; @@ -30,7 +30,6 @@ import java.util.Hashtable; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; -import java.util.stream.Stream; import javax.ws.rs.GET; import javax.ws.rs.Path; @@ -105,7 +104,7 @@ public class PublishingEndpointListener { localEndpoints.compute(id, (k,v) -> new SponsoredEndpoint(ed, v == null ? singleton(bundleId) - : concat(v.sponsors.stream(), Stream.of(bundleId)).collect(toSet()))); + : union(v.sponsors, singleton(bundleId)))); String data = toEndpointData(ed); listeners.forEach(s -> s.update(data)); break; diff --git a/provider/fastbin/src/main/java/org/apache/aries/rsa/provider/fastbin/FastBinProvider.java b/provider/fastbin/src/main/java/org/apache/aries/rsa/provider/fastbin/FastBinProvider.java index 16388e60..e1ace8d6 100644 --- a/provider/fastbin/src/main/java/org/apache/aries/rsa/provider/fastbin/FastBinProvider.java +++ b/provider/fastbin/src/main/java/org/apache/aries/rsa/provider/fastbin/FastBinProvider.java @@ -23,8 +23,6 @@ import java.lang.reflect.InvocationHandler; import java.lang.reflect.Proxy; import java.net.URI; import java.util.Arrays; -import java.util.Collection; -import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -52,6 +50,8 @@ import org.osgi.service.remoteserviceadmin.RemoteConstants; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.apache.aries.rsa.util.CollectionUtils.union; + @SuppressWarnings("rawtypes") public class FastBinProvider implements DistributionProvider { @@ -120,15 +120,6 @@ public class FastBinProvider implements DistributionProvider { return new String[] {FASTBIN_CONFIG_TYPE}; } - @SafeVarargs - private static <T> Set<T> union(Collection<T>... collections) { - Set<T> union = new HashSet<>(); - for (Collection<T> c : collections) - if (c != null) - union.addAll(c); - return union; - } - @Override public Endpoint exportService(final Object serviceO, BundleContext serviceContext, @@ -137,8 +128,8 @@ public class FastBinProvider implements DistributionProvider { effectiveProperties.put(RemoteConstants.SERVICE_IMPORTED_CONFIGS, getSupportedTypes()); Set<String> intents = union( - StringPlus.normalize(effectiveProperties.get(RemoteConstants.SERVICE_EXPORTED_INTENTS)), - StringPlus.normalize(effectiveProperties.get(RemoteConstants.SERVICE_EXPORTED_INTENTS_EXTRA))); + StringPlus.normalize(effectiveProperties.get(RemoteConstants.SERVICE_EXPORTED_INTENTS)), + StringPlus.normalize(effectiveProperties.get(RemoteConstants.SERVICE_EXPORTED_INTENTS_EXTRA))); intents.removeAll(Arrays.asList(SUPPORTED_INTENTS)); if (!intents.isEmpty()) { LOG.warn("Unsupported intents found: {}. Not exporting service", intents); diff --git a/provider/tcp/src/main/java/org/apache/aries/rsa/provider/tcp/TcpProvider.java b/provider/tcp/src/main/java/org/apache/aries/rsa/provider/tcp/TcpProvider.java index 9f6e9fca..30d880ff 100644 --- a/provider/tcp/src/main/java/org/apache/aries/rsa/provider/tcp/TcpProvider.java +++ b/provider/tcp/src/main/java/org/apache/aries/rsa/provider/tcp/TcpProvider.java @@ -27,9 +27,7 @@ import java.security.NoSuchAlgorithmException; import java.security.UnrecoverableKeyException; import java.security.cert.CertificateException; import java.util.Arrays; -import java.util.Collection; import java.util.HashMap; -import java.util.HashSet; import java.util.Map; import java.util.Set; @@ -51,6 +49,8 @@ import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import javax.net.ssl.SSLContext; +import static org.apache.aries.rsa.util.CollectionUtils.union; + /** * The main TCP distribution provider class, which can import remote endpoints * (by creating a local proxy that sends method invocations to the remote @@ -83,15 +83,6 @@ public class TcpProvider implements DistributionProvider { return new String[] {TCP_CONFIG_TYPE}; } - @SafeVarargs - private static <T> Set<T> union(Collection<T>... collections) { - Set<T> union = new HashSet<>(); - for (Collection<T> c : collections) - if (c != null) - union.addAll(c); - return union; - } - private void initSocketFactories(Config config) { try { mtls = config.isMtls(); diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/EventListenerBridge.java b/rsa/src/main/java/org/apache/aries/rsa/core/EventListenerBridge.java index 719cfb14..cb4f37e9 100644 --- a/rsa/src/main/java/org/apache/aries/rsa/core/EventListenerBridge.java +++ b/rsa/src/main/java/org/apache/aries/rsa/core/EventListenerBridge.java @@ -28,7 +28,8 @@ import org.osgi.service.remoteserviceadmin.EndpointListener; import java.util.*; import java.util.concurrent.ConcurrentHashMap; -import java.util.stream.Stream; + +import static org.apache.aries.rsa.util.CollectionUtils.union; /** * This service acts as a bridge between those consumers/producers of endpoint @@ -247,7 +248,7 @@ public class EventListenerBridge implements ServiceListener, ListenerHook { // our listener's scope is the combined scope of all known consumers we may dispatch to Hashtable<String, Object> props = new Hashtable<>(); props.put(OWN_LISTENER_PROP, "true"); // mark our own listener for exclusion from consumers - String[] scopes = Stream.concat(oldConsumers.stream(), newConsumers.stream()) + String[] scopes = union(oldConsumers, newConsumers).stream() .flatMap(s -> StringPlus.normalize(s.getProperty(EndpointEventListener.ENDPOINT_LISTENER_SCOPE)).stream()) .toArray(String[]::new); props.put(EndpointEventListener.ENDPOINT_LISTENER_SCOPE, scopes); diff --git a/spi/src/main/java/org/apache/aries/rsa/util/CollectionUtils.java b/spi/src/main/java/org/apache/aries/rsa/util/CollectionUtils.java new file mode 100644 index 00000000..91e43cb6 --- /dev/null +++ b/spi/src/main/java/org/apache/aries/rsa/util/CollectionUtils.java @@ -0,0 +1,48 @@ +/* + * 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.util; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + +/** + * Collection-related utilities that can be useful to SPI implementations. + */ +public class CollectionUtils { + + // private constructor to prevent instantiation + private CollectionUtils() {} + + /** + * Returns the union of all elements in all given collections. + * + * @param <T> the type of items in the collections + * @param collections the collections to combine + * @return the union of all elements in all the collections + */ + @SafeVarargs + public static <T> Set<T> union(Collection<? extends T>... collections) { + Set<T> union = new HashSet<>(); + for (Collection<? extends T> c : collections) + if (c != null) + union.addAll(c); + return union; + } +}
