You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@aries.apache.org by am...@apache.org on 2026/05/08 13:28:52 UTC
(aries-rsa) branch master updated (96489a39 -> d13e90db)
This is an automated email from the ASF dual-hosted git repository.
amichair pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/aries-rsa.git
from 96489a39 ARIES-2199: Use only-script version
new 37191ba9 ARIES-2212 Refactor ImportRegistrationImpl initialization to use one-time init instead of setters
new daa69700 ARIES-2207 Re-implement ImportRegistrationImpl more simple, clear and compliant
new 4bacc480 ARIES-2207 Re-implement ExportRegistrationImpl more simple, clear and compliant
new d874b29c ARIES-2208 Fix incorrect closing of RSA factory/core/instance/registrations
new d13e90db ARIES-2209 Merge ExportReferenceImpl into ExportRegistrationImpl
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../rsa/core/DistributionProviderTracker.java | 42 ++-
.../apache/aries/rsa/core/ExportReferenceImpl.java | 77 -----
.../aries/rsa/core/ExportRegistrationImpl.java | 260 ++++++++++-------
.../aries/rsa/core/ImportRegistrationImpl.java | 324 +++++++++++----------
.../aries/rsa/core/RemoteServiceAdminCore.java | 155 +++-------
.../aries/rsa/core/RemoteServiceAdminFactory.java | 27 +-
.../aries/rsa/core/RemoteServiceAdminInstance.java | 74 +++--
.../rsa/core/DistributionProviderTrackerTest.java | 8 +-
.../aries/rsa/core/ImportRegistrationImplTest.java | 30 +-
.../exporter/ServiceExportsRepository.java | 14 +-
10 files changed, 486 insertions(+), 525 deletions(-)
delete mode 100644 rsa/src/main/java/org/apache/aries/rsa/core/ExportReferenceImpl.java
(aries-rsa) 03/05: ARIES-2207 Re-implement ExportRegistrationImpl more simple, clear and compliant
Posted by am...@apache.org.
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 4bacc480251e5fc3f2234b0d489f4bcc0bf0f9b2
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 23 23:32:34 2026 +0200
ARIES-2207 Re-implement ExportRegistrationImpl more simple, clear and compliant
---
.../aries/rsa/core/ExportRegistrationImpl.java | 230 ++++++++++++---------
.../exporter/ServiceExportsRepository.java | 14 +-
2 files changed, 138 insertions(+), 106 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
index 08149d3e..a2fd7129 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
@@ -24,6 +24,7 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.aries.rsa.core.event.EventProducer;
import org.apache.aries.rsa.spi.Endpoint;
@@ -40,121 +41,149 @@ public class ExportRegistrationImpl implements ExportRegistration {
private static final Logger LOG = LoggerFactory.getLogger(ExportRegistrationImpl.class);
- private final CloseHandler closeHandler;
- private ExportReferenceImpl exportReference;
- private final Closeable server;
- private final Throwable exception;
+ /**
+ * According to the specs, multiple ExportRegistration can be linked together,
+ * i.e. they have a shared state - this is implemented via this inner class,
+ * where every group of linked export registrations reference a single Shared instance.
+ */
+ private static class Shared {
- private final ExportRegistrationImpl parent;
- private int instanceCount;
- private volatile boolean closed;
+ private CloseHandler closeHandler;
+ private EventProducer eventProducer;
+ private Closeable server;
+ private Throwable exception;
- private final EventProducer sender;
-
- private ExportRegistrationImpl(ExportRegistrationImpl parent,
- CloseHandler rsaCore,
- EventProducer sender,
- ExportReferenceImpl exportReference,
- Closeable server,
- Throwable exception) {
- this.sender = sender;
- this.parent = parent != null ? parent.parent : this; // a parent points to itself
- this.parent.addInstance();
- this.closeHandler = rsaCore;
- this.exportReference = exportReference;
- this.server = server;
- this.exception = exception;
- }
+ private int instanceCount;
+
+ void addInstance() {
+ synchronized (this) {
+ instanceCount++;
+ }
+ }
- // create a clone of the provided ExportRegistrationImpl that is linked to it
- public ExportRegistrationImpl(ExportRegistrationImpl parent) {
- this(parent, parent.closeHandler, parent.sender, new ExportReferenceImpl(parent.exportReference),
- parent.server, parent.exception);
+ void removeInstance() {
+ synchronized (this) {
+ if (--instanceCount != 0) {
+ return;
+ }
+ LOG.debug("closing ExportRegistration after removing last linked instance");
+ if (server != null) {
+ try {
+ server.close();
+ } catch (IOException ioe) {
+ LOG.warn("Error closing ExportRegistration", ioe);
+ }
+ }
+ }
+ }
}
- // create a new (parent) instance which was exported successfully with the given server
- public ExportRegistrationImpl(ServiceReference sref, Endpoint endpoint, CloseHandler closeHandler, EventProducer sender) {
- this(null, closeHandler, sender, new ExportReferenceImpl(sref, endpoint.description()), endpoint, null);
+ // state shared between all linked export registrations in this instance's group
+ private final Shared shared;
+
+ // per-instance state that is not shared
+ private volatile ExportReferenceImpl exportReference;
+ private AtomicBoolean closing = new AtomicBoolean();
+ private volatile boolean closed;
+
+
+ /**
+ * Constructs an export registration that is linked
+ * (shares state) with the given export registration.
+ * <p>
+ * The {@link #close} method must eventually be invoked on this instance.
+ *
+ * @param er the export registration that this instance is linked to
+ */
+ public ExportRegistrationImpl(ExportRegistrationImpl er) {
+ exportReference = new ExportReferenceImpl(er.exportReference);
+ shared = er.shared;
+ shared.addInstance();
}
- // create a new (parent) instance which failed to be exported with the given exception
- public ExportRegistrationImpl(Throwable exception, CloseHandler closeHandler, EventProducer sender) {
- this(null, closeHandler, sender, null, null, exception);
+ /**
+ * Constructs a new export registration that is not linked to any other instance
+ * in a successful state.
+ * <p>
+ * The {@link #close} method must eventually be invoked on this instance.
+ *
+ * @param sref the exported service reference
+ * @param endpoint the exported endpoint info
+ * @param closeHandler a callback function that will be invoked when the registration is closed
+ * @param eventProducer an event producer that will be invoked when the endpoint is updated
+ */
+ // create a new (parent) instance which was exported successfully with the given server
+ public ExportRegistrationImpl(ServiceReference sref, Endpoint endpoint,
+ CloseHandler closeHandler, EventProducer eventProducer) {
+ exportReference = new ExportReferenceImpl(sref, endpoint.description());
+ shared = new Shared();
+ shared.closeHandler = closeHandler;
+ shared.eventProducer = eventProducer;
+ shared.server = endpoint;
+ shared.addInstance();
}
- private void ensureParent() {
- if (parent != this) {
- throw new IllegalStateException("this method may only be called on the parent");
- }
+ /**
+ * Constructs a new export registration that is not linked to any other instance
+ * in a failed state.
+ * <p>
+ * The {@link #close} method must eventually be invoked on this instance.
+ *
+ * @param exception the exception that occurred during initialization
+ * @param closeHandler a callback function that will be invoked when the registration is closed
+ * @param eventProducer an event producer that will be invoked when the endpoint is updated
+ */
+ public ExportRegistrationImpl(Throwable exception, CloseHandler closeHandler, EventProducer eventProducer) {
+ shared = new Shared();
+ shared.closeHandler = closeHandler;
+ shared.eventProducer = eventProducer;
+ shared.exception = exception;
+ shared.addInstance();
}
/**
- * Returns the ExportReference even if this
- * instance is closed or has an exception.
+ * Returns whether this registration is invalid
+ * (has an initialization exception) or closed.
*
- * @return the export reference
+ * @return whether this registration is invalid or closed
*/
- public ExportReference getExportReferenceAlways() {
- return exportReference;
+ private boolean isInvalid() {
+ return shared.exception != null || closed;
}
+ @Override
public ExportReference getExportReference() {
if (closed) {
return null;
- } else {
- if (exportReference == null) {
- throw new IllegalStateException(getException());
- } else {
- return exportReference;
- }
}
+ if (shared.exception != null) {
+ throw new IllegalStateException("export registration is invalid");
+ }
+ return exportReference;
}
+ @Override
public Throwable getException() {
- return closed ? null : exception;
+ return closed ? null : shared.exception;
}
+ @Override
public final void close() {
- closeHandler.onClose(this);
- synchronized (this) {
- if (closed) {
- return;
- }
- closed = true;
+ // we do this in two steps: first the 'closing' state to make sure we
+ // proceed at most once (only if valid and not already closing/closed),
+ // with all the references still valid while calling the close handler,
+ // and only after it's done we reach the 'closed' state where all
+ // getters return null (as required by the spec)
+ LOG.debug("closing export registration");
+ if (closed || closing.getAndSet(true)) {
+ return;
}
-
+ shared.closeHandler.onClose(this);
if (exportReference != null) {
exportReference.close();
}
- parent.removeInstance();
- }
-
- private void addInstance() {
- ensureParent();
- synchronized (this) {
- instanceCount++;
- }
- }
-
- private void removeInstance() {
- ensureParent();
- synchronized (this) {
- instanceCount--;
- if (instanceCount <= 0) {
- LOG.debug("really closing ExportRegistration now!");
- closeServer();
- }
- }
- }
-
- private void closeServer() {
- if (server != null) {
- try {
- server.close();
- } catch (IOException e) {
- LOG.warn("Error closing ExportRegistration", e);
- }
- }
+ shared.removeInstance();
+ closed = true;
}
@Override
@@ -162,40 +191,39 @@ public class ExportRegistrationImpl implements ExportRegistration {
if (closed) {
return "ExportRegistration closed";
}
- EndpointDescription endpoint = getExportReference().getExportedEndpoint();
- ServiceReference serviceReference = getExportReference().getExportedService();
- String r = "EndpointDescription for ServiceReference " + serviceReference;
+ EndpointDescription endpoint = exportReference.getExportedEndpoint();
+ ServiceReference serviceReference = exportReference.getExportedService();
+ String s = "EndpointDescription for ServiceReference " + serviceReference;
- r += "\n*** EndpointDescription: ****\n";
+ s += "\n*** EndpointDescription: ****\n";
if (endpoint == null) {
- r += "---> NULL <---- \n";
+ s += "---> NULL <---- \n";
} else {
Set<Map.Entry<String, Object>> props = endpoint.getProperties().entrySet();
for (Map.Entry<String, Object> entry : props) {
Object value = entry.getValue();
- r += entry.getKey() + " => "
+ s += entry.getKey() + " => "
+ (value instanceof Object[] ? Arrays.toString((Object[]) value) : value) + "\n";
}
}
- return r;
+ return s;
}
@Override
public EndpointDescription update(Map<String, ?> properties) {
- ExportReference ref = getExportReference();
- if (ref == null) {
- return null;
+ if (isInvalid()) {
+ throw new IllegalStateException("export registration is invalid or closed");
}
- Map<String, Object> oldProps = ref.getExportedEndpoint().getProperties();
+ Map<String, Object> oldProps = exportReference.getExportedEndpoint().getProperties();
Map<String, Object> props = new HashMap<>(properties);
props.putIfAbsent(RemoteConstants.ENDPOINT_ID, oldProps.get(RemoteConstants.ENDPOINT_ID));
props.putIfAbsent(RemoteConstants.SERVICE_IMPORTED_CONFIGS, oldProps.get(RemoteConstants.SERVICE_IMPORTED_CONFIGS));
- ServiceReference<?> sref = ref.getExportedService();
- EndpointDescription epd = new EndpointDescription(sref, props);
- exportReference = new ExportReferenceImpl(sref, epd);
- this.sender.notifyUpdate(ref);
- return epd;
+ ServiceReference<?> sref = exportReference.getExportedService();
+ EndpointDescription endpoint = new EndpointDescription(sref, props);
+ exportReference = new ExportReferenceImpl(sref, endpoint);
+ shared.eventProducer.notifyUpdate(exportReference);
+ return endpoint;
}
}
diff --git a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
index 179df20c..61e25f7c 100644
--- a/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
+++ b/topology-manager/src/main/java/org/apache/aries/rsa/topologymanager/exporter/ServiceExportsRepository.java
@@ -66,7 +66,7 @@ public class ServiceExportsRepository implements Closeable {
public synchronized void addService(ServiceReference<?> sref, Collection<ExportRegistration> registrations) {
Map<ExportRegistration, EndpointDescription> regs = new LinkedHashMap<>();
for (ExportRegistration reg : registrations) {
- ExportReference ref = reg.getExportReference();
+ ExportReference ref = reg.getException() != null ? null : reg.getExportReference();
EndpointDescription endpoint = ref == null ? null : ref.getExportedEndpoint();
if (endpoint != null) {
regs.put(reg, endpoint);
@@ -83,10 +83,14 @@ public class ServiceExportsRepository implements Closeable {
if (regs != null) {
Map<String, ?> props = getServiceProps(sref);
for (Map.Entry<ExportRegistration, EndpointDescription> entry: regs.entrySet()) {
- EndpointDescription updated = entry.getKey().update(props);
- if (updated != null) {
- entry.setValue(updated);
- notifier.sendEvent(new EndpointEvent(EndpointEvent.MODIFIED, updated));
+ try {
+ EndpointDescription updated = entry.getKey().update(props);
+ if (updated != null) {
+ entry.setValue(updated);
+ notifier.sendEvent(new EndpointEvent(EndpointEvent.MODIFIED, updated));
+ }
+ } catch (IllegalStateException ise) {
+ LOG.error("export registration is invalid or closed", ise);
}
}
}
(aries-rsa) 01/05: ARIES-2212 Refactor ImportRegistrationImpl initialization to use one-time init instead of setters
Posted by am...@apache.org.
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 37191ba920ab0bed2216c77fe21422ae95efcd0a
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 23 11:43:59 2026 +0200
ARIES-2212 Refactor ImportRegistrationImpl initialization to use one-time init instead of setters
---
.../aries/rsa/core/ImportRegistrationImpl.java | 80 +++++++++++-----------
.../aries/rsa/core/RemoteServiceAdminCore.java | 10 ++-
.../aries/rsa/core/ImportRegistrationImplTest.java | 8 +--
3 files changed, 49 insertions(+), 49 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
index f7151cce..13cbe900 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
@@ -51,23 +51,28 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
private EventProducer eventProducer;
- public ImportRegistrationImpl(Throwable ex) {
- exception = ex;
- initParent();
- }
-
+ /**
+ * Constructs a new import registration.
+ *
+ * @param endpoint the imported endpoint info
+ * @param closeHandler a callback function that will be invoked when the registration is closed
+ * @param eventProducer an event producer that will be invoked when the endpoint is updated
+ */
public ImportRegistrationImpl(EndpointDescription endpoint, CloseHandler closeHandler, EventProducer eventProducer) {
this.endpoint = endpoint;
this.closeHandler = closeHandler;
this.eventProducer = eventProducer;
- initParent();
+ parent = this;
+ children = new ArrayList<>(1);
}
/**
- * Creates a clone of the given parent instance.
+ * Constructs an import registration that is linked
+ * (shares state) with the given import registration.
+ *
+ * @param ir the import registration that this instance is linked to
*/
public ImportRegistrationImpl(ImportRegistration ir) {
- // we always want a link to the parent...
parent = ((ImportRegistrationImpl)ir).getParent();
exception = parent.getException();
endpoint = parent.getImportedEndpointDescription();
@@ -77,9 +82,34 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
parent.instanceAdded(this);
}
- private void initParent() {
- parent = this;
- children = new ArrayList<>(1);
+ /**
+ * Initializes this import registration in an error state.
+ *
+ * @param exception the exception that occurred during initialization
+ */
+ public void init(Throwable exception) {
+ if (parent.exception != null || parent.clientServiceFactory != null || parent.importedService != null) {
+ throw new IllegalStateException("already initialized");
+ }
+ this.exception = exception;
+ }
+
+ /**
+ * Initializes this import registration in a successful state.
+ *
+ * @param csf the {@link ClientServiceFactory} which is the implementation
+ * of the locally registered service which provides proxies to the
+ * remote imported service.
+ * @param sreg the {@link ServiceRegistration} representing the locally
+ * registered {@link ClientServiceFactory} service which provides
+ * proxies to the remote imported service.
+ */
+ public void init(ClientServiceFactory csf, ServiceRegistration sreg) {
+ if (parent.exception != null || parent.clientServiceFactory != null || parent.importedService != null) {
+ throw new IllegalStateException("already initialized");
+ }
+ this.clientServiceFactory = csf;
+ this.importedService = sreg;
}
private void ensureParent() {
@@ -189,38 +219,10 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
return exception;
}
- public void setException(Throwable ex) {
- exception = ex;
- }
-
private synchronized boolean isInvalid() {
return exception != null || closed.get();
}
- /**
- * Sets the {@link ServiceRegistration} representing the locally
- * registered {@link ClientServiceFactory} service which provides
- * proxies to the remote imported service. It is set only on the parent.
- *
- * @param sreg the ServiceRegistration
- */
- public void setImportedServiceRegistration(ServiceRegistration sreg) {
- ensureParent();
- importedService = sreg;
- }
-
- /**
- * Sets the {@link ClientServiceFactory} which is the implementation
- * of the locally registered service which provides proxies to the
- * remote imported service. It is set only on the parent.
- *
- * @param csf the ClientServiceFactory
- */
- public void setClientServiceFactory(ClientServiceFactory csf) {
- ensureParent();
- clientServiceFactory = csf;
- }
-
public ImportRegistrationImpl getParent() {
return parent;
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
index a347d356..b05477ee 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
@@ -460,28 +460,26 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
protected ImportRegistrationImpl exposeServiceFactory(String[] interfaceNames,
- EndpointDescription epd,
+ EndpointDescription endpoint,
DistributionProvider handler) {
- ImportRegistrationImpl imReg = new ImportRegistrationImpl(epd, closeHandler, eventProducer);
+ ImportRegistrationImpl imReg = new ImportRegistrationImpl(endpoint, closeHandler, eventProducer);
try {
- EndpointDescription endpoint = imReg.getImportedEndpointDescription();
Dictionary<String, Object> serviceProps = new Hashtable<>(endpoint.getProperties());
serviceProps.put(RemoteConstants.SERVICE_IMPORTED, true);
serviceProps.remove(RemoteConstants.SERVICE_EXPORTED_INTERFACES);
ClientServiceFactory csf = new ClientServiceFactory(endpoint, handler, imReg);
- imReg.setClientServiceFactory(csf);
// Export the factory using the api context as it has very few imports.
// If the bundle publishing the factory does not import the service interface
// package then the factory is visible for all consumers which we want.
ServiceRegistration<?> csfReg = apictx.registerService(interfaceNames, csf, serviceProps);
- imReg.setImportedServiceRegistration(csfReg);
+ imReg.init(csf, csfReg);
} catch (Exception ex) {
// Only logging at debug level as this might be written to the log at the TopologyManager
LOG.debug("Can not proxy service with interfaces {}: {}",
Arrays.toString(interfaceNames), ex.getMessage(), ex);
- imReg.setException(ex);
+ imReg.init(ex);
}
return imReg;
}
diff --git a/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java b/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
index dd398a12..d388a390 100644
--- a/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
+++ b/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
@@ -35,7 +35,8 @@ public class ImportRegistrationImplTest {
Exception e = c.createMock(Exception.class);
c.replay();
- ImportRegistrationImpl i = new ImportRegistrationImpl(e);
+ ImportRegistrationImpl i = new ImportRegistrationImpl(null, null, null);
+ i.init(e);
assertEquals(e, i.getException());
assertNull(i.getImportedEndpointDescription());
@@ -72,20 +73,19 @@ public class ImportRegistrationImplTest {
c.replay();
ImportRegistrationImpl i1 = new ImportRegistrationImpl(endpoint, closeHandler, null);
+ i1.init(null, sr);
ImportRegistrationImpl i2 = new ImportRegistrationImpl(i1);
ImportRegistrationImpl i3 = new ImportRegistrationImpl(i2);
try {
- i2.setImportedServiceRegistration(sr);
+ i2.init(null, sr);
fail("An exception should be thrown here !");
} catch (IllegalStateException e) {
// must be thrown here
}
- i1.setImportedServiceRegistration(sr);
-
assertEquals(i1, i1.getParent());
assertEquals(i1, i2.getParent());
assertEquals(i1, i3.getParent());
(aries-rsa) 05/05: ARIES-2209 Merge ExportReferenceImpl into ExportRegistrationImpl
Posted by am...@apache.org.
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 d13e90db515b8dea366567e668aa892c22481106
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 23 23:41:52 2026 +0200
ARIES-2209 Merge ExportReferenceImpl into ExportRegistrationImpl
---
.../apache/aries/rsa/core/ExportReferenceImpl.java | 77 ----------------------
.../aries/rsa/core/ExportRegistrationImpl.java | 48 ++++++++------
.../aries/rsa/core/RemoteServiceAdminCore.java | 2 +-
3 files changed, 30 insertions(+), 97 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportReferenceImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportReferenceImpl.java
deleted file mode 100644
index 11e4f289..00000000
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportReferenceImpl.java
+++ /dev/null
@@ -1,77 +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.core;
-
-import org.osgi.framework.ServiceReference;
-import org.osgi.service.remoteserviceadmin.EndpointDescription;
-import org.osgi.service.remoteserviceadmin.ExportReference;
-
-import java.util.Objects;
-
-@SuppressWarnings("rawtypes")
-public class ExportReferenceImpl implements ExportReference {
-
- private ServiceReference serviceReference;
- private EndpointDescription endpoint;
-
- public ExportReferenceImpl(ServiceReference serviceReference, EndpointDescription endpoint) {
- this.serviceReference = serviceReference;
- this.endpoint = endpoint;
- }
-
- public ExportReferenceImpl(ExportReference exportReference) {
- this(exportReference.getExportedService(), exportReference.getExportedEndpoint());
- }
-
- public EndpointDescription getExportedEndpoint() {
- return endpoint;
- }
-
- public ServiceReference<?> getExportedService() {
- return serviceReference;
- }
-
- @Override
- public int hashCode() {
- final int prime = 31;
- int result = 1;
- result = prime * result + (endpoint == null ? 0 : endpoint.hashCode());
- result = prime * result + (serviceReference == null ? 0 : serviceReference.hashCode());
- return result;
- }
-
- @Override
- public boolean equals(Object obj) {
- if (this == obj) {
- return true;
- }
- if (obj == null || getClass() != obj.getClass()) {
- return false;
- }
- ExportReferenceImpl other = (ExportReferenceImpl) obj;
- boolean ed = Objects.equals(endpoint, other.endpoint);
- boolean sr = Objects.equals(serviceReference, other.serviceReference);
- return ed && sr;
- }
-
- synchronized void close() {
- this.endpoint = null;
- this.serviceReference = null;
- }
-}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
index 3c919c20..3f8a43d8 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
@@ -38,8 +38,14 @@ import org.osgi.service.remoteserviceadmin.RemoteConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Implements an ExportRegistration. Since there is a 1:1 relationship between
+ * an ExportRegistration and its ExportReference (they are basically two views
+ * of the same underlying data - one is the modifiable part and one the read-only
+ * part), this class implements both interfaces together.
+ */
@SuppressWarnings("rawtypes")
-public class ExportRegistrationImpl implements ExportRegistration {
+public class ExportRegistrationImpl implements ExportRegistration, ExportReference {
private static final Logger LOG = LoggerFactory.getLogger(ExportRegistrationImpl.class);
@@ -52,6 +58,8 @@ public class ExportRegistrationImpl implements ExportRegistration {
private Set<CloseHandler> closeHandlers = Collections.newSetFromMap(new ConcurrentHashMap<>());
private EventProducer eventProducer;
+ private ServiceReference serviceReference;
+ private volatile EndpointDescription endpoint;
private Closeable server;
private Throwable exception;
@@ -90,7 +98,6 @@ public class ExportRegistrationImpl implements ExportRegistration {
private final Shared shared;
// per-instance state that is not shared
- private volatile ExportReferenceImpl exportReference;
private AtomicBoolean closing = new AtomicBoolean();
private volatile boolean closed;
@@ -104,7 +111,6 @@ public class ExportRegistrationImpl implements ExportRegistration {
* @param er the export registration that this instance is linked to
*/
public ExportRegistrationImpl(ExportRegistrationImpl er) {
- exportReference = new ExportReferenceImpl(er.exportReference);
shared = er.shared;
shared.addInstance();
}
@@ -123,9 +129,10 @@ public class ExportRegistrationImpl implements ExportRegistration {
// create a new (parent) instance which was exported successfully with the given server
public ExportRegistrationImpl(ServiceReference sref, Endpoint endpoint,
CloseHandler closeHandler, EventProducer eventProducer) {
- exportReference = new ExportReferenceImpl(sref, endpoint.description());
shared = new Shared();
shared.eventProducer = eventProducer;
+ shared.serviceReference = sref;
+ shared.endpoint = endpoint.description();
shared.server = endpoint;
addCloseHandler(closeHandler);
shared.addInstance();
@@ -159,6 +166,16 @@ public class ExportRegistrationImpl implements ExportRegistration {
return shared.exception != null || closed;
}
+ @Override
+ public EndpointDescription getExportedEndpoint() {
+ return isInvalid() ? null : shared.endpoint;
+ }
+
+ @Override
+ public ServiceReference<?> getExportedService() {
+ return isInvalid() ? null : shared.serviceReference;
+ }
+
@Override
public ExportReference getExportReference() {
if (closed) {
@@ -167,7 +184,7 @@ public class ExportRegistrationImpl implements ExportRegistration {
if (shared.exception != null) {
throw new IllegalStateException("export registration is invalid");
}
- return exportReference;
+ return this; // this instance implements both interfaces
}
@Override
@@ -191,9 +208,6 @@ public class ExportRegistrationImpl implements ExportRegistration {
return;
}
shared.closeHandlers.forEach(h -> h.onClose(this));
- if (exportReference != null) {
- exportReference.close();
- }
shared.removeInstance();
closed = true;
}
@@ -203,15 +217,13 @@ public class ExportRegistrationImpl implements ExportRegistration {
if (closed) {
return "ExportRegistration closed";
}
- EndpointDescription endpoint = exportReference.getExportedEndpoint();
- ServiceReference serviceReference = exportReference.getExportedService();
- String s = "EndpointDescription for ServiceReference " + serviceReference;
+ String s = "EndpointDescription for ServiceReference " + shared.serviceReference;
s += "\n*** EndpointDescription: ****\n";
- if (endpoint == null) {
+ if (shared.endpoint == null) {
s += "---> NULL <---- \n";
} else {
- Set<Map.Entry<String, Object>> props = endpoint.getProperties().entrySet();
+ Set<Map.Entry<String, Object>> props = shared.endpoint.getProperties().entrySet();
for (Map.Entry<String, Object> entry : props) {
Object value = entry.getValue();
s += entry.getKey() + " => "
@@ -227,15 +239,13 @@ public class ExportRegistrationImpl implements ExportRegistration {
throw new IllegalStateException("export registration is invalid or closed");
}
- Map<String, Object> oldProps = exportReference.getExportedEndpoint().getProperties();
+ Map<String, Object> oldProps = shared.endpoint.getProperties();
Map<String, Object> props = new HashMap<>(properties);
props.putIfAbsent(RemoteConstants.ENDPOINT_ID, oldProps.get(RemoteConstants.ENDPOINT_ID));
props.putIfAbsent(RemoteConstants.SERVICE_IMPORTED_CONFIGS, oldProps.get(RemoteConstants.SERVICE_IMPORTED_CONFIGS));
- ServiceReference<?> sref = exportReference.getExportedService();
- EndpointDescription endpoint = new EndpointDescription(sref, props);
- exportReference = new ExportReferenceImpl(sref, endpoint);
- shared.eventProducer.notifyUpdate(exportReference);
- return endpoint;
+ shared.endpoint = new EndpointDescription(shared.serviceReference, props);
+ shared.eventProducer.notifyUpdate(this);
+ return shared.endpoint;
}
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
index c3f74f51..bcf62eb5 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
@@ -365,7 +365,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
for (Collection<ExportRegistration> exportRegistrations : exportedServices.values()) {
for (ExportRegistration er : exportRegistrations) {
if (er.getException() == null && er.getExportReference() != null) {
- ers.add(new ExportReferenceImpl(er.getExportReference()));
+ ers.add(er.getExportReference());
}
}
}
(aries-rsa) 02/05: ARIES-2207 Re-implement ImportRegistrationImpl more simple, clear and compliant
Posted by am...@apache.org.
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 daa69700a5d832037acf507048eff1ad17b50bcc
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Mon Mar 23 18:29:27 2026 +0200
ARIES-2207 Re-implement ImportRegistrationImpl more simple, clear and compliant
---
.../aries/rsa/core/ImportRegistrationImpl.java | 267 +++++++++++----------
.../aries/rsa/core/RemoteServiceAdminCore.java | 2 +-
.../aries/rsa/core/ImportRegistrationImplTest.java | 22 +-
3 files changed, 151 insertions(+), 140 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
index 13cbe900..ef5e4189 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
@@ -32,54 +32,128 @@ import org.osgi.service.remoteserviceadmin.ImportRegistration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Implements an ImportRegistration. Since there is a 1:1 relationship between
+ * an ImportRegistration and its ImportReference (they are basically two views
+ * of the same underlying data - one is the modifiable part and one the read-only
+ * part), this class implements both interfaces together.
+ */
@SuppressWarnings("rawtypes")
public class ImportRegistrationImpl implements ImportRegistration, ImportReference {
private static final Logger LOG = LoggerFactory.getLogger(ImportRegistrationImpl.class);
- private volatile Throwable exception;
- private volatile ServiceRegistration importedService; // used only in parent
- private EndpointDescription endpoint;
- private volatile ClientServiceFactory clientServiceFactory;
- private CloseHandler closeHandler;
- private AtomicBoolean closed = new AtomicBoolean();
- private AtomicBoolean closing = new AtomicBoolean();
- private boolean detached; // used only in parent
+ /**
+ * According to the specs, multiple ImportRegistrations can be linked together,
+ * i.e. they have a shared state - this is implemented via this inner class,
+ * where every group of linked import registrations reference a single Shared instance.
+ */
+ private static class Shared {
+
+ private volatile EndpointDescription endpoint;
+ private CloseHandler closeHandler;
+ private EventProducer eventProducer;
+ private volatile Throwable exception;
+ private volatile ClientServiceFactory clientServiceFactory;
+ private volatile ServiceRegistration importedService;
+ // all linked import registrations that share this state
+ private final List<ImportRegistrationImpl> instances = new ArrayList<>(1);
+
+ /**
+ * Add a linked ImportRegistration instance to the shared data.
+ *
+ * @param ir the instance to add
+ */
+ private synchronized void addInstance(ImportRegistrationImpl ir) {
+ instances.add(ir);
+ }
+
+ /**
+ * Remove a linked ImportRegistration from the shared data.
+ *
+ * @param ir the instance to remove
+ */
+ private void removeInstance(ImportRegistrationImpl ir) {
+ // close the underlying service only once on the last remove (not before or after)
+ synchronized (this) {
+ boolean removed = instances.remove(ir);
+ if (!removed || !instances.isEmpty()) {
+ return;
+ }
+ }
+
+ LOG.debug("closing ImportRegistration after removing last linked instance");
+ if (importedService != null) {
+ try {
+ importedService.unregister();
+ } catch (IllegalStateException ise) {
+ LOG.debug("imported service is already unregistered");
+ }
+ importedService = null;
+ }
+ if (clientServiceFactory != null) {
+ clientServiceFactory.setCloseable(true);
+ }
+ }
+
+ void closeAll() {
+ // make a copy to avoid ConcurrentModificationException
+ // when instances are removed during iteration
+ List<ImportRegistrationImpl> copy;
+ synchronized (this) {
+ copy = new ArrayList<>(instances);
+ }
+ LOG.info("closing all linked ImportRegistrations");
+ for (ImportRegistrationImpl ir : copy) {
+ ir.close();
+ }
+ }
+ }
- private ImportRegistrationImpl parent;
- private List<ImportRegistrationImpl> children; // used only in parent
+ // state shared between all linked import registrations in this instance's group
+ private Shared shared;
- private EventProducer eventProducer;
+ // per-instance state that is not shared
+ private AtomicBoolean closing = new AtomicBoolean();
+ private volatile boolean closed;
/**
- * Constructs a new import registration.
+ * Constructs a new import registration that is not linked to any other instance.
+ * <p>
+ * The {@link #close} method must eventually be invoked on this instance.
*
* @param endpoint the imported endpoint info
* @param closeHandler a callback function that will be invoked when the registration is closed
* @param eventProducer an event producer that will be invoked when the endpoint is updated
*/
public ImportRegistrationImpl(EndpointDescription endpoint, CloseHandler closeHandler, EventProducer eventProducer) {
- this.endpoint = endpoint;
- this.closeHandler = closeHandler;
- this.eventProducer = eventProducer;
- parent = this;
- children = new ArrayList<>(1);
+ shared = new Shared();
+ shared.endpoint = endpoint;
+ shared.closeHandler = closeHandler;
+ shared.eventProducer = eventProducer;
+ shared.addInstance(this);
}
/**
* Constructs an import registration that is linked
* (shares state) with the given import registration.
+ * <p>
+ * The {@link #close} method must eventually be invoked on this instance.
*
* @param ir the import registration that this instance is linked to
*/
public ImportRegistrationImpl(ImportRegistration ir) {
- parent = ((ImportRegistrationImpl)ir).getParent();
- exception = parent.getException();
- endpoint = parent.getImportedEndpointDescription();
- clientServiceFactory = parent.clientServiceFactory;
- closeHandler = parent.closeHandler;
+ shared = ((ImportRegistrationImpl)ir).shared;
+ shared.addInstance(this);
+ }
- parent.instanceAdded(this);
+ private void init(ClientServiceFactory csf, ServiceRegistration sreg, Throwable exception) {
+ if (isInvalid() || shared.clientServiceFactory != null || shared.importedService != null) {
+ throw new IllegalStateException("already initialized");
+ }
+ shared.clientServiceFactory = csf;
+ shared.importedService = sreg;
+ shared.exception = exception;
}
/**
@@ -88,10 +162,7 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
* @param exception the exception that occurred during initialization
*/
public void init(Throwable exception) {
- if (parent.exception != null || parent.clientServiceFactory != null || parent.importedService != null) {
- throw new IllegalStateException("already initialized");
- }
- this.exception = exception;
+ init(null, null, exception);
}
/**
@@ -105,134 +176,84 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
* proxies to the remote imported service.
*/
public void init(ClientServiceFactory csf, ServiceRegistration sreg) {
- if (parent.exception != null || parent.clientServiceFactory != null || parent.importedService != null) {
- throw new IllegalStateException("already initialized");
- }
- this.clientServiceFactory = csf;
- this.importedService = sreg;
- }
-
- private void ensureParent() {
- if (parent != this) {
- throw new IllegalStateException("this method may only be called on the parent");
- }
+ init(csf, sreg, null);
}
/**
- * Called on parent when a child is added.
+ * Returns whether this registration is invalid
+ * (has an initialization exception) or closed.
*
- * @param iri the child
+ * @return whether this registration is invalid or closed
*/
- private synchronized void instanceAdded(ImportRegistrationImpl iri) {
- ensureParent();
- children.add(iri);
+ private boolean isInvalid() {
+ return shared.exception != null || closed;
}
/**
- * Called on parent when a child is closed.
- *
- * @param iri the child
+ * Closes this instance.
+ * <p>
+ * If this is the last instance in its group of linked instances,
+ * the underlying import registration is also closed, otherwise
+ * the import itself remains open for use by the other instances.
*/
- private void instanceClosed(ImportRegistration iri) {
- ensureParent();
- synchronized (this) {
- children.remove(iri);
- if (!children.isEmpty() || detached || !closing.get()) {
- return;
- }
- detached = true;
- }
-
- LOG.debug("really closing ImportRegistration now");
-
- if (importedService != null) {
- try {
- importedService.unregister();
- } catch (IllegalStateException ise) {
- LOG.debug("imported service is already unregistered");
- }
- importedService = null;
- }
- if (clientServiceFactory != null) {
- clientServiceFactory.setCloseable(true);
- }
- }
-
+ @Override
public void close() {
- LOG.debug("close() called");
- synchronized (this) {
- boolean curClosing = closing.getAndSet(true);
- if (curClosing || isInvalid()) {
- return;
- }
+ // we do this in two steps: first the 'closing' state to make sure we
+ // proceed at most once (only if valid and not already closing/closed),
+ // with all the references still valid while calling the close handler,
+ // and only after it's done we reach the 'closed' state where all
+ // getters return null (as required by the spec)
+ LOG.debug("closing import registration");
+ if (closed || closing.getAndSet(true)) {
+ return;
}
- closeHandler.onClose(this);
- parent.instanceClosed(this);
- closed.set(true);
+ shared.closeHandler.onClose(this);
+ shared.removeInstance(this);
+ closed = true;
}
/**
- * Closes all ImportRegistrations which share the same parent as this one.
+ * Closes this instance, all linked instances, and the underlying service.
*/
public void closeAll() {
- if (this == parent) {
- LOG.info("closing down all child ImportRegistrations");
-
- // we must iterate over a copy of children since close() removes the child
- // from the list (which would cause a ConcurrentModificationException)
- for (ImportRegistrationImpl ir : copyChildren()) {
- ir.close();
- }
- this.close();
- } else {
- parent.closeAll();
- }
+ shared.closeAll(); // invokes close() on all shared instances (including this one)
}
- private List<ImportRegistrationImpl> copyChildren() {
- synchronized (this) {
- return new ArrayList<>(children);
+ @Override
+ @SuppressWarnings("unchecked")
+ public boolean update(EndpointDescription endpoint) {
+ if (isInvalid()) {
+ throw new IllegalStateException("import registration is invalid or closed");
}
- }
-
- public EndpointDescription getImportedEndpointDescription() {
- return isInvalid() ? null : endpoint;
+ if (!endpoint.getId().equals(shared.endpoint.getId())) {
+ throw new IllegalArgumentException("wrong endpoint id " + endpoint.getId());
+ }
+ shared.endpoint = endpoint;
+ shared.importedService.setProperties(new Hashtable<>(endpoint.getProperties()));
+ shared.eventProducer.notifyUpdate(this);
+ return true;
}
@Override
- public EndpointDescription getImportedEndpoint() {
- return getImportedEndpointDescription();
+ public ServiceReference<?> getImportedService() {
+ return isInvalid() ? null : shared.importedService.getReference();
}
@Override
- public ServiceReference<?> getImportedService() {
- return isInvalid() || parent.importedService == null ? null : parent.importedService.getReference();
+ public EndpointDescription getImportedEndpoint() {
+ return isInvalid() ? null : shared.endpoint;
}
@Override
public ImportReference getImportReference() {
- return this;
+ if (shared.exception != null) {
+ throw new IllegalStateException("import registration is invalid");
+ }
+ return isInvalid() ? null : this; // this instance implements both interfaces
}
@Override
public Throwable getException() {
- return exception;
- }
-
- private synchronized boolean isInvalid() {
- return exception != null || closed.get();
- }
-
- public ImportRegistrationImpl getParent() {
- return parent;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public boolean update(EndpointDescription endpoint) {
- this.endpoint = endpoint;
- importedService.setProperties(new Hashtable(endpoint.getProperties()));
- eventProducer.notifyUpdate(this);
- return true;
+ return closed ? null : shared.exception;
}
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
index b05477ee..41f97622 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
@@ -582,7 +582,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
synchronized (importedServices) {
LOG.debug("Removing importRegistration {}", iri);
- ImportReference importRef = iri.getImportReference();
+ ImportReference importRef = iri.getException() != null ? null : iri.getImportReference();
if (importRef == null) {
return;
}
diff --git a/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java b/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
index d388a390..aa20a929 100644
--- a/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
+++ b/rsa/src/test/java/org/apache/aries/rsa/core/ImportRegistrationImplTest.java
@@ -39,9 +39,8 @@ public class ImportRegistrationImplTest {
i.init(e);
assertEquals(e, i.getException());
- assertNull(i.getImportedEndpointDescription());
+ assertNull(i.getImportedEndpoint());
assertNull(i.getImportedService());
- assertEquals(i, i.getParent());
}
@Test
@@ -55,8 +54,7 @@ public class ImportRegistrationImplTest {
ImportRegistrationImpl i = new ImportRegistrationImpl(endpoint, closeHandler, null);
assertNull(i.getException());
- assertEquals(i, i.getParent());
- assertEquals(endpoint, i.getImportedEndpointDescription());
+ assertEquals(endpoint, i.getImportedEndpoint());
}
@SuppressWarnings("rawtypes")
@@ -86,13 +84,9 @@ public class ImportRegistrationImplTest {
// must be thrown here
}
- assertEquals(i1, i1.getParent());
- assertEquals(i1, i2.getParent());
- assertEquals(i1, i3.getParent());
-
- assertEquals(endpoint, i1.getImportedEndpointDescription());
- assertEquals(endpoint, i2.getImportedEndpointDescription());
- assertEquals(endpoint, i3.getImportedEndpointDescription());
+ assertEquals(endpoint, i1.getImportedEndpoint());
+ assertEquals(endpoint, i2.getImportedEndpoint());
+ assertEquals(endpoint, i3.getImportedEndpoint());
c.verify();
c.reset();
@@ -105,7 +99,7 @@ public class ImportRegistrationImplTest {
i3.close();
i3.close(); // shouldn't change anything
- assertNull(i3.getImportedEndpointDescription());
+ assertNull(i3.getImportedEndpoint());
c.verify();
c.reset();
@@ -147,10 +141,6 @@ public class ImportRegistrationImplTest {
ImportRegistrationImpl i3 = new ImportRegistrationImpl(i2);
- assertEquals(i1, i1.getParent());
- assertEquals(i1, i2.getParent());
- assertEquals(i1, i3.getParent());
-
c.verify();
c.reset();
(aries-rsa) 04/05: ARIES-2208 Fix incorrect closing of RSA factory/core/instance/registrations
Posted by am...@apache.org.
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 d874b29c91befa3769be6557494cbc497e36ffa9
Author: Amichai Rothman <am...@apache.org>
AuthorDate: Wed Mar 25 16:08:00 2026 +0200
ARIES-2208 Fix incorrect closing of RSA factory/core/instance/registrations
---
.../rsa/core/DistributionProviderTracker.java | 42 +++---
.../aries/rsa/core/ExportRegistrationImpl.java | 20 ++-
.../aries/rsa/core/ImportRegistrationImpl.java | 23 +++-
.../aries/rsa/core/RemoteServiceAdminCore.java | 141 ++++-----------------
.../aries/rsa/core/RemoteServiceAdminFactory.java | 27 +++-
.../aries/rsa/core/RemoteServiceAdminInstance.java | 74 +++++++----
.../rsa/core/DistributionProviderTrackerTest.java | 8 +-
7 files changed, 160 insertions(+), 175 deletions(-)
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/DistributionProviderTracker.java b/rsa/src/main/java/org/apache/aries/rsa/core/DistributionProviderTracker.java
index 12505e02..72714774 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/DistributionProviderTracker.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/DistributionProviderTracker.java
@@ -21,6 +21,7 @@ package org.apache.aries.rsa.core;
import static org.osgi.service.remoteserviceadmin.RemoteConstants.REMOTE_CONFIGS_SUPPORTED;
import static org.osgi.service.remoteserviceadmin.RemoteConstants.REMOTE_INTENTS_SUPPORTED;
+import java.util.Arrays;
import java.util.Dictionary;
import java.util.Hashtable;
@@ -37,8 +38,17 @@ import org.osgi.util.tracker.ServiceTracker;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-@SuppressWarnings("rawtypes")
-public class DistributionProviderTracker extends ServiceTracker<DistributionProvider, ServiceRegistration> {
+/**
+ * Tracks registered DistributionProviders, and for each one registers
+ * a {@link RemoteServiceAdminFactory} with an associated {@link RemoteServiceAdminCore}
+ * so that each TopologyManager will be given its own {@link RemoteServiceAdminInstance}
+ * to manage its exports/imports (backed by the shared core).
+ * <p>
+ * When the DistributionProvider is unregistered, we unregister its factory,
+ * which causes the framework to invoke ungetService for all of its instances,
+ * so they all shut down cleanly, close their exports/imports, etc.
+ */
+public class DistributionProviderTracker extends ServiceTracker<DistributionProvider, ServiceRegistration<RemoteServiceAdminFactory>> {
private static final Logger LOG = LoggerFactory.getLogger(DistributionProviderTracker.class);
public DistributionProviderTracker(BundleContext context) {
@@ -46,25 +56,26 @@ public class DistributionProviderTracker extends ServiceTracker<DistributionProv
}
@Override
- public ServiceRegistration addingService(ServiceReference<DistributionProvider> reference) {
+ @SuppressWarnings("unchecked")
+ public ServiceRegistration<RemoteServiceAdminFactory> addingService(ServiceReference<DistributionProvider> reference) {
DistributionProvider provider = context.getService(reference);
if (provider == null) {
// Can happen if the service is created by a service factory and an exception occurs
return null;
}
- LOG.debug("RemoteServiceAdmin Implementation is starting up");
+ LOG.debug("Initializing RemoteServiceAdmin for DistributionProvider {} ({})",
+ provider, Arrays.asList(provider.getSupportedTypes()));
BundleContext apiContext = getAPIContext();
EventProducer eventProducer = new EventProducer(context);
- RemoteServiceAdminCore rsaCore = new RemoteServiceAdminCore(context,
- apiContext,
- eventProducer,
- provider);
+ // we create one RSA core per tracked provider, accessed by its corresponding ServiceFactory
+ RemoteServiceAdminCore rsaCore = new RemoteServiceAdminCore(context, apiContext, eventProducer, provider);
RemoteServiceAdminFactory rsaf = new RemoteServiceAdminFactory(rsaCore);
Dictionary<String, Object> props = new Hashtable<>();
props.put(REMOTE_INTENTS_SUPPORTED, getPropertyNullSafe(reference, REMOTE_INTENTS_SUPPORTED));
props.put(REMOTE_CONFIGS_SUPPORTED, getPropertyNullSafe(reference, REMOTE_CONFIGS_SUPPORTED));
- LOG.info("Registering RemoteServiceAdmin for provider {}", provider.getClass().getName());
- return context.registerService(RemoteServiceAdmin.class.getName(), rsaf, props);
+ LOG.info("Registering RemoteServiceAdmin factory for provider {}", provider.getClass().getName());
+ return (ServiceRegistration<RemoteServiceAdminFactory>)
+ context.registerService(RemoteServiceAdmin.class.getName(), rsaf, props);
}
private Object getPropertyNullSafe(ServiceReference<DistributionProvider> reference, String key) {
@@ -84,10 +95,11 @@ public class DistributionProviderTracker extends ServiceTracker<DistributionProv
@Override
public void removedService(ServiceReference<DistributionProvider> reference,
- ServiceRegistration reg) {
- LOG.debug("RemoteServiceAdmin Implementation is shutting down now");
- reg.unregister();
- super.removedService(reference, reg);
+ ServiceRegistration<RemoteServiceAdminFactory> factoryRegistration) {
+ LOG.debug("Unregistering RemoteServiceAdmin factory for removed DistributionProvider");
+ // the provider is gone - unregister its corresponding factory,
+ // which will also cause all of its RSA instances (and their imports/exports) to be closed
+ factoryRegistration.unregister();
+ super.removedService(reference, factoryRegistration);
}
-
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
index a2fd7129..3c919c20 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ExportRegistrationImpl.java
@@ -21,9 +21,11 @@ package org.apache.aries.rsa.core;
import java.io.Closeable;
import java.io.IOException;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.aries.rsa.core.event.EventProducer;
@@ -48,13 +50,19 @@ public class ExportRegistrationImpl implements ExportRegistration {
*/
private static class Shared {
- private CloseHandler closeHandler;
+ private Set<CloseHandler> closeHandlers = Collections.newSetFromMap(new ConcurrentHashMap<>());
private EventProducer eventProducer;
private Closeable server;
private Throwable exception;
private int instanceCount;
+ void addCloseHandler(CloseHandler closeHandler) {
+ if (closeHandler != null) {
+ closeHandlers.add(closeHandler);
+ }
+ }
+
void addInstance() {
synchronized (this) {
instanceCount++;
@@ -117,9 +125,9 @@ public class ExportRegistrationImpl implements ExportRegistration {
CloseHandler closeHandler, EventProducer eventProducer) {
exportReference = new ExportReferenceImpl(sref, endpoint.description());
shared = new Shared();
- shared.closeHandler = closeHandler;
shared.eventProducer = eventProducer;
shared.server = endpoint;
+ addCloseHandler(closeHandler);
shared.addInstance();
}
@@ -135,9 +143,9 @@ public class ExportRegistrationImpl implements ExportRegistration {
*/
public ExportRegistrationImpl(Throwable exception, CloseHandler closeHandler, EventProducer eventProducer) {
shared = new Shared();
- shared.closeHandler = closeHandler;
shared.eventProducer = eventProducer;
shared.exception = exception;
+ addCloseHandler(closeHandler);
shared.addInstance();
}
@@ -167,6 +175,10 @@ public class ExportRegistrationImpl implements ExportRegistration {
return closed ? null : shared.exception;
}
+ public void addCloseHandler(CloseHandler closeHandler) {
+ shared.addCloseHandler(closeHandler);
+ }
+
@Override
public final void close() {
// we do this in two steps: first the 'closing' state to make sure we
@@ -178,7 +190,7 @@ public class ExportRegistrationImpl implements ExportRegistration {
if (closed || closing.getAndSet(true)) {
return;
}
- shared.closeHandler.onClose(this);
+ shared.closeHandlers.forEach(h -> h.onClose(this));
if (exportReference != null) {
exportReference.close();
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
index ef5e4189..589e37f0 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/ImportRegistrationImpl.java
@@ -18,9 +18,8 @@
*/
package org.apache.aries.rsa.core;
-import java.util.ArrayList;
-import java.util.Hashtable;
-import java.util.List;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.aries.rsa.core.event.EventProducer;
@@ -50,15 +49,21 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
*/
private static class Shared {
- private volatile EndpointDescription endpoint;
- private CloseHandler closeHandler;
+ private Set<CloseHandler> closeHandlers = Collections.newSetFromMap(new ConcurrentHashMap<>());
private EventProducer eventProducer;
+ private volatile EndpointDescription endpoint;
private volatile Throwable exception;
private volatile ClientServiceFactory clientServiceFactory;
private volatile ServiceRegistration importedService;
// all linked import registrations that share this state
private final List<ImportRegistrationImpl> instances = new ArrayList<>(1);
+ void addCloseHandler(CloseHandler closeHandler) {
+ if (closeHandler != null) {
+ closeHandlers.add(closeHandler);
+ }
+ }
+
/**
* Add a linked ImportRegistration instance to the shared data.
*
@@ -129,8 +134,8 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
public ImportRegistrationImpl(EndpointDescription endpoint, CloseHandler closeHandler, EventProducer eventProducer) {
shared = new Shared();
shared.endpoint = endpoint;
- shared.closeHandler = closeHandler;
shared.eventProducer = eventProducer;
+ addCloseHandler(closeHandler);
shared.addInstance(this);
}
@@ -189,6 +194,10 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
return shared.exception != null || closed;
}
+ public void addCloseHandler(CloseHandler closeHandler) {
+ shared.addCloseHandler(closeHandler);
+ }
+
/**
* Closes this instance.
* <p>
@@ -207,7 +216,7 @@ public class ImportRegistrationImpl implements ImportRegistration, ImportReferen
if (closed || closing.getAndSet(true)) {
return;
}
- shared.closeHandler.onClose(this);
+ shared.closeHandlers.forEach(h -> h.onClose(this));
shared.removeInstance(this);
closed = true;
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
index 41f97622..c3f74f51 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminCore.java
@@ -41,8 +41,6 @@ import org.apache.aries.rsa.util.EndpointHelper;
import org.apache.aries.rsa.util.StringPlus;
import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;
-import org.osgi.framework.ServiceEvent;
-import org.osgi.framework.ServiceListener;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.remoteserviceadmin.EndpointDescription;
@@ -68,7 +66,6 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
private final BundleContext bctx;
private final EventProducer eventProducer;
- private ServiceListener exportedServiceListener;
private DistributionProvider provider;
private BundleContext apictx;
private CloseHandler closeHandler;
@@ -82,13 +79,21 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
this.eventProducer = eventProducer;
this.provider = provider;
this.closeHandler = new CloseHandler() {
- public void onClose(ExportRegistration exportReg) {
- removeExportRegistration(exportReg);
- ExportReference exportReference = exportReg.getExportReference();
- if (exportReference != null) {
- ServiceReference<?> serviceReference = exportReference.getExportedService();
- if (serviceReference != null)
- getBundleContext(serviceReference).ungetService(serviceReference);
+ public void onClose(ExportRegistration reg) {
+ removeExportRegistration(reg);
+ if (reg.getException() != null) {
+ return; // there is no reference to close, and an exception if we try getting it
+ }
+ ExportReference ref = reg.getExportReference();
+ ServiceReference<?> sref = ref == null ? null : ref.getExportedService();
+ Bundle bundle = sref == null ? null : sref.getBundle();
+ BundleContext context = bundle == null ? null : bundle.getBundleContext();
+ // the bundle/context may already be closed, e.g. when called from
+ // the service factory ungetService when the bundle is stopped -
+ // the context is already invalid at that point, and services are
+ // automatically unregistered by the framework
+ if (context != null) {
+ context.ungetService(sref);
}
}
@@ -96,17 +101,6 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
removeImportRegistration(importReg);
}
};
- createServiceListener();
- }
-
- // listen for exported services being unregistered, so we can close the export
- protected void createServiceListener() {
- this.exportedServiceListener = event -> {
- if (event.getType() == ServiceEvent.UNREGISTERING) {
- removeServiceExports(event.getServiceReference());
- }
- };
- this.bctx.addServiceListener(exportedServiceListener);
}
@Override
@@ -125,7 +119,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
Map<String, Object> key = makeKey(serviceProperties);
- List<ExportRegistration> regs = getExistingAndLock(key, interfaceNames);
+ List<ExportRegistration> regs = getExistingOrLock(key, interfaceNames);
if (regs == null) {
regs = new ArrayList<>();
try {
@@ -177,7 +171,7 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
}
- private List<ExportRegistration> getExistingAndLock(Map<String, Object> key, List<String> interfaces) {
+ private List<ExportRegistration> getExistingOrLock(Map<String, Object> key, List<String> interfaces) {
synchronized (exportedServices) {
// check if it is already exported...
Collection<ExportRegistration> regs = exportedServices.get(key);
@@ -340,22 +334,17 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
// create a new list with copies of the exportRegistrations
List<ExportRegistration> copy = new ArrayList<>(regs.size());
- for (ExportRegistration exportRegistration : regs) {
- if (exportRegistration instanceof ExportRegistrationImpl) {
- ExportRegistrationImpl exportRegistrationImpl = (ExportRegistrationImpl) exportRegistration;
- if (exportRegistration.getException() == null) {
- // Can only retrieve reference if we have no exception
- EndpointDescription epd = exportRegistration.getExportReference().getExportedEndpoint();
- // create one copy for each distinct endpoint description
- if (!copiedEndpoints.contains(epd)) {
- copiedEndpoints.add(epd);
- copy.add(new ExportRegistrationImpl(exportRegistrationImpl));
- // also increase service reference count
- ServiceReference<?> serviceReference = exportRegistration.getExportReference().getExportedService();
- BundleContext serviceContext = getBundleContext(serviceReference);
- serviceContext.getService(serviceReference); // unget it when export is closed
- }
- }
+ for (ExportRegistration reg : regs) {
+ ExportRegistrationImpl exportRegistrationImpl = (ExportRegistrationImpl) reg;
+ ExportReference ref = reg.getException() != null ? null : reg.getExportReference();
+ EndpointDescription endpoint = ref == null ? null : ref.getExportedEndpoint();
+ // create one copy for each distinct endpoint description
+ if (endpoint != null && copiedEndpoints.add(endpoint)) {
+ copy.add(new ExportRegistrationImpl(exportRegistrationImpl));
+ // also increase service reference count
+ ServiceReference<?> sref = ref.getExportedService();
+ BundleContext serviceContext = getBundleContext(sref);
+ serviceContext.getService(sref); // unget it when export is closed
}
}
@@ -484,33 +473,6 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
return imReg;
}
- /**
- * Removes and closes all exports for the given service.
- * This is called when the service is unregistered.
- *
- * @param sref the service whose exports should be removed and closed
- */
- protected void removeServiceExports(ServiceReference<?> sref) {
- List<ExportRegistration> regs = new ArrayList<>(1);
- synchronized (exportedServices) {
- for (Collection<ExportRegistration> value : exportedServices.values()) {
- for (ExportRegistration er : value) {
- if (er.getException() != null ||
- er.getExportReference() == null ||
- er.getExportReference().getExportedService().equals(sref)) {
- regs.add(er);
- }
- }
- }
- // do this outside of iteration to avoid concurrent modification
- for (ExportRegistration er : regs) {
- LOG.debug("closing export for service {}", sref);
- er.close();
- }
- }
-
- }
-
/**
* Removes the provided Export Registration from the internal management structures.
* This is called from the ExportRegistration itself when it is closed (so should
@@ -539,45 +501,6 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
}
- // remove all export registrations associated with the given bundle
- protected void removeExportRegistrations(Bundle exportingBundle) {
- List<ExportRegistration> bundleExports = getExportsForBundle(exportingBundle);
- for (ExportRegistration export : bundleExports) {
- export.close();
- }
- }
-
- // remove all import registrations
- protected void closeImportRegistrations() {
- Collection<ImportRegistration> copy = new ArrayList<>();
- synchronized (importedServices) {
- for (Collection<ImportRegistration> irs : importedServices.values()) {
- copy.addAll(irs);
- }
- }
- for (ImportRegistration ir : copy) {
- ir.close();
- }
- }
-
- private List<ExportRegistration> getExportsForBundle(Bundle exportingBundle) {
- synchronized (exportedServices) {
- List<ExportRegistration> bundleRegs = new ArrayList<>();
- for (Collection<ExportRegistration> regs : exportedServices.values()) {
- if (!regs.isEmpty()) {
- ExportRegistration exportRegistration = regs.iterator().next();
- if (exportRegistration.getException() == null && exportRegistration.getExportReference() != null) {
- Bundle regBundle = exportRegistration.getExportReference().getExportedService().getBundle();
- if (exportingBundle.equals(regBundle)) {
- bundleRegs.addAll(regs);
- }
- }
- }
- }
- return bundleRegs;
- }
- }
-
protected void removeImportRegistration(ImportRegistration iri) {
synchronized (importedServices) {
LOG.debug("Removing importRegistration {}", iri);
@@ -599,14 +522,6 @@ public class RemoteServiceAdminCore implements RemoteServiceAdmin {
}
}
- public void close() {
- LOG.info("Closing {}", this.getClass().getSimpleName());
- closeImportRegistrations();
- if (exportedServiceListener != null) {
- bctx.removeServiceListener(exportedServiceListener);
- }
- }
-
static void overlayProperties(Map<String, Object> serviceProperties,
Map<String, Object> additionalProperties) {
Map<String, String> keysLowerCase = new HashMap<>();
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminFactory.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminFactory.java
index fe4fcf72..92730573 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminFactory.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminFactory.java
@@ -25,27 +25,42 @@ import org.osgi.service.remoteserviceadmin.RemoteServiceAdmin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * The {@link ServiceFactory} instance that supplies bundles each with their own RSA instance.
+ * <p>
+ * For every bundle, the first time it gets the service the factory's getService is invoked
+ * to create the instance, and then the framework caches the instance and returns it when
+ * the bundle subsequently calls get/unget while managing its reference count. Only when
+ * the reference count reaches zero (the last time unget is called) is the factory's
+ * ungetService invoked to destroy the instance.
+ * <p>
+ * We use a ServiceFactory according to the spec's recommendation, in order to implement
+ * the requirement that if a TopologyManager bundle is stopped, all registrations that
+ * were made by it must be closed, and when the RSA bundle is stopped all of its
+ * registrations must be closed as well. In both cases, when using a factory, our
+ * ungetService method will be invoked when either bundle is stopped, since the framework
+ * automatically ungets all remaining services when stopped.
+ */
public class RemoteServiceAdminFactory implements ServiceFactory<RemoteServiceAdmin> {
private static final Logger LOG = LoggerFactory.getLogger(RemoteServiceAdminFactory.class);
private final RemoteServiceAdminCore rsaCore;
- private int instances;
public RemoteServiceAdminFactory(RemoteServiceAdminCore rsaCore) {
this.rsaCore = rsaCore;
}
public synchronized RemoteServiceAdmin getService(Bundle b, ServiceRegistration<RemoteServiceAdmin> sreg) {
- LOG.debug("new RemoteServiceAdmin ServiceInstance created for Bundle {}", b.getSymbolicName());
- instances++;
+ LOG.debug("new RemoteServiceAdmin instance created for Bundle {}", b.getSymbolicName());
return new RemoteServiceAdminInstance(b.getBundleContext(), rsaCore);
}
public synchronized void ungetService(Bundle b, ServiceRegistration<RemoteServiceAdmin> sreg,
RemoteServiceAdmin serviceObject) {
- LOG.debug("RemoteServiceAdmin ServiceInstance removed for Bundle {}", b.getSymbolicName());
- instances--;
- ((RemoteServiceAdminInstance)serviceObject).close(b, instances == 0);
+ LOG.debug("RemoteServiceAdmin instance destroyed for Bundle {}", b.getSymbolicName());
+ // close all registrations that were created by the calling bundle
+ // (typically TopologyManager) when it is stopped, as required by the spec
+ ((RemoteServiceAdminInstance)serviceObject).close();
}
}
diff --git a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminInstance.java b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminInstance.java
index 27aef96a..c7059483 100644
--- a/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminInstance.java
+++ b/rsa/src/main/java/org/apache/aries/rsa/core/RemoteServiceAdminInstance.java
@@ -21,11 +21,10 @@ package org.apache.aries.rsa.core;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Collection;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
-import org.osgi.framework.Bundle;
import org.osgi.framework.BundleContext;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceReference;
@@ -37,54 +36,67 @@ import org.osgi.service.remoteserviceadmin.ImportReference;
import org.osgi.service.remoteserviceadmin.ImportRegistration;
import org.osgi.service.remoteserviceadmin.RemoteServiceAdmin;
-public class RemoteServiceAdminInstance implements RemoteServiceAdmin {
+/**
+ * An RSA instance that is created and returned by the {@link RemoteServiceAdminFactory},
+ * one per calling bundle. All instances delegate the heavy lifting to a shared
+ * RSA core implementation, but each instance keeps track of the registrations that
+ * were performed through it so that it can later close them when the corresponding bundle
+ * (TopologyManager) is closed.
+ */
+public class RemoteServiceAdminInstance implements RemoteServiceAdmin, CloseHandler {
- // Context of the bundle requesting the RemoteServiceAdmin
- private final BundleContext bctx;
+ // Context of the bundle that is using this the RemoteServiceAdmin (i.e. the TopologyManager)
+ private final BundleContext context;
private final RemoteServiceAdminCore rsaCore;
+ private final List<ExportRegistration> exportRegistrations = new CopyOnWriteArrayList<>();
+ private final List<ImportRegistration> importRegistrations = new CopyOnWriteArrayList<>();
- private boolean closed;
-
- public RemoteServiceAdminInstance(BundleContext bc, RemoteServiceAdminCore core) {
- bctx = bc;
- rsaCore = core;
+ public RemoteServiceAdminInstance(BundleContext context, RemoteServiceAdminCore rsaCore) {
+ this.context = context;
+ this.rsaCore = rsaCore;
}
@Override
- @SuppressWarnings("rawtypes")
public List<ExportRegistration> exportService(final ServiceReference ref, final Map properties) {
- return closed ? Collections.emptyList() : rsaCore.exportService(ref, properties);
+ List<ExportRegistration> regs = rsaCore.exportService(ref, properties);
+ // we need to keep track of our open registrations, and be notified when they are closed
+ regs.forEach(reg -> ((ExportRegistrationImpl)reg).addCloseHandler(this));
+ exportRegistrations.addAll(regs);
+ return regs;
}
@Override
public Collection<ExportReference> getExportedServices() {
checkPermission(new EndpointPermission("*", EndpointPermission.READ));
- return closed ? null : rsaCore.getExportedServices();
+ return rsaCore.getExportedServices();
}
@Override
public Collection<ImportReference> getImportedEndpoints() {
checkPermission(new EndpointPermission("*", EndpointPermission.READ));
- return closed ? null : rsaCore.getImportedEndpoints();
+ return rsaCore.getImportedEndpoints();
}
@Override
public ImportRegistration importService(final EndpointDescription endpoint) {
- String frameworkUUID = bctx.getProperty(Constants.FRAMEWORK_UUID);
+ String frameworkUUID = context.getProperty(Constants.FRAMEWORK_UUID);
checkPermission(new EndpointPermission(endpoint, frameworkUUID, EndpointPermission.IMPORT));
- return AccessController.doPrivileged(new PrivilegedAction<ImportRegistration>() {
- public ImportRegistration run() {
- return closed ? null : rsaCore.importService(endpoint);
- }
- });
+ ImportRegistration reg = AccessController.doPrivileged((PrivilegedAction<ImportRegistration>)
+ () -> rsaCore.importService(endpoint));
+ if (reg != null) {
+ // we need to keep track of our open registrations, and be notified when they are closed
+ ((ImportRegistrationImpl)reg).addCloseHandler(this);
+ importRegistrations.add(reg);
+ }
+ return reg;
}
- public void close(Bundle bundle, boolean closeAll) {
- closed = true;
- rsaCore.removeExportRegistrations(bundle);
- if (closeAll) {
- rsaCore.close();
- }
+ /**
+ * Close all registrations that were made through this RSA instance.
+ */
+ public void close() {
+ importRegistrations.forEach(ImportRegistration::close);
+ exportRegistrations.forEach(ExportRegistration::close);
}
private void checkPermission(EndpointPermission permission) {
@@ -93,4 +105,14 @@ public class RemoteServiceAdminInstance implements RemoteServiceAdmin {
sm.checkPermission(permission);
}
}
+
+ @Override
+ public void onClose(ExportRegistration reg) {
+ exportRegistrations.remove(reg); // registration was closed, no need to track it anymore
+ }
+
+ @Override
+ public void onClose(ImportRegistration reg) {
+ importRegistrations.remove(reg); // registration was closed, no need to track it anymore
+ }
}
diff --git a/rsa/src/test/java/org/apache/aries/rsa/core/DistributionProviderTrackerTest.java b/rsa/src/test/java/org/apache/aries/rsa/core/DistributionProviderTrackerTest.java
index 6e1fa8aa..2b10771d 100644
--- a/rsa/src/test/java/org/apache/aries/rsa/core/DistributionProviderTrackerTest.java
+++ b/rsa/src/test/java/org/apache/aries/rsa/core/DistributionProviderTrackerTest.java
@@ -45,6 +45,7 @@ public class DistributionProviderTrackerTest {
public void testAddingRemoved() throws InvalidSyntaxException {
IMocksControl c = EasyMock.createControl();
DistributionProvider provider = c.createMock(DistributionProvider.class);
+ expect(provider.getSupportedTypes()).andReturn(new String[] { "config-type" }).anyTimes();
ServiceReference<DistributionProvider> providerRef = c.createMock(ServiceReference.class);
expect(providerRef.getProperty(RemoteConstants.REMOTE_INTENTS_SUPPORTED)).andReturn("");
@@ -59,8 +60,6 @@ public class DistributionProviderTrackerTest {
expect(context.registerService(EasyMock.isA(String.class), EasyMock.isA(ServiceFactory.class),
EasyMock.isA(Dictionary.class)))
.andReturn(rsaReg).atLeastOnce();
- context.addServiceListener(anyObject(ServiceListener.class));
- expectLastCall().anyTimes();
final BundleContext apiContext = c.createMock(BundleContext.class);
c.replay();
@@ -69,7 +68,7 @@ public class DistributionProviderTrackerTest {
return apiContext;
}
};
- tracker.addingService(providerRef);
+ ServiceRegistration<RemoteServiceAdminFactory> factoryRegistration = tracker.addingService(providerRef);
c.verify();
c.reset();
@@ -77,7 +76,7 @@ public class DistributionProviderTrackerTest {
EasyMock.expectLastCall();
EasyMock.expect(context.ungetService(providerRef)).andReturn(true);
c.replay();
- tracker.removedService(providerRef, rsaReg);
+ tracker.removedService(providerRef, factoryRegistration);
c.verify();
}
@@ -85,6 +84,7 @@ public class DistributionProviderTrackerTest {
public void testAddingWithNullValues() throws InvalidSyntaxException {
IMocksControl c = EasyMock.createControl();
DistributionProvider provider = c.createMock(DistributionProvider.class);
+ expect(provider.getSupportedTypes()).andReturn(new String[] { "config-type" }).anyTimes();
ServiceReference<DistributionProvider> providerRef = c.createMock(ServiceReference.class);
expect(providerRef.getProperty(RemoteConstants.REMOTE_INTENTS_SUPPORTED)).andReturn(null);