Skip to content

Commit

Permalink
fix: Prevent publishing birth on shutdown
Browse files Browse the repository at this point in the history 
Signed-off-by: MMaiero <[email protected]>
  • Loading branch information
@MMaiero
MMaiero committed Nov 4, 2024
1 parent 9ac6272 commit 62fbec0
Show file tree
Hide file tree
Showing 2 changed files with 48 additions and 27 deletions.
38 changes: 30 additions & 8 deletions ...lipse/kura/internal/cloudconnection/eclipseiot/mqtt/cloud/CloudConnectionManagerImpl.java
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
/*******************************************************************************
* Copyright (c) 2011, 2024 Eurotech and/or its affiliates and others
*
*
* This program and the accompanying materials are made
* available under the terms of the Eclipse Public License 2.0
* which is available at https://www.eclipse.org/legal/epl-2.0/
*
*
* SPDX-License-Identifier: EPL-2.0
*
*
* Contributors:
* Eurotech
*******************************************************************************/
Expand Down Expand Up @@ -82,6 +82,8 @@
import org.eclipse.kura.position.PositionService;
import org.eclipse.kura.system.SystemAdminService;
import org.eclipse.kura.system.SystemService;
import org.osgi.framework.Bundle;
import org.osgi.framework.FrameworkUtil;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.component.ComponentContext;
Expand Down Expand Up @@ -148,7 +150,7 @@ public class CloudConnectionManagerImpl
private final Set<CloudDeliveryListener> registeredCloudDeliveryListeners;

private ScheduledFuture<?> scheduledBirthPublisherFuture;
private ScheduledExecutorService scheduledBirthPublisher = Executors.newScheduledThreadPool(1);
private final ScheduledExecutorService scheduledBirthPublisher = Executors.newScheduledThreadPool(1);

public CloudConnectionManagerImpl() {
this.messageId = new AtomicInteger();
Expand Down Expand Up @@ -283,7 +285,7 @@ public void unsetNetworkStatusService(NetworkStatusService networkStatusService)
protected void activate(ComponentContext componentContext, Map<String, Object> properties) {
this.ownPid = (String) properties.get(ConfigurationService.KURA_SERVICE_PID);

logger.info("activate {}...", ownPid);
logger.info("activate {}...", this.ownPid);

//
// save the bundle context and the properties
Expand Down Expand Up @@ -368,8 +370,8 @@ public void handleEvent(Event event) {
return;
}

if ((EVENT_TOPIC_DEPLOYMENT_ADMIN_INSTALL.equals(topic)
|| EVENT_TOPIC_DEPLOYMENT_ADMIN_UNINSTALL.equals(topic)) && this.dataService.isConnected()) {
if ((EVENT_TOPIC_DEPLOYMENT_ADMIN_INSTALL.equals(topic) || EVENT_TOPIC_DEPLOYMENT_ADMIN_UNINSTALL.equals(topic))
&& this.dataService.isConnected()) {
logger.debug("CloudConnectionManagerImpl: received install/uninstall event, publishing BIRTH.");
tryPublishBirthCertificate(false);
}
Expand Down Expand Up @@ -655,6 +657,11 @@ private void setupDeviceSubscriptions() throws KuraException {
}

private void publishBirthCertificate(boolean isNewConnection) throws KuraException {
if (isFrameworkStopping()) {
logger.info("framework is stopping.. not republishing birth certificate");
return;
}

readModemProfile();
LifecycleMessage birthToPublish = new LifecycleMessage(this.options, this).asBirthCertificateMessage();

Expand Down Expand Up @@ -819,7 +826,7 @@ public String publish(KuraMessage message) throws KuraException {
}

String getOwnPid() {
return ownPid;
return this.ownPid;
}

@Override
Expand Down Expand Up @@ -861,6 +868,21 @@ public void unregisterCloudDeliveryListener(CloudDeliveryListener cloudDeliveryL
this.registeredCloudDeliveryListeners.remove(cloudDeliveryListener);
}

private boolean isFrameworkStopping() {
try {
final Bundle ownBundle = FrameworkUtil.getBundle(CloudConnectionManagerImpl.class);

if (ownBundle == null) {
return false; // not running in an OSGi framework? e.g. unit test
}

return ownBundle.getBundleContext().getBundle(0).getState() == Bundle.STOPPING;
} catch (final Exception e) {
logger.warn("unexpected exception while checking if framework is shutting down", e);
return false;
}
}

private void readModemProfile() {
this.networkStatusService.ifPresent(statusService -> {
List<ModemInterfaceStatus> modemStatuses = getModemsStatuses(statusService);
Expand Down
37 changes: 18 additions & 19 deletions ...g.eclipse.kura.core.cloud/src/main/java/org/eclipse/kura/core/cloud/CloudServiceImpl.java
View file
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
/*******************************************************************************
* Copyright (c) 2011, 2024 Eurotech and/or its affiliates and others
*
*
* This program and the accompanying materials are made
* available under the terms of the Eclipse Public License 2.0
* which is available at https://www.eclipse.org/legal/epl-2.0/
*
*
* SPDX-License-Identifier: EPL-2.0
*
*
* Contributors:
* Eurotech
*******************************************************************************/
Expand Down Expand Up @@ -171,12 +171,12 @@ public class CloudServiceImpl
private ServiceRegistration<?> notificationPublisherRegistration;
private final CloudNotificationPublisher notificationPublisher;

private Set<TamperDetectionService> tamperDetectionServices = new HashSet<>();
private final Set<TamperDetectionService> tamperDetectionServices = new HashSet<>();

private String ownPid;

private ScheduledFuture<?> scheduledBirthPublisherFuture;
private ScheduledExecutorService scheduledBirthPublisher = Executors.newScheduledThreadPool(1);
private final ScheduledExecutorService scheduledBirthPublisher = Executors.newScheduledThreadPool(1);
private LifecycleMessage lastBirthMessage;
private LifecycleMessage lastAppMessage;

Expand Down Expand Up @@ -329,7 +329,7 @@ public void unsetNetworkStatusService(NetworkStatusService networkStatusService)

protected void activate(ComponentContext componentContext, Map<String, Object> properties) {
this.ownPid = (String) properties.get(ConfigurationService.KURA_SERVICE_PID);
logger.info("activate {}...", ownPid);
logger.info("activate {}...", this.ownPid);

//
// save the bundle context and the properties
Expand Down Expand Up @@ -856,6 +856,11 @@ private void setupCloudConnection(boolean isNewConnection) throws KuraException
}

private void publishBirthCertificate(boolean isNewConnection) throws KuraException {
if (isFrameworkStopping()) {
logger.info("framework is stopping.. not republishing birth certificate");
return;
}

readModemProfile();
LifecycleMessage birthToPublish = new LifecycleMessage(this.options, this).asBirthCertificateMessage();

Expand All @@ -871,6 +876,10 @@ private void publishDisconnectCertificate() throws KuraException {
}

private void publishAppCertificate() {
if (isFrameworkStopping()) {
logger.info("framework is stopping.. not republishing app certificate");
return;
}
publishWithDelay(new LifecycleMessage(this.options, this).asAppCertificateMessage());
}

Expand All @@ -895,12 +904,12 @@ private void publishWithDelay(LifecycleMessage message) {

if (Objects.nonNull(this.lastBirthMessage)) {
logger.debug("CloudServiceImpl: publishing cached BIRTH message.");
publishLifeCycleMessage(lastBirthMessage);
publishLifeCycleMessage(this.lastBirthMessage);
}

if (Objects.nonNull(this.lastAppMessage)) {
logger.debug("CloudServiceImpl: publishing cached APP message.");
publishLifeCycleMessage(lastAppMessage);
publishLifeCycleMessage(this.lastAppMessage);
}

} catch (KuraException e) {
Expand Down Expand Up @@ -986,11 +995,6 @@ private void postConnectionStateChangeEvent(final boolean isConnected) {
public void registerRequestHandler(String appId, RequestHandler requestHandler) {
this.registeredRequestHandlers.put(appId, requestHandler);

if (isFrameworkStopping()) {
logger.info("framework is stopping.. not republishing app certificate");
return;
}

if (isConnected()) {
publishAppCertificate();
}
Expand All @@ -1000,11 +1004,6 @@ public void registerRequestHandler(String appId, RequestHandler requestHandler)
public void unregister(String appId) {
this.registeredRequestHandlers.remove(appId);

if (isFrameworkStopping()) {
logger.info("framework is stopping.. not republishing app certificate");
return;
}

if (isConnected()) {
publishAppCertificate();
}
Expand Down Expand Up @@ -1193,7 +1192,7 @@ public void unregisterCloudDeliveryListener(CloudDeliveryListener cloudDeliveryL
}

String getOwnPid() {
return ownPid;
return this.ownPid;
}

void withTamperDetectionServices(final Consumer<Set<TamperDetectionService>> consumer) {
Expand Down

0 comments on commit 62fbec0

Please sign in to comment.