From ffc2b4a6741e1f31ffad035df12f619a77c29960 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 16:58:39 +0200 Subject: [PATCH 01/16] Close the GOPACS REST client and cancel tracked tasks on undeploy GOPACSHandler kept its ResteasyClient in a field but undeploy() only cancelled the scheduled futures and undeployed the web service. Every update of an EmsGOPACSAsset does stop-then-start, so each edit left the previous client and its connection pool allocated. Closing is safe with the shared executor: createClient(ExecutorService) goes through ResteasyClientBuilder.executorService, which leaves Container.EXECUTOR untouched. The constructor now closes the client if client.target(...) throws between createClient and the end of construction, since no reference escapes a constructor that threw and undeploy() could never run. Scheduling now goes through a single schedule() helper that records every ScheduledFuture. Previously only the futures from processRawMessage were tracked; the power updates scheduled by schedulePowerUpdate were not, so they survived undeploy(). The list is synchronized because tasks are scheduled from Undertow request threads and cancelled from the service thread, and completed futures are pruned on each call so the list does not grow for the handler's lifetime. Tests cover: undeploy on a test-support handler is a no-op, undeploy cancels every pending future exactly once, and completed futures are pruned on the next schedule call. --- .../ems/manager/gopacs/GOPACSHandler.java | 86 +++++++++++++------ .../manager/gopacs/GOPACSHandlerTest.groovy | 63 +++++++++++++- 2 files changed, 121 insertions(+), 28 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index ca8618d..b708355 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -141,7 +141,10 @@ public class GOPACSHandler protected final int flexOfferDelaySeconds; protected ObjectMapper objectMapper; - List> scheduledFutureList = new ArrayList<>(); + // Scheduled from Undertow request threads (delayed replies, FlexOffers, power updates) and + // cancelled from the service thread on undeploy, so the list itself must be thread-safe. + private final List> scheduledFutureList = + Collections.synchronizedList(new ArrayList<>()); public static class Factory { protected Container container; @@ -218,14 +221,22 @@ protected GOPACSHandler( this.client = createClient(org.openremote.container.Container.EXECUTOR); - String addressBookUrl = - container.getConfig().getOrDefault(GOPACS_PARTICIPANT_URL, DEFAULT_GOPACS_PARTICIPANT_URL); - String oAuth2Url = - container.getConfig().getOrDefault(GOPACS_OAUTH2_URL, DEFAULT_GOPACS_OAUTH2_URL); - - this.gopacsAddressBookResource = - client.target(addressBookUrl).proxy(GOPACSAddressBookResource.class); - this.gopacsAuthResource = client.target(oAuth2Url).proxy(GOPACSAuthResource.class); + try { + String addressBookUrl = + container + .getConfig() + .getOrDefault(GOPACS_PARTICIPANT_URL, DEFAULT_GOPACS_PARTICIPANT_URL); + String oAuth2Url = + container.getConfig().getOrDefault(GOPACS_OAUTH2_URL, DEFAULT_GOPACS_OAUTH2_URL); + + this.gopacsAddressBookResource = + client.target(addressBookUrl).proxy(GOPACSAddressBookResource.class); + this.gopacsAuthResource = client.target(oAuth2Url).proxy(GOPACSAuthResource.class); + } catch (RuntimeException e) { + // No reference escapes a constructor that threw, so undeploy() can never close this client. + client.close(); + throw e; + } this.gopacsServerResource = new GOPACSServerResourceImpl(this::processRawMessage); this.participantResolutionService = new ParticipantResolutionService(this); @@ -323,12 +334,38 @@ protected void deploy(Container container) { .setCorsAllowedHeaders(CORSConfig.DEFAULT_CORS_ALLOW_ALL)); } + /** + * Schedules a task, tracking its {@link ScheduledFuture} so it can be cancelled on {@link + * #undeploy()}. Already-completed futures are pruned first so the tracked list does not grow + * unbounded across the handler's lifetime. + */ + protected ScheduledFuture schedule(Runnable task, long delayMillis) { + scheduledFutureList.removeIf(ScheduledFuture::isDone); + ScheduledFuture future = + scheduledExecutorService.schedule(task, delayMillis, TimeUnit.MILLISECONDS); + scheduledFutureList.add(future); + return future; + } + + protected int pendingTaskCount() { + return scheduledFutureList.size(); + } + public void undeploy() { - for (ScheduledFuture scheduledFuture : scheduledFutureList) { - scheduledFuture.cancel(true); + synchronized (scheduledFutureList) { + for (ScheduledFuture scheduledFuture : scheduledFutureList) { + scheduledFuture.cancel(true); + } + scheduledFutureList.clear(); + } + if (webService != null) { + webService.undeploy(getDeploymentName(contractedEAN)); + } + if (client != null) { + // createClient(ExecutorService) does not take ownership of the shared Container.EXECUTOR, + // so closing the client here only releases the client's own HTTP resources. + client.close(); } - scheduledFutureList.clear(); - webService.undeploy(getDeploymentName(contractedEAN)); } @Override @@ -613,8 +650,7 @@ protected void schedulePowerUpdate(LocalTime start, String attributeName, double long ispStartMillis = start.atDate(LocalDate.now()).atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); long delay = ispStartMillis - currentTimeMillis; - scheduledExecutorService.schedule( - () -> updatePowerValues(attributeName, power), delay, TimeUnit.MILLISECONDS); + schedule(() -> updatePowerValues(attributeName, power), delay); } protected void updatePowerValues(String attributeName, double power) { @@ -709,10 +745,9 @@ protected void processRawMessage(String transportXml) { payloadMessage.getRecipientDomain(), UftpRoleInformation.getRecipientRoleBySenderRole(sender.role())); // Send delayed so the HTTP 200 on the transport call goes out first, like the accepted path - scheduledExecutorService.schedule( + schedule( () -> notifyNewOutgoingMessage(OutgoingUftpMessage.create(responder, rejection)), - this.responseDelaySeconds, - TimeUnit.SECONDS); + TimeUnit.SECONDS.toMillis(this.responseDelaySeconds)); return; } @@ -722,19 +757,19 @@ protected void processRawMessage(String transportXml) { notifyNewIncomingMessage(incomingUftpMessage); // Send response delayed to ensure HTTP response is sent first - scheduledExecutorService.schedule( + schedule( () -> { uftpReceivedMessageService.process(incomingUftpMessage); }, - this.responseDelaySeconds, - TimeUnit.SECONDS); // 10s delay to ensure HTTP response is sent + TimeUnit.SECONDS.toMillis(this.responseDelaySeconds)); // 10s delay to ensure HTTP + // response is sent // Check if the message is a FlexRequest and schedule sendFlexOffer with delay if (payloadMessage instanceof FlexRequest flexRequest) { UftpParticipant participant = new UftpParticipant(signedMessage); // Schedule FlexOffer to be sent after a short delay to ensure HTTP response is sent first - scheduledExecutorService.schedule( + schedule( () -> { try { sendFlexOffer(participant, flexRequest); @@ -742,10 +777,9 @@ protected void processRawMessage(String transportXml) { LOG.log(Level.SEVERE, "Error sending delayed FlexOffer", e); } }, - this.flexOfferDelaySeconds, - TimeUnit - .SECONDS); // 30s delay to ensure FlexRequestResponse is sent and processed by the - // other party + TimeUnit.SECONDS.toMillis( + this.flexOfferDelaySeconds)); // 30s delay to ensure FlexRequestResponse is sent + // and processed by the other party } } catch (UftpConnectorException e) { LOG.log(Level.SEVERE, "Error processing raw message", e); diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index 6bc3936..f0f1ddc 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -64,6 +64,9 @@ class GOPACSHandlerTest extends Specification { RecordingGOPACSHandler handler String privKeyB64 String pubB64 + // Every ScheduledFuture handed back by the executor stub, in schedule() call order, so tests can + // verify that undeploy() actually cancels the tasks the handler scheduled. + List scheduledFutures def setup() { assetProcessingService = Mock(AssetProcessingService) @@ -71,11 +74,15 @@ class GOPACSHandlerTest extends Specification { timerService = Stub(TimerService) { getCurrentTimeMillis() >> PERIOD.atStartOfDay(ZoneOffset.UTC).toInstant().toEpochMilli() } - // Run every scheduled task inline so processRawMessage is fully synchronous. + scheduledFutures = [] + // Run every scheduled task inline so processRawMessage is fully synchronous, and hand back a + // Mock(ScheduledFuture) (rather than a Stub) so a cancel(true) call on it is observable. executor = Stub(ScheduledExecutorService) { schedule(_ as Runnable, _ as Long, _ as TimeUnit) >> { Runnable r, long d, TimeUnit u -> r.run() - Stub(ScheduledFuture) + def future = Mock(ScheduledFuture) + scheduledFutures << future + future } } @@ -253,6 +260,58 @@ class GOPACSHandlerTest extends Specification { bareHandler.sent[1] instanceof FlexOffer } + def "undeploy is a no-op when the handler was built with the test-support constructor"() { + when: "undeploy is called on a handler with no webService or client (test-support ctor)" + handler.undeploy() + + then: "no exception is thrown" + noExceptionThrown() + } + + def "undeploy cancels every pending scheduled task, and a second undeploy cancels nothing further"() { + given: "a processed FlexRequest that scheduled tasks through the executor stub" + signAndProcess(flexRequestXml(CONTRACTED_EAN)) + def pending = new ArrayList<>(scheduledFutures) + + expect: "at least one task was scheduled" + !pending.isEmpty() + + when: "the handler is undeployed" + handler.undeploy() + + then: "every scheduled future is cancelled" + pending.each { 1 * it.cancel(true) } + + when: "undeploy is called a second time" + handler.undeploy() + + then: "nothing is cancelled again, since the tracked list was already cleared" + pending.each { 0 * it.cancel(true) } + } + + def "a completed scheduled task is pruned from the pending list on the next schedule call"() { + given: "a processed FlexRequest that scheduled at least one pending task" + signAndProcess(flexRequestXml(CONTRACTED_EAN)) + def firstBatch = new ArrayList<>(scheduledFutures) + assert !firstBatch.isEmpty() + assert handler.pendingTaskCount() == firstBatch.size() + + and: "every task from that batch is now already completed" + firstBatch.each { it.isDone() >> true } + + when: "a further message triggers another schedule() call" + signAndProcess(flexOrderXml(CONTRACTED_EAN, [4000, 8000])) + + then: "the completed tasks were pruned, leaving only the newly scheduled ones" + // scheduledFutures only ever grows by appending, so the futures scheduled since firstBatch was + // captured are exactly the tail beyond that snapshot. (Not "scheduledFutures - firstBatch": + // ScheduledFuture extends Comparable, so Groovy's list minus()/== falls back to + // compareTo() rather than equals() for these mocks, and an unstubbed compareTo() always + // answers 0 -- making every mock instance compare "equal" to every other one.) + def secondBatch = scheduledFutures.subList(firstBatch.size(), scheduledFutures.size()) + handler.pendingTaskCount() == secondBatch.size() + } + // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- private static String flexRequestXml(String congestionPoint) { From 632547540c2fb996514ad1c54e437be6dc706853 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:01:16 +0200 Subject: [PATCH 02/16] Deploy the GOPACS endpoint from deploy() instead of the constructor GOPACSHandler deployed its JAX-RS application as the last step of its constructor. The deployed resource routes straight to processRawMessage, so the handler was reachable from Undertow request threads before the constructor had returned, and the Java memory model gives no guarantee about what those threads see of an object whose reference escaped mid-construction. deploy() is now a public no-arg method called after construction, the same shape DistroEnergyHandler and GOPACSRedispatchHandler.startPolling already use. EmsOptimisationService registers the handler in its map before calling deploy(), so a deployment that throws still leaves the handler reachable from stop() and its client gets closed. --- .../ems/manager/EmsOptimisationService.java | 7 +++++-- .../ems/manager/gopacs/GOPACSHandler.java | 14 +++++++++++--- .../manager/gopacs/GOPACSHandlerHttpTest.groovy | 1 + 3 files changed, 17 insertions(+), 5 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index c98e97f..08ccea4 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -318,8 +318,11 @@ private void startGopacsHandler(String contractedEan, String realm, String asset return; } LOG.fine("Deploying GOPACS for EAN: " + contractedEan); - gopacsHandlerMap.put( - contractedEan, gopacsHandlerFactory.createHandler(contractedEan, realm, assetId)); + GOPACSHandler handler = gopacsHandlerFactory.createHandler(contractedEan, realm, assetId); + // Registered before the endpoint deploys, so a handler whose deploy() throws is still reachable + // from stop() and gets its client closed. + gopacsHandlerMap.put(contractedEan, handler); + handler.deploy(); LOG.fine("Deployed GOPACS for EAN: " + contractedEan); } diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index b708355..064b7a1 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -122,6 +122,7 @@ public class GOPACSHandler protected final ScheduledExecutorService scheduledExecutorService; protected final TimerService timerService; protected final WebService webService; + protected final Container container; protected final ResteasyClient client; protected final GOPACSAddressBookResource gopacsAddressBookResource; @@ -161,6 +162,7 @@ public GOPACSHandler createHandler( protected GOPACSHandler( String contractedEAN, String realm, String electricitySupplierAssetId, Container container) { + this.container = container; this.devMode = container.isDevMode(); this.contractedEAN = contractedEAN; this.realm = realm; @@ -252,8 +254,6 @@ protected GOPACSHandler( this.objectMapper = new ObjectMapper(); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); - - deploy(container); } /** @@ -281,6 +281,7 @@ protected GOPACSHandler( this.timerService = timerService; this.scheduledExecutorService = scheduledExecutorService; this.webService = null; + this.container = null; this.gopacsBrokerUrl = ""; this.responseDelaySeconds = 0; @@ -311,7 +312,14 @@ protected static String getDeploymentName(String contractedEAN) { return "GOPACS: " + contractedEAN; } - protected void deploy(Container container) { + /** + * Deploys the JAX-RS endpoint that receives UFTP messages. + * + *

Separate from the constructor so the endpoint cannot hand a request to a partly constructed + * handler: the deployed resource routes straight to {@link #processRawMessage(String)}, which + * would otherwise be reachable from Undertow threads before construction completed. + */ + public void deploy() { LOG.info("Deploying JAX-RS deployment for instance : " + this); List singletons = diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy index e24f9c8..456a02a 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerHttpTest.groovy @@ -156,6 +156,7 @@ class GOPACSHandlerHttpTest extends Specification implements ManagerContainerTra // JAX-RS deployment). EmsOptimisationService is excluded from the service list above so it // cannot spin up a second handler on the same /gopacs deployment path. handler = new GOPACSHandler(CONTRACTED_EAN, MASTER_REALM, ASSET_ID, runningContainer) + handler.deploy() } // Reset cross-test state: the participant cache leaks across features otherwise, and the request From cfc3d07d59c5cf1cb81d2d30d1620bff1b9b2d04 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:01:16 +0200 Subject: [PATCH 03/16] Undeploy GOPACS handlers when the EMS service stops EmsOptimisationService.stop() cancelled the redispatch pollers and the optimisation timers but never touched gopacsHandlerMap, so the GOPACS handlers kept their JAX-RS deployment and REST client across a service stop. Undeploy and clear them alongside the other handler maps. --- .../extension/ems/manager/EmsOptimisationService.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 08ccea4..e47c39c 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -187,6 +187,8 @@ public void start(Container container) throws Exception { @Override public void stop(Container container) throws Exception { + gopacsHandlerMap.forEach((ean, handler) -> handler.undeploy()); + gopacsHandlerMap.clear(); gopacsRedispatchHandlerMap.forEach((ean, handler) -> handler.stopPolling()); gopacsRedispatchHandlerMap.clear(); energyOptimisationAssetsMap.forEach((assetId, scheduledFuture) -> stopOptimisation(assetId)); From a6a5487fbc0405bd2f0469979ad7b7368b8502e4 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:14:01 +0200 Subject: [PATCH 04/16] Add a lifecycle test for GOPACS handlers in EmsOptimisationService Covers processAssetChange for EmsGOPACSAsset CREATE, UPDATE and DELETE and the new stop() teardown, through a recording GOPACSHandler subclass built by the real constructor (fake Container with OAuth config and a temporary private-key file) whose deploy()/undeploy() only count calls. Same pattern as EmsOptimisationServiceDistroEnergyTest; no mocking of the concrete class and no extra test dependency. --- .../EmsOptimisationServiceGopacsTest.groovy | 182 ++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy new file mode 100644 index 0000000..11508a4 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -0,0 +1,182 @@ +/* + * Copyright 2026, OpenRemote Inc. + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as + * published by the Free Software Foundation, either version 3 of the + * License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +package org.openremote.extension.ems.manager + +import org.openremote.container.timer.TimerService +import org.openremote.container.web.WebService +import org.openremote.extension.ems.agent.EmsGOPACSAsset +import org.openremote.extension.ems.manager.gopacs.GOPACSHandler +import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.datapoint.AssetPredictedDatapointService +import org.openremote.model.Container +import org.openremote.model.PersistenceEvent +import org.openremote.model.util.ValueUtil +import spock.lang.Shared +import spock.lang.Specification + +import java.util.concurrent.ScheduledExecutorService + +/** + * Exercises {@link EmsOptimisationService#processAssetChange} and {@link EmsOptimisationService#stop} + * for {@link EmsGOPACSAsset} persistence events, via a recording {@code GOPACSHandler} subclass + * rather than a mock of the class. The real constructor runs, so the fake {@code Container} carries + * the OAuth client config and a readable private-key file; {@code deploy()}/{@code undeploy()} are + * overridden so no JAX-RS deployment happens and no client is closed. + * + * {@code gopacsHandlerMap} is keyed by contracted EAN. These tests catch stop calls that use a key + * the running handler was never registered under: {@code Map.remove} with the wrong key is a silent + * no-op, so the old handler keeps its endpoint and client alive next to its replacement. + */ +class EmsOptimisationServiceGopacsTest extends Specification { + + static final String ASSET_ID = "gopacsAsset1" + static final String OTHER_ASSET_ID = "gopacsAsset2" + static final String REALM = "master" + static final String EAN = "ean.265987182507322951" + static final String OTHER_EAN = "ean.265987182507322952" + + @Shared File privateKeyFile + List createdHandlers + EmsOptimisationService service + + def setupSpec() { + // Populates the asset model registry from this extension's own AssetModelProvider SPI + // registration, which real asset construction needs; a plain Specification has no container + // to do this at startup. + ValueUtil.initialise(null) + // The constructor only checks the file is readable and reads it into a field; the key is never + // used because the recording subclass never sends anything. + privateKeyFile = File.createTempFile("gopacs-test-key", ".txt") + privateKeyFile.text = "not-a-real-key" + } + + def cleanupSpec() { + privateKeyFile.delete() + } + + def setup() { + def handlerContainer = Stub(Container) { + getConfig() >> [ + (GOPACSHandler.GOPACS_CLIENT_ID): "client-id", + (GOPACSHandler.GOPACS_CLIENT_SECRET): "client-secret", + (GOPACSHandler.GOPACS_PRIVATE_KEY_FILE): privateKeyFile.absolutePath, + ] + getService(AssetProcessingService) >> Stub(AssetProcessingService) + getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) + getService(TimerService) >> Stub(TimerService) + getService(WebService) >> Stub(WebService) + getScheduledExecutor() >> Stub(ScheduledExecutorService) + } + + createdHandlers = [] + + service = new EmsOptimisationService() + service.gopacsHandlerFactory = new GOPACSHandler.Factory(handlerContainer) { + @Override + GOPACSHandler createHandler(String contractedEan, String realm, String assetId) { + def handler = + new RecordingGOPACSHandler(contractedEan, realm, assetId, handlerContainer) + createdHandlers << handler + return handler + } + } + } + + private static EmsGOPACSAsset gopacsAsset(String ean = EAN, String assetId = ASSET_ID) { + def asset = new EmsGOPACSAsset("gopacs").setId(assetId).setRealm(REALM) + asset.getAttributes().getOrCreate(EmsGOPACSAsset.CONTRACTED_EAN).setValue(ean) + return asset + } + + private static PersistenceEvent event( + PersistenceEvent.Cause cause, EmsGOPACSAsset asset) { + return new PersistenceEvent<>(cause, asset, null, null, null) + } + + def "CREATE deploys a handler for the asset's EAN"() { + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) + + then: + createdHandlers.size() == 1 + createdHandlers[0].contractedEAN == EAN + createdHandlers[0].deployCount == 1 + } + + def "DELETE undeploys the handler"() { + given: "a deployed handler" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) + def original = createdHandlers[0] + + when: + service.processAssetChange(event(PersistenceEvent.Cause.DELETE, gopacsAsset())) + + then: + original.undeployCount == 1 + } + + def "UPDATE with the same EAN undeploys the old handler before deploying its replacement"() { + given: "a deployed handler" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) + def original = createdHandlers[0] + + when: + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset())) + + then: "the old handler is stopped and exactly one new handler replaces it" + original.undeployCount == 1 + createdHandlers.size() == 2 + createdHandlers[1].deployCount == 1 + } + + def "stop undeploys every deployed handler"() { + given: "handlers for two assets" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID))) + service.processAssetChange( + event(PersistenceEvent.Cause.CREATE, gopacsAsset(OTHER_EAN, OTHER_ASSET_ID))) + + when: + service.stop(null) + + then: + createdHandlers.size() == 2 + createdHandlers.every { it.undeployCount == 1 } + } + + // Records deploy()/undeploy() calls instead of deploying the JAX-RS endpoint or closing the client. + static class RecordingGOPACSHandler extends GOPACSHandler { + int deployCount = 0 + int undeployCount = 0 + + RecordingGOPACSHandler( + String contractedEan, String realm, String assetId, Container container) { + super(contractedEan, realm, assetId, container) + } + + @Override + void deploy() { + deployCount++ + } + + @Override + void undeploy() { + undeployCount++ + } + } +} From 76d3d78bb0f0ff12e3626d519f6d74645ad1f7ac Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:14:55 +0200 Subject: [PATCH 05/16] Stop the GOPACS handler by asset id when the asset is updated processAssetChange stopped the handler on UPDATE with the EAN from the updated entity, but gopacsHandlerMap is keyed by EAN and the entity carries the new value. When the contracted EAN was edited through the asset editor the stop was a silent no-op, so the handler registered under the old EAN kept its JAX-RS deployment and REST client alive next to the new one. The attribute-event path already handles this case with the old value, but an asset merge does not emit attribute events. Undeploy every handler deployed for the asset id instead, whichever EAN it is registered under. GOPACSHandler exposes the asset id it was deployed for; the field name (electricitySupplierAssetId) predates the EmsGOPACSAsset type and is left alone. The lifecycle test gains the changed-EAN case, which failed before this change. --- .../ems/manager/EmsOptimisationService.java | 20 ++++++++++++++++++- .../ems/manager/gopacs/GOPACSHandler.java | 5 +++++ .../EmsOptimisationServiceGopacsTest.groovy | 15 ++++++++++++++ 3 files changed, 39 insertions(+), 1 deletion(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index e47c39c..d86d5af 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -336,6 +336,22 @@ private void stopGopacsHandler(String contractedEan) { } } + /** + * Undeploys every GOPACS handler deployed for the asset, whichever EAN it is registered under. + */ + private void stopGopacsHandlersForAsset(String assetId) { + gopacsHandlerMap + .entrySet() + .removeIf( + entry -> { + if (!assetId.equals(entry.getValue().getAssetId())) { + return false; + } + entry.getValue().undeploy(); + return true; + }); + } + private void startRedispatchHandler(String contractedEan, String realm, String assetId) { if (contractedEan.isBlank()) { LOG.warning("Unable to start redispatch handler because EAN is blank"); @@ -381,7 +397,9 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { } } if (persistenceEvent.getCause() == PersistenceEvent.Cause.UPDATE) { - stopGopacsHandler(contractedEan); + // The entity carries the new EAN, so stopping by EAN would miss a handler + // registered under the previous one and leave it running next to the new one. + stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); startGopacsHandler( contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); // Redispatch handler is managed via attribute events (redispatchEnabled) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index 064b7a1..62da28f 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -308,6 +308,11 @@ protected GOPACSHandler( this.objectMapper = new ObjectMapper(); } + /** Id of the {@link EmsGOPACSAsset} this handler was deployed for. */ + public String getAssetId() { + return electricitySupplierAssetId; + } + protected static String getDeploymentName(String contractedEAN) { return "GOPACS: " + contractedEAN; } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index 11508a4..7054f4e 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -145,6 +145,21 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdHandlers[1].deployCount == 1 } + def "UPDATE with a changed EAN undeploys the handler registered under the old EAN"() { + given: "a deployed handler for the original EAN" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN))) + def original = createdHandlers[0] + + when: "the asset is saved with a different EAN" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(OTHER_EAN))) + + then: "the old handler is stopped rather than left running under the old key" + original.undeployCount == 1 + createdHandlers.size() == 2 + createdHandlers[1].contractedEAN == OTHER_EAN + createdHandlers[1].deployCount == 1 + } + def "stop undeploys every deployed handler"() { given: "handlers for two assets" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID))) From f8d3cfcc5a55730edf9e321f3a2d0b8356d1b5ed Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:19:09 +0200 Subject: [PATCH 06/16] Re-key the redispatch poller when the contracted EAN is updated processAssetChange left the redispatch poller untouched on UPDATE, on the grounds that it is managed through attribute events. An asset merge emits none, so editing the contracted EAN through the asset editor left the poller running under the old EAN: it kept polling and resolving EAN effectivity for a congestion point the asset no longer has, and there was no key it could be stopped by afterwards. Unlike the GOPACS handler, the poller is not simply restarted on every save. Its announcement bookkeeping (lastProcessedAnnouncementId and the set of recorded announcement ids) is in memory only and history entries are not deduplicated against the asset, so a restart re-records every open announcement. UPDATE therefore only stops a poller registered for the asset under an EAN other than the current one, and starts a new one under the current EAN if redispatch is still enabled. A poller already running under the current EAN is left alone. GOPACSRedispatchHandler exposes the asset id it was started for. The lifecycle test gains CREATE, DELETE, same-EAN UPDATE, changed-EAN UPDATE (enabled and disabled) and stop() cases for the poller; the two changed-EAN cases failed before this change. --- .../ems/manager/EmsOptimisationService.java | 28 ++++- .../gopacs/GOPACSRedispatchHandler.java | 5 + .../EmsOptimisationServiceGopacsTest.groovy | 118 +++++++++++++++++- 3 files changed, 146 insertions(+), 5 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index d86d5af..bdf1c5e 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -373,6 +373,24 @@ private void stopRedispatchHandler(String contractedEan) { } } + /** + * Stops every redispatch poller for the asset that is registered under an EAN other than the + * current one, and reports whether any was stopped. + */ + private boolean stopStaleRedispatchHandlersForAsset(String assetId, String currentEan) { + return gopacsRedispatchHandlerMap + .entrySet() + .removeIf( + entry -> { + if (currentEan.equals(entry.getKey()) + || !assetId.equals(entry.getValue().getAssetId())) { + return false; + } + entry.getValue().stopPolling(); + return true; + }); + } + protected void processAssetChange(PersistenceEvent persistenceEvent) { if (persistenceEvent.getEntity() instanceof EmsEnergyOptimisationAsset emsEnergyOptimisationAsset) { @@ -402,7 +420,15 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); startGopacsHandler( contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - // Redispatch handler is managed via attribute events (redispatchEnabled) + // Redispatch polling is otherwise managed via attribute events. A poller + // already running under this EAN is left alone, since its announcement + // bookkeeping is in memory and a restart would re-record history; only one + // stranded under a previous EAN is stopped and, if still enabled, restarted. + if (stopStaleRedispatchHandlersForAsset(emsGOPACSAsset.getId(), contractedEan) + && emsGOPACSAsset.getRedispatchEnabled().orElse(false)) { + startRedispatchHandler( + contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); + } } }); } diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java index 96c300d..281c28a 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java @@ -177,6 +177,11 @@ protected GOPACSRedispatchHandler( + " min)"); } + /** Id of the {@link EmsGOPACSAsset} this poller was started for. */ + public String getAssetId() { + return assetId; + } + public void startPolling() { if (apiKey == null || apiKey.isBlank()) { LOG.severe( diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index 7054f4e..ce12d09 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -22,7 +22,9 @@ import org.openremote.container.timer.TimerService import org.openremote.container.web.WebService import org.openremote.extension.ems.agent.EmsGOPACSAsset import org.openremote.extension.ems.manager.gopacs.GOPACSHandler +import org.openremote.extension.ems.manager.gopacs.GOPACSRedispatchHandler import org.openremote.manager.asset.AssetProcessingService +import org.openremote.manager.asset.AssetStorageService import org.openremote.manager.datapoint.AssetPredictedDatapointService import org.openremote.model.Container import org.openremote.model.PersistenceEvent @@ -37,7 +39,8 @@ import java.util.concurrent.ScheduledExecutorService * for {@link EmsGOPACSAsset} persistence events, via a recording {@code GOPACSHandler} subclass * rather than a mock of the class. The real constructor runs, so the fake {@code Container} carries * the OAuth client config and a readable private-key file; {@code deploy()}/{@code undeploy()} are - * overridden so no JAX-RS deployment happens and no client is closed. + * overridden so no JAX-RS deployment happens and no client is closed. The redispatch poller gets + * the same treatment with {@code startPolling()}/{@code stopPolling()}. * * {@code gopacsHandlerMap} is keyed by contracted EAN. These tests catch stop calls that use a key * the running handler was never registered under: {@code Map.remove} with the wrong key is a silent @@ -53,6 +56,7 @@ class EmsOptimisationServiceGopacsTest extends Specification { @Shared File privateKeyFile List createdHandlers + List createdRedispatchHandlers EmsOptimisationService service def setupSpec() { @@ -78,6 +82,7 @@ class EmsOptimisationServiceGopacsTest extends Specification { (GOPACSHandler.GOPACS_PRIVATE_KEY_FILE): privateKeyFile.absolutePath, ] getService(AssetProcessingService) >> Stub(AssetProcessingService) + getService(AssetStorageService) >> Stub(AssetStorageService) getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) getService(TimerService) >> Stub(TimerService) getService(WebService) >> Stub(WebService) @@ -85,6 +90,7 @@ class EmsOptimisationServiceGopacsTest extends Specification { } createdHandlers = [] + createdRedispatchHandlers = [] service = new EmsOptimisationService() service.gopacsHandlerFactory = new GOPACSHandler.Factory(handlerContainer) { @@ -96,11 +102,24 @@ class EmsOptimisationServiceGopacsTest extends Specification { return handler } } + service.gopacsRedispatchHandlerFactory = + new GOPACSRedispatchHandler.Factory(handlerContainer) { + @Override + GOPACSRedispatchHandler createHandler( + String contractedEan, String realm, String assetId) { + def handler = + new RecordingRedispatchHandler(contractedEan, realm, assetId, handlerContainer) + createdRedispatchHandlers << handler + return handler + } + } } - private static EmsGOPACSAsset gopacsAsset(String ean = EAN, String assetId = ASSET_ID) { + private static EmsGOPACSAsset gopacsAsset( + String ean = EAN, String assetId = ASSET_ID, boolean redispatchEnabled = false) { def asset = new EmsGOPACSAsset("gopacs").setId(assetId).setRealm(REALM) asset.getAttributes().getOrCreate(EmsGOPACSAsset.CONTRACTED_EAN).setValue(ean) + asset.setRedispatchEnabled(redispatchEnabled) return asset } @@ -117,6 +136,19 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdHandlers.size() == 1 createdHandlers[0].contractedEAN == EAN createdHandlers[0].deployCount == 1 + + and: "no redispatch poller, since redispatch is not enabled on the asset" + createdRedispatchHandlers.isEmpty() + } + + def "CREATE with redispatch enabled also starts a redispatch poller for the EAN"() { + when: + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + + then: + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].contractedEAN == EAN + createdRedispatchHandlers[0].startCount == 1 } def "DELETE undeploys the handler"() { @@ -131,6 +163,18 @@ class EmsOptimisationServiceGopacsTest extends Specification { original.undeployCount == 1 } + def "DELETE stops the redispatch poller"() { + given: "a running poller" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: + service.processAssetChange(event(PersistenceEvent.Cause.DELETE, gopacsAsset(EAN, ASSET_ID, true))) + + then: + original.stopCount == 1 + } + def "UPDATE with the same EAN undeploys the old handler before deploying its replacement"() { given: "a deployed handler" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) @@ -160,9 +204,52 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdHandlers[1].deployCount == 1 } + def "UPDATE with the same EAN leaves the redispatch poller running"() { + given: "a running poller" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved without changing the EAN" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, true))) + + then: "the poller keeps its in-memory announcement bookkeeping rather than being restarted" + original.stopCount == 0 + createdRedispatchHandlers.size() == 1 + } + + def "UPDATE with a changed EAN re-keys the redispatch poller"() { + given: "a running poller for the original EAN" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved with a different EAN" + service.processAssetChange( + event(PersistenceEvent.Cause.UPDATE, gopacsAsset(OTHER_EAN, ASSET_ID, true))) + + then: "the stale poller is stopped and a new one polls under the new EAN" + original.stopCount == 1 + createdRedispatchHandlers.size() == 2 + createdRedispatchHandlers[1].contractedEAN == OTHER_EAN + createdRedispatchHandlers[1].startCount == 1 + } + + def "UPDATE with a changed EAN and redispatch disabled stops the stale poller without starting one"() { + given: "a running poller for the original EAN" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved with a different EAN and redispatch switched off" + service.processAssetChange( + event(PersistenceEvent.Cause.UPDATE, gopacsAsset(OTHER_EAN, ASSET_ID, false))) + + then: + original.stopCount == 1 + createdRedispatchHandlers.size() == 1 + } + def "stop undeploys every deployed handler"() { - given: "handlers for two assets" - service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID))) + given: "handlers for two assets, one of them polling redispatch" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) service.processAssetChange( event(PersistenceEvent.Cause.CREATE, gopacsAsset(OTHER_EAN, OTHER_ASSET_ID))) @@ -172,6 +259,8 @@ class EmsOptimisationServiceGopacsTest extends Specification { then: createdHandlers.size() == 2 createdHandlers.every { it.undeployCount == 1 } + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].stopCount == 1 } // Records deploy()/undeploy() calls instead of deploying the JAX-RS endpoint or closing the client. @@ -194,4 +283,25 @@ class EmsOptimisationServiceGopacsTest extends Specification { undeployCount++ } } + + // Records startPolling()/stopPolling() calls instead of scheduling polls or closing the client. + static class RecordingRedispatchHandler extends GOPACSRedispatchHandler { + int startCount = 0 + int stopCount = 0 + + RecordingRedispatchHandler( + String contractedEan, String realm, String assetId, Container container) { + super(contractedEan, realm, assetId, container) + } + + @Override + void startPolling() { + startCount++ + } + + @Override + void stopPolling() { + stopCount++ + } + } } From 24a4b8e0a704c7d285b4a1fe83b6f94a68b77c33 Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:21:41 +0200 Subject: [PATCH 07/16] Apply redispatchEnabled from an asset save to the redispatch poller Toggling redispatchEnabled through the asset editor had no effect until an attribute event or a service restart: processAssetChange only acted on the poller when the EAN changed, and an asset merge emits no attribute events. The UPDATE branch now reconciles the poller with the saved asset state: exactly one poller under the current EAN when redispatch is enabled, none otherwise. A poller already running under the current EAN is still left alone, for the same reason as before: its announcement bookkeeping is in memory only and a restart would re-record history. Replaces the stale-EAN-only helper from the previous commit. The lifecycle test gains the enable-on-save and disable-on-save cases, which failed before this change. --- .../ems/manager/EmsOptimisationService.java | 32 ++++++++++--------- .../EmsOptimisationServiceGopacsTest.groovy | 26 +++++++++++++++ 2 files changed, 43 insertions(+), 15 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index bdf1c5e..3bb27fa 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -374,21 +374,29 @@ private void stopRedispatchHandler(String contractedEan) { } /** - * Stops every redispatch poller for the asset that is registered under an EAN other than the - * current one, and reports whether any was stopped. + * Brings the redispatch poller for the asset in line with its saved state: exactly one poller + * under the current EAN when redispatch is enabled, none otherwise. A poller already running + * under the current EAN is left alone, since its announcement bookkeeping is in memory only and a + * restart would re-record every open announcement in the history. */ - private boolean stopStaleRedispatchHandlersForAsset(String assetId, String currentEan) { - return gopacsRedispatchHandlerMap + private void reconcileRedispatchHandler(EmsGOPACSAsset asset, String contractedEan) { + String assetId = asset.getId(); + boolean enabled = asset.getRedispatchEnabled().orElse(false); + gopacsRedispatchHandlerMap .entrySet() .removeIf( entry -> { - if (currentEan.equals(entry.getKey()) - || !assetId.equals(entry.getValue().getAssetId())) { + if (!assetId.equals(entry.getValue().getAssetId()) + || (enabled && contractedEan.equals(entry.getKey()))) { return false; } entry.getValue().stopPolling(); return true; }); + GOPACSRedispatchHandler current = gopacsRedispatchHandlerMap.get(contractedEan); + if (enabled && (current == null || !assetId.equals(current.getAssetId()))) { + startRedispatchHandler(contractedEan, asset.getRealm(), assetId); + } } protected void processAssetChange(PersistenceEvent persistenceEvent) { @@ -420,15 +428,9 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); startGopacsHandler( contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - // Redispatch polling is otherwise managed via attribute events. A poller - // already running under this EAN is left alone, since its announcement - // bookkeeping is in memory and a restart would re-record history; only one - // stranded under a previous EAN is stopped and, if still enabled, restarted. - if (stopStaleRedispatchHandlersForAsset(emsGOPACSAsset.getId(), contractedEan) - && emsGOPACSAsset.getRedispatchEnabled().orElse(false)) { - startRedispatchHandler( - contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - } + // Redispatch polling is also managed via attribute events, but an asset merge + // emits none, so the saved EAN and redispatchEnabled are applied here too. + reconcileRedispatchHandler(emsGOPACSAsset, contractedEan); } }); } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index ce12d09..a3331da 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -217,6 +217,32 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdRedispatchHandlers.size() == 1 } + def "UPDATE that enables redispatch on save starts a poller"() { + given: "an asset without redispatch" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, false))) + + when: "the asset is saved with redispatch switched on" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, true))) + + then: + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].contractedEAN == EAN + createdRedispatchHandlers[0].startCount == 1 + } + + def "UPDATE that disables redispatch on save stops the poller"() { + given: "a running poller" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved with redispatch switched off" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, false))) + + then: + original.stopCount == 1 + createdRedispatchHandlers.size() == 1 + } + def "UPDATE with a changed EAN re-keys the redispatch poller"() { given: "a running poller for the original EAN" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) From 0c8a51f3f25aa7ad7bd25a72d0c89da301df4a8e Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:31:47 +0200 Subject: [PATCH 08/16] Stop scheduling GOPACS tasks once undeploy has started A request already inside processRawMessage could schedule work after undeploy() had cancelled and cleared the tracked futures. That task outlived the handler and later ran against a closed client. The endpoint is now undeployed first, so no further request can reach processRawMessage, and schedule() refuses under the same monitor the cancel loop holds. Either a task is tracked and then cancelled, or it is never scheduled at all. Also drops the trailing comments that the formatter had split across the schedule() calls and the statements after them. They named fixed 10s and 30s delays while both values come from configuration. --- .../ems/manager/gopacs/GOPACSHandler.java | 45 ++++++++++++------- .../manager/gopacs/GOPACSHandlerTest.groovy | 24 ++++++++++ 2 files changed, 53 insertions(+), 16 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index 62da28f..ace295d 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -147,6 +147,11 @@ public class GOPACSHandler private final List> scheduledFutureList = Collections.synchronizedList(new ArrayList<>()); + // Guarded by the scheduledFutureList monitor, which is what makes the check in schedule() and + // the cancel loop in undeploy() exclusive: either the task is tracked and then cancelled, or it + // is never scheduled at all. + private boolean undeployed; + public static class Factory { protected Container container; @@ -351,13 +356,21 @@ public void deploy() { * Schedules a task, tracking its {@link ScheduledFuture} so it can be cancelled on {@link * #undeploy()}. Already-completed futures are pruned first so the tracked list does not grow * unbounded across the handler's lifetime. + * + *

Nothing is scheduled once {@link #undeploy()} has started. A request that was already inside + * {@link #processRawMessage(String)} at that point would otherwise queue work that outlives the + * handler and later runs against a closed client. */ - protected ScheduledFuture schedule(Runnable task, long delayMillis) { - scheduledFutureList.removeIf(ScheduledFuture::isDone); - ScheduledFuture future = - scheduledExecutorService.schedule(task, delayMillis, TimeUnit.MILLISECONDS); - scheduledFutureList.add(future); - return future; + protected void schedule(Runnable task, long delayMillis) { + synchronized (scheduledFutureList) { + if (undeployed) { + LOG.fine("Handler is undeployed, dropping scheduled task for EAN: " + contractedEAN); + return; + } + scheduledFutureList.removeIf(ScheduledFuture::isDone); + scheduledFutureList.add( + scheduledExecutorService.schedule(task, delayMillis, TimeUnit.MILLISECONDS)); + } } protected int pendingTaskCount() { @@ -365,15 +378,18 @@ protected int pendingTaskCount() { } public void undeploy() { + // Undeploy the endpoint first so no further request can reach processRawMessage and start work + // that the cancel loop below would then have to race. + if (webService != null) { + webService.undeploy(getDeploymentName(contractedEAN)); + } synchronized (scheduledFutureList) { + undeployed = true; for (ScheduledFuture scheduledFuture : scheduledFutureList) { scheduledFuture.cancel(true); } scheduledFutureList.clear(); } - if (webService != null) { - webService.undeploy(getDeploymentName(contractedEAN)); - } if (client != null) { // createClient(ExecutorService) does not take ownership of the shared Container.EXECUTOR, // so closing the client here only releases the client's own HTTP resources. @@ -769,19 +785,18 @@ protected void processRawMessage(String transportXml) { new UftpParticipant(signedMessage), payloadMessage, transportXml, payloadXml); notifyNewIncomingMessage(incomingUftpMessage); - // Send response delayed to ensure HTTP response is sent first + // Delayed so the HTTP response on the transport call goes out first schedule( () -> { uftpReceivedMessageService.process(incomingUftpMessage); }, - TimeUnit.SECONDS.toMillis(this.responseDelaySeconds)); // 10s delay to ensure HTTP - // response is sent + TimeUnit.SECONDS.toMillis(this.responseDelaySeconds)); // Check if the message is a FlexRequest and schedule sendFlexOffer with delay if (payloadMessage instanceof FlexRequest flexRequest) { UftpParticipant participant = new UftpParticipant(signedMessage); - // Schedule FlexOffer to be sent after a short delay to ensure HTTP response is sent first + // Delayed so the FlexRequestResponse is sent and processed by the other party first schedule( () -> { try { @@ -790,9 +805,7 @@ protected void processRawMessage(String transportXml) { LOG.log(Level.SEVERE, "Error sending delayed FlexOffer", e); } }, - TimeUnit.SECONDS.toMillis( - this.flexOfferDelaySeconds)); // 30s delay to ensure FlexRequestResponse is sent - // and processed by the other party + TimeUnit.SECONDS.toMillis(this.flexOfferDelaySeconds)); } } catch (UftpConnectorException e) { LOG.log(Level.SEVERE, "Error processing raw message", e); diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index f0f1ddc..884adf7 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -312,6 +312,30 @@ class GOPACSHandlerTest extends Specification { handler.pendingTaskCount() == secondBatch.size() } + def "a request still in flight when undeploy starts cannot schedule further work"() { + given: "a handler that has been undeployed" + handler.undeploy() + + when: "a request that was already being processed reaches a schedule() call" + handler.schedule({} as Runnable, 0L) + + then: "nothing is handed to the executor, so no task outlives the handler" + scheduledFutures.isEmpty() + handler.pendingTaskCount() == 0 + } + + def "a message processed after undeploy schedules nothing"() { + given: "a handler that has been undeployed" + handler.undeploy() + + when: "a FlexRequest is processed on a request thread that was already in flight" + signAndProcess(flexRequestXml(CONTRACTED_EAN)) + + then: "neither the delayed response nor the delayed FlexOffer is scheduled" + scheduledFutures.isEmpty() + handler.pendingTaskCount() == 0 + } + // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- private static String flexRequestXml(String congestionPoint) { From 696aae005cc147baa2e8a6fd3a0412f803051d1c Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:32:17 +0200 Subject: [PATCH 09/16] Use concurrent maps for the GOPACS handler registries Both registries are touched from the persistence route, the attribute event subscription and the container stop thread. Stopping handlers by asset id and reconciling the redispatch poller iterate them with entrySet().removeIf while stop() walks them with forEach, so a concurrent put or remove on a HashMap could throw ConcurrentModificationException or corrupt the table. --- .../extension/ems/manager/EmsOptimisationService.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 3bb27fa..b6aadd5 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -72,8 +72,11 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer private final Map> energyOptimisationAssetsMap = new ConcurrentHashMap<>(); private final Map energyOptimisationTimersMap = new HashMap<>(); - private final Map gopacsHandlerMap = new HashMap<>(); - private final Map gopacsRedispatchHandlerMap = new HashMap<>(); + // Read and written from the persistence route, the attribute event subscription and the container + // stop thread, and iterated while entries are removed, so a plain HashMap is not enough. + private final Map gopacsHandlerMap = new ConcurrentHashMap<>(); + private final Map gopacsRedispatchHandlerMap = + new ConcurrentHashMap<>(); @SuppressWarnings("unchecked") @Override From 6fa9b26e629654ea898d3685fdc82544cb2cfdec Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:33:26 +0200 Subject: [PATCH 10/16] Undeploy the GOPACS handler that a replacement displaces Both registries are keyed by contracted EAN, and registering a new handler put straight over any entry already under that key. The displaced handler kept its JAX-RS endpoint and REST client open with no key left to stop it by, so two assets sharing an EAN leaked exactly what this branch set out to stop leaking. Stopping the current entry first covers every caller, including the redispatch reconcile, which could reach this case for a poller owned by another asset. --- .../ems/manager/EmsOptimisationService.java | 6 ++++ .../EmsOptimisationServiceGopacsTest.groovy | 31 +++++++++++++++++++ 2 files changed, 37 insertions(+) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index b6aadd5..ecc05f3 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -323,6 +323,9 @@ private void startGopacsHandler(String contractedEan, String realm, String asset return; } LOG.fine("Deploying GOPACS for EAN: " + contractedEan); + // Registering over a live handler would leave the displaced one holding its endpoint and + // client with no key left to stop it by, which is what happens when two assets share an EAN. + stopGopacsHandler(contractedEan); GOPACSHandler handler = gopacsHandlerFactory.createHandler(contractedEan, realm, assetId); // Registered before the endpoint deploys, so a handler whose deploy() throws is still reachable // from stop() and gets its client closed. @@ -361,6 +364,9 @@ private void startRedispatchHandler(String contractedEan, String realm, String a return; } LOG.fine("Starting redispatch handler for EAN: " + contractedEan); + // Same reasoning as startGopacsHandler: the displaced poller would keep polling and never + // close its client. + stopRedispatchHandler(contractedEan); GOPACSRedispatchHandler handler = gopacsRedispatchHandlerFactory.createHandler(contractedEan, realm, assetId); gopacsRedispatchHandlerMap.put(contractedEan, handler); diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index a3331da..bcb481c 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -273,6 +273,37 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdRedispatchHandlers.size() == 1 } + def "a handler deployed for an EAN already in use undeploys the one it displaces"() { + given: "a deployed handler for the EAN" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID))) + def original = createdHandlers[0] + + when: "a second asset is created with the same contracted EAN" + service.processAssetChange( + event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, OTHER_ASSET_ID))) + + then: "the displaced handler is undeployed instead of being left with no key to stop it by" + original.undeployCount == 1 + createdHandlers.size() == 2 + createdHandlers[1].assetId == OTHER_ASSET_ID + } + + def "a redispatch poller started for an EAN already in use stops the one it displaces"() { + given: "a running poller for the EAN" + service.processAssetChange( + event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "a second asset is created with the same contracted EAN and redispatch enabled" + service.processAssetChange( + event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, OTHER_ASSET_ID, true))) + + then: "the displaced poller is stopped and its client closed" + original.stopCount == 1 + createdRedispatchHandlers.size() == 2 + createdRedispatchHandlers[1].assetId == OTHER_ASSET_ID + } + def "stop undeploys every deployed handler"() { given: "handlers for two assets, one of them polling redispatch" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) From 475253f986a54905819644bb674b405da776174b Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:34:28 +0200 Subject: [PATCH 11/16] Stop GOPACS handlers by asset id when the asset is deleted DELETE stopped by the EAN on the deleted entity while UPDATE stops by asset id. The entity carries the EAN as saved, which is not necessarily the key the running handler is registered under, so the same silent no-op that UPDATE had applied here. The asset-scoped stop for the poller is now a helper shared with the redispatch reconcile, which was doing the same removeIf inline. --- .../ems/manager/EmsOptimisationService.java | 30 ++++++++++++------- .../EmsOptimisationServiceGopacsTest.groovy | 26 ++++++++++++++++ 2 files changed, 46 insertions(+), 10 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index ecc05f3..bb1608e 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -383,25 +383,33 @@ private void stopRedispatchHandler(String contractedEan) { } /** - * Brings the redispatch poller for the asset in line with its saved state: exactly one poller - * under the current EAN when redispatch is enabled, none otherwise. A poller already running - * under the current EAN is left alone, since its announcement bookkeeping is in memory only and a - * restart would re-record every open announcement in the history. + * Stops every redispatch poller started for the asset, whichever EAN it is registered under, + * except one already polling under {@code keepEan}. Pass {@code null} to stop them all. */ - private void reconcileRedispatchHandler(EmsGOPACSAsset asset, String contractedEan) { - String assetId = asset.getId(); - boolean enabled = asset.getRedispatchEnabled().orElse(false); + private void stopRedispatchHandlersForAsset(String assetId, String keepEan) { gopacsRedispatchHandlerMap .entrySet() .removeIf( entry -> { if (!assetId.equals(entry.getValue().getAssetId()) - || (enabled && contractedEan.equals(entry.getKey()))) { + || entry.getKey().equals(keepEan)) { return false; } entry.getValue().stopPolling(); return true; }); + } + + /** + * Brings the redispatch poller for the asset in line with its saved state: exactly one poller + * under the current EAN when redispatch is enabled, none otherwise. A poller already running + * under the current EAN is left alone, since its announcement bookkeeping is in memory only and a + * restart would re-record every open announcement in the history. + */ + private void reconcileRedispatchHandler(EmsGOPACSAsset asset, String contractedEan) { + String assetId = asset.getId(); + boolean enabled = asset.getRedispatchEnabled().orElse(false); + stopRedispatchHandlersForAsset(assetId, enabled ? contractedEan : null); GOPACSRedispatchHandler current = gopacsRedispatchHandlerMap.get(contractedEan); if (enabled && (current == null || !assetId.equals(current.getAssetId()))) { startRedispatchHandler(contractedEan, asset.getRealm(), assetId); @@ -420,8 +428,10 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { .ifPresent( contractedEan -> { if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { - stopGopacsHandler(contractedEan); - stopRedispatchHandler(contractedEan); + // By asset id for the same reason as UPDATE below: the entity carries the EAN as + // saved, which is not necessarily the key the handler is registered under. + stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); + stopRedispatchHandlersForAsset(emsGOPACSAsset.getId(), null); } if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { startGopacsHandler( diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index bcb481c..4fe8d6d 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -175,6 +175,32 @@ class EmsOptimisationServiceGopacsTest extends Specification { original.stopCount == 1 } + def "DELETE undeploys the handler registered under the EAN it was created with"() { + given: "a handler deployed for the original EAN" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN))) + def original = createdHandlers[0] + + when: "the asset is deleted carrying a different EAN than the one it was registered under" + service.processAssetChange(event(PersistenceEvent.Cause.DELETE, gopacsAsset(OTHER_EAN))) + + then: "the handler is stopped by asset id rather than missed by key" + original.undeployCount == 1 + } + + def "DELETE stops the redispatch poller registered under the EAN it was started with"() { + given: "a poller running for the original EAN" + service.processAssetChange( + event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true))) + def original = createdRedispatchHandlers[0] + + when: "the asset is deleted carrying a different EAN than the one it was registered under" + service.processAssetChange( + event(PersistenceEvent.Cause.DELETE, gopacsAsset(OTHER_EAN, ASSET_ID, true))) + + then: "the poller is stopped by asset id rather than missed by key" + original.stopCount == 1 + } + def "UPDATE with the same EAN undeploys the old handler before deploying its replacement"() { given: "a deployed handler" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) From 677aca689eeb921981e0234a34415fb9d85a8d3b Mon Sep 17 00:00:00 2001 From: Michael Date: Wed, 16 Sep 2026 17:35:12 +0200 Subject: [PATCH 12/16] Close the redispatch REST client if the constructor fails GOPACSRedispatchHandler creates its client and then builds two proxies from it. If client.target(...) throws, no reference escapes the constructor, so stopPolling() can never run and the client is left open. Same guard the GOPACS handler already got. --- .../manager/gopacs/GOPACSRedispatchHandler.java | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java index 281c28a..1f475a2 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSRedispatchHandler.java @@ -160,10 +160,18 @@ protected GOPACSRedispatchHandler( container.getConfig().getOrDefault(GOPACS_REDISPATCH_URL, DEFAULT_GOPACS_REDISPATCH_URL); this.client = createClient(org.openremote.container.Container.EXECUTOR); - this.announcementResource = - client.target(redispatchUrl).proxy(GOPACSAnnouncementResource.class); - this.eanEffectivityResource = - client.target(redispatchUrl).proxy(GOPACSEanEffectivityResource.class); + + try { + this.announcementResource = + client.target(redispatchUrl).proxy(GOPACSAnnouncementResource.class); + this.eanEffectivityResource = + client.target(redispatchUrl).proxy(GOPACSEanEffectivityResource.class); + } catch (RuntimeException e) { + // No reference escapes a constructor that threw, so stopPolling() can never close this + // client. + client.close(); + throw e; + } this.objectMapper = new ObjectMapper(); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); From 7f46321d34fb905be086dc7d21d27994aad57ec0 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 17 Sep 2026 09:30:15 +0200 Subject: [PATCH 13/16] Keep the GOPACS handler deployed when a save leaves the EAN unchanged Every save of the asset tore the handler down and built a new one, so a rename dropped the participant cache and cancelled any UFTP conversation waiting on a delayed response or FlexOffer. The redispatch poller was already spared for the same reason. The handler reads only the EAN, the realm and the asset id from the asset, everything else from container config, so leaving it alone when the EAN still maps to this asset is safe. CREATE and UPDATE now run the same reconcile, since creating an asset is the case where nothing is deployed yet. That drops the branch that duplicated the start calls, and with stopping done by asset id the EAN no longer has to be unwrapped before choosing what to do, so clearing the EAN on a save now stops the handler instead of being ignored. Also drops the keepEan parameter from the poller's asset-scoped stop. Deciding which poller to leave alone belongs in the reconcile, which now returns early, and the helper matches its GOPACS counterpart again. --- .../ems/manager/EmsOptimisationService.java | 88 +++++++++---------- .../EmsOptimisationServiceGopacsTest.groovy | 22 +++-- 2 files changed, 61 insertions(+), 49 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index bb1608e..977d365 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -382,17 +382,13 @@ private void stopRedispatchHandler(String contractedEan) { } } - /** - * Stops every redispatch poller started for the asset, whichever EAN it is registered under, - * except one already polling under {@code keepEan}. Pass {@code null} to stop them all. - */ - private void stopRedispatchHandlersForAsset(String assetId, String keepEan) { + /** Stops every redispatch poller started for the asset, whichever EAN it is registered under. */ + private void stopRedispatchHandlersForAsset(String assetId) { gopacsRedispatchHandlerMap .entrySet() .removeIf( entry -> { - if (!assetId.equals(entry.getValue().getAssetId()) - || entry.getKey().equals(keepEan)) { + if (!assetId.equals(entry.getValue().getAssetId())) { return false; } entry.getValue().stopPolling(); @@ -401,18 +397,37 @@ private void stopRedispatchHandlersForAsset(String assetId, String keepEan) { } /** - * Brings the redispatch poller for the asset in line with its saved state: exactly one poller - * under the current EAN when redispatch is enabled, none otherwise. A poller already running - * under the current EAN is left alone, since its announcement bookkeeping is in memory only and a + * Brings the GOPACS handler for the asset in line with its saved state: one handler under the + * current EAN, none while the EAN is blank. A handler already deployed for this asset under that + * EAN is left alone, because redeploying drops its participant cache and every UFTP conversation + * waiting on a delayed response or FlexOffer. + */ + private void reconcileGopacsHandler(EmsGOPACSAsset asset) { + String contractedEan = asset.getContractedEan().orElse(""); + GOPACSHandler current = gopacsHandlerMap.get(contractedEan); + if (current != null && asset.getId().equals(current.getAssetId())) { + return; + } + stopGopacsHandlersForAsset(asset.getId()); + startGopacsHandler(contractedEan, asset.getRealm(), asset.getId()); + } + + /** + * Brings the redispatch poller for the asset in line with its saved state: one poller under the + * current EAN when redispatch is enabled, none otherwise. A poller already running for this asset + * under that EAN is left alone, because its announcement bookkeeping is in memory only and a * restart would re-record every open announcement in the history. */ - private void reconcileRedispatchHandler(EmsGOPACSAsset asset, String contractedEan) { - String assetId = asset.getId(); + private void reconcileRedispatchHandler(EmsGOPACSAsset asset) { + String contractedEan = asset.getContractedEan().orElse(""); boolean enabled = asset.getRedispatchEnabled().orElse(false); - stopRedispatchHandlersForAsset(assetId, enabled ? contractedEan : null); GOPACSRedispatchHandler current = gopacsRedispatchHandlerMap.get(contractedEan); - if (enabled && (current == null || !assetId.equals(current.getAssetId()))) { - startRedispatchHandler(contractedEan, asset.getRealm(), assetId); + if (enabled && current != null && asset.getId().equals(current.getAssetId())) { + return; + } + stopRedispatchHandlersForAsset(asset.getId()); + if (enabled) { + startRedispatchHandler(contractedEan, asset.getRealm(), asset.getId()); } } @@ -423,35 +438,20 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { stopOptimisation(emsEnergyOptimisationAsset.getId()); } } else if (persistenceEvent.getEntity() instanceof EmsGOPACSAsset emsGOPACSAsset) { - emsGOPACSAsset - .getContractedEan() - .ifPresent( - contractedEan -> { - if (persistenceEvent.getCause() == PersistenceEvent.Cause.DELETE) { - // By asset id for the same reason as UPDATE below: the entity carries the EAN as - // saved, which is not necessarily the key the handler is registered under. - stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); - stopRedispatchHandlersForAsset(emsGOPACSAsset.getId(), null); - } - if (persistenceEvent.getCause() == PersistenceEvent.Cause.CREATE) { - startGopacsHandler( - contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - if (emsGOPACSAsset.getRedispatchEnabled().orElse(false)) { - startRedispatchHandler( - contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - } - } - if (persistenceEvent.getCause() == PersistenceEvent.Cause.UPDATE) { - // The entity carries the new EAN, so stopping by EAN would miss a handler - // registered under the previous one and leave it running next to the new one. - stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); - startGopacsHandler( - contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - // Redispatch polling is also managed via attribute events, but an asset merge - // emits none, so the saved EAN and redispatchEnabled are applied here too. - reconcileRedispatchHandler(emsGOPACSAsset, contractedEan); - } - }); + switch (persistenceEvent.getCause()) { + // Stopping goes by asset id because the entity carries the EAN as saved, which is not + // necessarily the key the running handler was registered under. + case DELETE -> { + stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); + stopRedispatchHandlersForAsset(emsGOPACSAsset.getId()); + } + // Redispatch polling is also managed via attribute events, but an asset merge emits none, + // so the saved EAN and redispatchEnabled are applied here too. + case CREATE, UPDATE -> { + reconcileGopacsHandler(emsGOPACSAsset); + reconcileRedispatchHandler(emsGOPACSAsset); + } + } } } diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index 4fe8d6d..9403200 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -201,18 +201,30 @@ class EmsOptimisationServiceGopacsTest extends Specification { original.stopCount == 1 } - def "UPDATE with the same EAN undeploys the old handler before deploying its replacement"() { + def "UPDATE with the same EAN leaves the deployed handler alone"() { given: "a deployed handler" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) def original = createdHandlers[0] - when: + when: "the asset is saved without changing the EAN, for instance a rename" service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset())) - then: "the old handler is stopped and exactly one new handler replaces it" + then: "the handler keeps its participant cache and any UFTP conversation in flight" + original.undeployCount == 0 + createdHandlers.size() == 1 + } + + def "UPDATE that clears the EAN undeploys the handler without deploying a replacement"() { + given: "a deployed handler" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) + def original = createdHandlers[0] + + when: "the asset is saved with the contracted EAN emptied" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset(""))) + + then: "no EAN means no endpoint, so the handler is stopped and none takes its place" original.undeployCount == 1 - createdHandlers.size() == 2 - createdHandlers[1].deployCount == 1 + createdHandlers.size() == 1 } def "UPDATE with a changed EAN undeploys the handler registered under the old EAN"() { From d87da7e2fef6f2072d907454d33900a52c55eeff Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 17 Sep 2026 10:12:53 +0200 Subject: [PATCH 14/16] Give the GOPACS lifecycle a single owner across both event paths A merge does raise attribute events. AssetStorageService.publishModificationEvents() publishes one per created attribute on CREATE, and on UPDATE one per added or modified attribute, or one per attribute outright when the save touched anything besides the attributes. So every save of a GOPACS asset reached this service twice: once through the persistence route, once through the attribute event subscription, which ran their own separate stop/start pairs. That undid the previous commit. Renaming the asset republishes an unchanged contractedEan, and the attribute path tore the handler down and built a new one anyway, losing the participant cache and every UFTP conversation waiting on a delayed response or FlexOffer. The two paths also run on different threads: PersistenceTopic is a seda endpoint with multipleConsumers, and the subscriber dispatch route hands events to its own executor. Interleaved, both stops could find nothing to stop and both puts land, leaving the first handler registered over by the second with its endpoint and client still open. Both paths now call the same reconcile, so whichever arrives second finds the handler already in its target state and leaves it alone, and the whole transition is held under a lifecycle lock so the two cannot interleave. Container start and stop take the same route, which drops the duplicated start calls in start(). The tests play back a save as the manager performs it, the persistence event followed by the attribute events raised from it, rather than the persistence event on its own. --- .../ems/manager/EmsOptimisationService.java | 107 ++++++--------- .../EmsOptimisationServiceGopacsTest.groovy | 124 ++++++++++++++++++ 2 files changed, 166 insertions(+), 65 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java index 977d365..fb296e9 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/EmsOptimisationService.java @@ -77,6 +77,11 @@ public class EmsOptimisationService extends RouteBuilder implements ContainerSer private final Map gopacsHandlerMap = new ConcurrentHashMap<>(); private final Map gopacsRedispatchHandlerMap = new ConcurrentHashMap<>(); + // Every save of a GOPACS asset reaches this service twice, over two threads: once as a + // persistence event and once as the attribute events AssetStorageService raises from it. Held + // across a whole transition so the two cannot interleave their stop/start pairs and leave a + // handler registered over a live one that never gets undeployed. + private final Object gopacsLifecycleLock = new Object(); @SuppressWarnings("unchecked") @Override @@ -156,21 +161,7 @@ public void start(Container container) throws Exception { .attributeName(EmsGOPACSAsset.CONTRACTED_EAN.getName())) .stream() .map(asset -> (EmsGOPACSAsset) asset) - .forEach( - gopacsAsset -> { - startGopacsHandler( - gopacsAsset.getContractedEan().orElse(""), - gopacsAsset.getRealm(), - gopacsAsset.getId()); - - // Start redispatch handler if enabled - if (gopacsAsset.getRedispatchEnabled().orElse(false)) { - startRedispatchHandler( - gopacsAsset.getContractedEan().orElse(""), - gopacsAsset.getRealm(), - gopacsAsset.getId()); - } - }); + .forEach(this::reconcileGopacsAsset); // List of asset types that are part of the core EMS service String[] assetTypes = { @@ -190,10 +181,12 @@ public void start(Container container) throws Exception { @Override public void stop(Container container) throws Exception { - gopacsHandlerMap.forEach((ean, handler) -> handler.undeploy()); - gopacsHandlerMap.clear(); - gopacsRedispatchHandlerMap.forEach((ean, handler) -> handler.stopPolling()); - gopacsRedispatchHandlerMap.clear(); + synchronized (gopacsLifecycleLock) { + gopacsHandlerMap.forEach((ean, handler) -> handler.undeploy()); + gopacsHandlerMap.clear(); + gopacsRedispatchHandlerMap.forEach((ean, handler) -> handler.stopPolling()); + gopacsRedispatchHandlerMap.clear(); + } energyOptimisationAssetsMap.forEach((assetId, scheduledFuture) -> stopOptimisation(assetId)); energyOptimisationTimersMap.clear(); } @@ -396,6 +389,27 @@ private void stopRedispatchHandlersForAsset(String assetId) { }); } + /** + * Brings the GOPACS handler and the redispatch poller for the asset in line with its saved state. + * This is the only thing that starts or stops either of them for a live asset, so the persistence + * event and the attribute events raised from the same save converge on one outcome instead of + * each running their own stop/start pair. + */ + private void reconcileGopacsAsset(EmsGOPACSAsset asset) { + synchronized (gopacsLifecycleLock) { + reconcileGopacsHandler(asset); + reconcileRedispatchHandler(asset); + } + } + + /** Undeploys the GOPACS handler and the redispatch poller of a deleted asset. */ + private void stopGopacsAsset(String assetId) { + synchronized (gopacsLifecycleLock) { + stopGopacsHandlersForAsset(assetId); + stopRedispatchHandlersForAsset(assetId); + } + } + /** * Brings the GOPACS handler for the asset in line with its saved state: one handler under the * current EAN, none while the EAN is blank. A handler already deployed for this asset under that @@ -441,16 +455,8 @@ protected void processAssetChange(PersistenceEvent persistenceEvent) { switch (persistenceEvent.getCause()) { // Stopping goes by asset id because the entity carries the EAN as saved, which is not // necessarily the key the running handler was registered under. - case DELETE -> { - stopGopacsHandlersForAsset(emsGOPACSAsset.getId()); - stopRedispatchHandlersForAsset(emsGOPACSAsset.getId()); - } - // Redispatch polling is also managed via attribute events, but an asset merge emits none, - // so the saved EAN and redispatchEnabled are applied here too. - case CREATE, UPDATE -> { - reconcileGopacsHandler(emsGOPACSAsset); - reconcileRedispatchHandler(emsGOPACSAsset); - } + case DELETE -> stopGopacsAsset(emsGOPACSAsset.getId()); + case CREATE, UPDATE -> reconcileGopacsAsset(emsGOPACSAsset); } } } @@ -794,42 +800,13 @@ private void processAttributeEventEmsGOPACSAsset(AttributeEvent attributeEvent) String attributeName = attributeEvent.getName(); - if (attributeName.equals(EmsGOPACSAsset.CONTRACTED_EAN.getName())) { - attributeEvent - .getOldValue(String.class) - .ifPresent( - oldEan -> { - stopGopacsHandler(oldEan); - stopRedispatchHandler(oldEan); - }); - attributeEvent - .getValue(String.class) - .ifPresent( - contractedEan -> { - startGopacsHandler( - contractedEan, attributeEvent.getRealm(), attributeEvent.getId()); - if (gopacsAsset.getRedispatchEnabled().orElse(false)) { - startRedispatchHandler( - contractedEan, attributeEvent.getRealm(), attributeEvent.getId()); - } - }); - } - - // Handle redispatch enabled/disabled toggle - if (attributeName.equals(EmsGOPACSAsset.REDISPATCH_ENABLED.getName())) { - gopacsAsset - .getContractedEan() - .ifPresent( - contractedEan -> { - boolean enabled = (Boolean) attributeEvent.getValue().orElse(false); - if (enabled) { - stopRedispatchHandler(contractedEan); // Stop existing if any - startRedispatchHandler( - contractedEan, gopacsAsset.getRealm(), gopacsAsset.getId()); - } else { - stopRedispatchHandler(contractedEan); - } - }); + // A save reaches this service as a persistence event and again as the attribute events + // AssetStorageService raises from it, and a save that touches anything besides the attributes + // republishes every attribute whether it changed or not. Both paths run the same reconcile, so + // whichever arrives second finds the handler already in its target state and leaves it alone. + if (attributeName.equals(EmsGOPACSAsset.CONTRACTED_EAN.getName()) + || attributeName.equals(EmsGOPACSAsset.REDISPATCH_ENABLED.getName())) { + reconcileGopacsAsset(gopacsAsset); } // Handle bid confirmation diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index 9403200..6b08681 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -28,7 +28,10 @@ import org.openremote.manager.asset.AssetStorageService import org.openremote.manager.datapoint.AssetPredictedDatapointService import org.openremote.model.Container import org.openremote.model.PersistenceEvent +import org.openremote.model.attribute.Attribute +import org.openremote.model.attribute.AttributeEvent import org.openremote.model.util.ValueUtil +import org.openremote.model.value.AttributeDescriptor import spock.lang.Shared import spock.lang.Specification @@ -45,6 +48,11 @@ import java.util.concurrent.ScheduledExecutorService * {@code gopacsHandlerMap} is keyed by contracted EAN. These tests catch stop calls that use a key * the running handler was never registered under: {@code Map.remove} with the wrong key is a silent * no-op, so the old handler keeps its endpoint and client alive next to its replacement. + * + * Saving an asset reaches this service twice: once as the persistence event, and again as the + * attribute events {@code AssetStorageService.publishModificationEvents()} raises from the same + * merge. {@link #save} plays back both, so the tests below cover the whole save rather than the + * persistence event on its own. */ class EmsOptimisationServiceGopacsTest extends Specification { @@ -57,6 +65,7 @@ class EmsOptimisationServiceGopacsTest extends Specification { @Shared File privateKeyFile List createdHandlers List createdRedispatchHandlers + Map storedAssets EmsOptimisationService service def setupSpec() { @@ -91,8 +100,16 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdHandlers = [] createdRedispatchHandlers = [] + storedAssets = [:] service = new EmsOptimisationService() + // The attribute event path reads the saved asset back, so the tests keep the assets they save + // in storedAssets and hand them out here. + service.services = Services.builder() + .withAssetStorageService(Stub(AssetStorageService) { + find(_ as String) >> { String assetId -> storedAssets[assetId] } + }) + .build() service.gopacsHandlerFactory = new GOPACSHandler.Factory(handlerContainer) { @Override GOPACSHandler createHandler(String contractedEan, String realm, String assetId) { @@ -128,6 +145,40 @@ class EmsOptimisationServiceGopacsTest extends Specification { return new PersistenceEvent<>(cause, asset, null, null, null) } + /** + * The event {@code AssetStorageService.publishModificationEvents()} raises for one attribute of a + * saved asset. + */ + private static AttributeEvent attributeEvent( + EmsGOPACSAsset asset, Attribute attribute, Object oldValue) { + return new AttributeEvent(asset, attribute, "AssetStorageService", + attribute.getValue().orElse(null), attribute.getTimestamp().orElse(0L), oldValue, 0L) + } + + private static AttributeEvent attributeEvent( + EmsGOPACSAsset asset, AttributeDescriptor descriptor, Object oldValue) { + return attributeEvent(asset, asset.getAttributes().get(descriptor.getName()).orElseThrow(), + oldValue) + } + + /** + * Saves the asset the way the manager does: the persistence event for the merge, then an attribute + * event per attribute. A save that touches anything besides the attributes, a rename for instance, + * republishes every attribute whether its value changed or not, which is the case these tests are + * mostly about. {@code oldEan} is the EAN the save replaced, or null when the EAN is unchanged. + */ + private void save(PersistenceEvent.Cause cause, EmsGOPACSAsset asset, String oldEan = null) { + storedAssets[asset.getId()] = asset + service.processAssetChange(event(cause, asset)) + asset.getAttributes().values().each { attribute -> + def oldValue = + attribute.getName() == EmsGOPACSAsset.CONTRACTED_EAN.getName() && oldEan != null + ? oldEan + : attribute.getValue().orElse(null) + service.processAttributeEvent(attributeEvent(asset, attribute, oldValue)) + } + } + def "CREATE deploys a handler for the asset's EAN"() { when: service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) @@ -311,6 +362,79 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdRedispatchHandlers.size() == 1 } + def "a save that leaves the EAN alone keeps the handler through the attribute events it raises"() { + given: "a deployed handler" + save(PersistenceEvent.Cause.CREATE, gopacsAsset()) + def original = createdHandlers[0] + + when: "the asset is saved without changing the EAN, for instance a rename" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset()) + + then: "the republished contractedEan does not undo what the persistence event reconciled" + original.undeployCount == 0 + createdHandlers.size() == 1 + } + + def "a save that leaves redispatch enabled keeps the poller through the attribute events it raises"() { + given: "a running poller" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true)) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved without changing the EAN or the redispatch toggle" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, true)) + + then: "the poller keeps its in-memory announcement bookkeeping" + original.stopCount == 0 + createdRedispatchHandlers.size() == 1 + } + + def "a save that changes the EAN ends with one handler, for the new EAN"() { + given: "a deployed handler for the original EAN" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN)) + def original = createdHandlers[0] + + when: "the asset is saved with a different EAN" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset(OTHER_EAN), EAN) + + then: "the persistence event makes the swap and the attribute event leaves it standing" + original.undeployCount == 1 + createdHandlers.size() == 2 + createdHandlers[1].contractedEAN == OTHER_EAN + createdHandlers[1].deployCount == 1 + createdHandlers[1].undeployCount == 0 + } + + def "an attribute write that enables redispatch starts a poller"() { + given: "a deployed handler with redispatch off" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, false)) + + when: "redispatch is switched on, which raises an attribute event and no persistence event" + def asset = gopacsAsset(EAN, ASSET_ID, true) + storedAssets[ASSET_ID] = asset + service.processAttributeEvent(attributeEvent(asset, EmsGOPACSAsset.REDISPATCH_ENABLED, false)) + + then: "the attribute event path still drives the lifecycle on its own" + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].contractedEAN == EAN + createdRedispatchHandlers[0].startCount == 1 + } + + def "an attribute write that disables redispatch stops the poller"() { + given: "a running poller" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true)) + def original = createdRedispatchHandlers[0] + + when: "redispatch is switched off through an attribute write" + def asset = gopacsAsset(EAN, ASSET_ID, false) + storedAssets[ASSET_ID] = asset + service.processAttributeEvent(attributeEvent(asset, EmsGOPACSAsset.REDISPATCH_ENABLED, true)) + + then: "the poller stops and the GOPACS handler is left deployed" + original.stopCount == 1 + createdRedispatchHandlers.size() == 1 + createdHandlers[0].undeployCount == 0 + } + def "a handler deployed for an EAN already in use undeploys the one it displaces"() { given: "a deployed handler for the EAN" service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID))) From 28c50eed58b3eb89fb6d6509522acaf1f0f320c7 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 17 Sep 2026 10:23:16 +0200 Subject: [PATCH 15/16] Keep GOPACSHandler's scheduling internals private schedule() was widened to protected and pendingTaskCount() added purely so GOPACSHandlerTest could reach them. Neither belongs in the production API: schedule() is called only from within the handler, and the pending count is internal bookkeeping no caller has any business reading. schedule() goes back to private and pendingTaskCount() is gone. The test reaches both reflectively, which it has to do because Groovy dispatches to the recording subclass and will not find a private member of the superclass. --- .../ems/manager/gopacs/GOPACSHandler.java | 6 +---- .../manager/gopacs/GOPACSHandlerTest.groovy | 25 +++++++++++++++---- 2 files changed, 21 insertions(+), 10 deletions(-) diff --git a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java index ace295d..b571442 100644 --- a/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java +++ b/ems/src/main/java/org/openremote/extension/ems/manager/gopacs/GOPACSHandler.java @@ -361,7 +361,7 @@ public void deploy() { * {@link #processRawMessage(String)} at that point would otherwise queue work that outlives the * handler and later runs against a closed client. */ - protected void schedule(Runnable task, long delayMillis) { + private void schedule(Runnable task, long delayMillis) { synchronized (scheduledFutureList) { if (undeployed) { LOG.fine("Handler is undeployed, dropping scheduled task for EAN: " + contractedEAN); @@ -373,10 +373,6 @@ protected void schedule(Runnable task, long delayMillis) { } } - protected int pendingTaskCount() { - return scheduledFutureList.size(); - } - public void undeploy() { // Undeploy the endpoint first so no further request can reach processRawMessage and start work // that the cancel loop below would then have to race. diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy index 884adf7..91221f7 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/gopacs/GOPACSHandlerTest.groovy @@ -294,7 +294,7 @@ class GOPACSHandlerTest extends Specification { signAndProcess(flexRequestXml(CONTRACTED_EAN)) def firstBatch = new ArrayList<>(scheduledFutures) assert !firstBatch.isEmpty() - assert handler.pendingTaskCount() == firstBatch.size() + assert pendingTaskCount(handler) == firstBatch.size() and: "every task from that batch is now already completed" firstBatch.each { it.isDone() >> true } @@ -309,7 +309,7 @@ class GOPACSHandlerTest extends Specification { // compareTo() rather than equals() for these mocks, and an unstubbed compareTo() always // answers 0 -- making every mock instance compare "equal" to every other one.) def secondBatch = scheduledFutures.subList(firstBatch.size(), scheduledFutures.size()) - handler.pendingTaskCount() == secondBatch.size() + pendingTaskCount(handler) == secondBatch.size() } def "a request still in flight when undeploy starts cannot schedule further work"() { @@ -317,11 +317,11 @@ class GOPACSHandlerTest extends Specification { handler.undeploy() when: "a request that was already being processed reaches a schedule() call" - handler.schedule({} as Runnable, 0L) + schedule(handler, {} as Runnable, 0L) then: "nothing is handed to the executor, so no task outlives the handler" scheduledFutures.isEmpty() - handler.pendingTaskCount() == 0 + pendingTaskCount(handler) == 0 } def "a message processed after undeploy schedules nothing"() { @@ -333,11 +333,26 @@ class GOPACSHandlerTest extends Specification { then: "neither the delayed response nor the delayed FlexOffer is scheduled" scheduledFutures.isEmpty() - handler.pendingTaskCount() == 0 + pendingTaskCount(handler) == 0 } // ---- Embedded UFTP payload fixtures (attribute-style XML, matching the example message format) ---- + // schedule() and the list of futures it tracks are private to GOPACSHandler, and Groovy will not + // dispatch to a private member of a superclass, so these two reach them reflectively rather than + // widening the production API for the tests. + private static int pendingTaskCount(GOPACSHandler handler) { + def field = GOPACSHandler.getDeclaredField("scheduledFutureList") + field.setAccessible(true) + return (field.get(handler) as List).size() + } + + private static void schedule(GOPACSHandler handler, Runnable task, long delayMillis) { + def method = GOPACSHandler.getDeclaredMethod("schedule", Runnable, Long.TYPE) + method.setAccessible(true) + method.invoke(handler, task, delayMillis) + } + private static String flexRequestXml(String congestionPoint) { """ Date: Thu, 17 Sep 2026 10:35:00 +0200 Subject: [PATCH 16/16] Cover EAN-clear and redispatch-toggle through the combined save path save() already replayed the persistence event and its attribute events for an EAN change and a rename, but clearing the EAN and toggling redispatch were only tested through processAssetChange() directly. Both are handled by the same nonAttributeChange republish as a rename, so they are exactly the case the lifecycle-lock fix in the previous commit is meant to cover. save() now takes an optional oldRedispatchEnabled, alongside the existing oldEan, so a test can state the toggle's previous value when the republished attribute event needs one. --- .../EmsOptimisationServiceGopacsTest.groovy | 58 +++++++++++++++++-- 1 file changed, 53 insertions(+), 5 deletions(-) diff --git a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy index 6b08681..bded402 100644 --- a/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -166,15 +166,23 @@ class EmsOptimisationServiceGopacsTest extends Specification { * event per attribute. A save that touches anything besides the attributes, a rename for instance, * republishes every attribute whether its value changed or not, which is the case these tests are * mostly about. {@code oldEan} is the EAN the save replaced, or null when the EAN is unchanged. + * {@code oldRedispatchEnabled} is the toggle's previous value, or null when it is unchanged. */ - private void save(PersistenceEvent.Cause cause, EmsGOPACSAsset asset, String oldEan = null) { + private void save( + PersistenceEvent.Cause cause, EmsGOPACSAsset asset, String oldEan = null, + Boolean oldRedispatchEnabled = null) { storedAssets[asset.getId()] = asset service.processAssetChange(event(cause, asset)) asset.getAttributes().values().each { attribute -> - def oldValue = - attribute.getName() == EmsGOPACSAsset.CONTRACTED_EAN.getName() && oldEan != null - ? oldEan - : attribute.getValue().orElse(null) + def oldValue + if (attribute.getName() == EmsGOPACSAsset.CONTRACTED_EAN.getName() && oldEan != null) { + oldValue = oldEan + } else if (attribute.getName() == EmsGOPACSAsset.REDISPATCH_ENABLED.getName() + && oldRedispatchEnabled != null) { + oldValue = oldRedispatchEnabled + } else { + oldValue = attribute.getValue().orElse(null) + } service.processAttributeEvent(attributeEvent(asset, attribute, oldValue)) } } @@ -404,6 +412,46 @@ class EmsOptimisationServiceGopacsTest extends Specification { createdHandlers[1].undeployCount == 0 } + def "a save that clears the EAN through the attribute events it raises undeploys the handler without deploying a replacement"() { + given: "a deployed handler" + save(PersistenceEvent.Cause.CREATE, gopacsAsset()) + def original = createdHandlers[0] + + when: "the asset is saved with the contracted EAN emptied" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset(""), EAN) + + then: "the republished empty EAN does not deploy a handler behind the stop the persistence event already made" + original.undeployCount == 1 + createdHandlers.size() == 1 + } + + def "a save that enables redispatch through the attribute events it raises starts a poller"() { + given: "an asset without redispatch" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, false)) + + when: "the asset is saved with redispatch switched on" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, true), null, false) + + then: "the republished toggle does not stop the poller the persistence event already started" + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].contractedEAN == EAN + createdRedispatchHandlers[0].startCount == 1 + } + + def "a save that disables redispatch through the attribute events it raises stops the poller"() { + given: "a running poller" + save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, true)) + def original = createdRedispatchHandlers[0] + + when: "the asset is saved with redispatch switched off" + save(PersistenceEvent.Cause.UPDATE, gopacsAsset(EAN, ASSET_ID, false), null, true) + + then: "the republished toggle does not start a second poller behind the stop the persistence event already made" + original.stopCount == 1 + createdRedispatchHandlers.size() == 1 + createdHandlers[0].undeployCount == 0 + } + def "an attribute write that enables redispatch starts a poller"() { given: "a deployed handler with redispatch off" save(PersistenceEvent.Cause.CREATE, gopacsAsset(EAN, ASSET_ID, false))