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 efcec78..cf31cc4 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,8 +77,16 @@ 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<>(); + // 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(); // Keyed by asset id rather than portfolio: unlike GOPACS there is no external routing key, and // the asset id stays stable when the portfolio is edited. @@ -163,21 +171,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); // Start Distro Energy handler for all Distro Energy assets services @@ -209,8 +203,12 @@ public void start(Container container) throws Exception { @Override public void stop(Container container) throws Exception { - 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(); + } distroEnergyHandlerMap.forEach((assetId, handler) -> handler.undeploy()); distroEnergyHandlerMap.clear(); energyOptimisationAssetsMap.forEach((assetId, scheduledFuture) -> stopOptimisation(assetId)); @@ -342,8 +340,14 @@ 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)); + // 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. + gopacsHandlerMap.put(contractedEan, handler); + handler.deploy(); LOG.fine("Deployed GOPACS for EAN: " + contractedEan); } @@ -355,12 +359,31 @@ 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"); 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); @@ -376,6 +399,76 @@ private void stopRedispatchHandler(String contractedEan) { } } + /** 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())) { + return false; + } + entry.getValue().stopPolling(); + return true; + }); + } + + /** + * 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 + * 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 = asset.getContractedEan().orElse(""); + boolean enabled = asset.getRedispatchEnabled().orElse(false); + GOPACSRedispatchHandler current = gopacsRedispatchHandlerMap.get(contractedEan); + if (enabled && current != null && asset.getId().equals(current.getAssetId())) { + return; + } + stopRedispatchHandlersForAsset(asset.getId()); + if (enabled) { + startRedispatchHandler(contractedEan, asset.getRealm(), asset.getId()); + } + } + private void startDistroEnergyHandler(EmsDistroEnergyAsset distroEnergyAsset) { String assetId = distroEnergyAsset.getId(); String portfolio = distroEnergyAsset.getPortfolio().orElse(""); @@ -449,29 +542,12 @@ 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) { - stopGopacsHandler(contractedEan); - stopRedispatchHandler(contractedEan); - } - 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) { - stopGopacsHandler(contractedEan); - startGopacsHandler( - contractedEan, emsGOPACSAsset.getRealm(), emsGOPACSAsset.getId()); - // Redispatch handler is managed via attribute events (redispatchEnabled) - } - }); + 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 -> stopGopacsAsset(emsGOPACSAsset.getId()); + case CREATE, UPDATE -> reconcileGopacsAsset(emsGOPACSAsset); + } } else if (persistenceEvent.getEntity() instanceof EmsDistroEnergyAsset emsDistroEnergyAsset) { // distroEnergyHandlerMap is keyed by asset id, not portfolio: stop by asset id so this // actually finds the handler to remove, and so DELETE cleans up even if the portfolio @@ -833,42 +909,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/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..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 @@ -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; @@ -141,7 +142,15 @@ 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<>()); + + // 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; @@ -158,6 +167,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; @@ -218,14 +228,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); @@ -241,8 +259,6 @@ protected GOPACSHandler( this.objectMapper = new ObjectMapper(); objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); objectMapper.setVisibility(PropertyAccessor.FIELD, JsonAutoDetect.Visibility.ANY); - - deploy(container); } /** @@ -270,6 +286,7 @@ protected GOPACSHandler( this.timerService = timerService; this.scheduledExecutorService = scheduledExecutorService; this.webService = null; + this.container = null; this.gopacsBrokerUrl = ""; this.responseDelaySeconds = 0; @@ -296,11 +313,23 @@ 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; } - 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 = @@ -323,12 +352,45 @@ 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. + * + *

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. + */ + private 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)); + } + } + public void undeploy() { - for (ScheduledFuture scheduledFuture : scheduledFutureList) { - scheduledFuture.cancel(true); + // 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 (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 +675,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 +770,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; } @@ -721,20 +781,19 @@ protected void processRawMessage(String transportXml) { new UftpParticipant(signedMessage), payloadMessage, transportXml, payloadXml); notifyNewIncomingMessage(incomingUftpMessage); - // Send response delayed to ensure HTTP response is sent first - scheduledExecutorService.schedule( + // Delayed so the HTTP response on the transport call goes out first + schedule( () -> { uftpReceivedMessageService.process(incomingUftpMessage); }, - this.responseDelaySeconds, - TimeUnit.SECONDS); // 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 - scheduledExecutorService.schedule( + // Delayed so the FlexRequestResponse is sent and processed by the other party first + schedule( () -> { try { sendFlexOffer(participant, flexRequest); @@ -742,10 +801,7 @@ 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)); } } catch (UftpConnectorException e) { LOG.log(Level.SEVERE, "Error processing raw message", e); 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..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); @@ -177,6 +185,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 new file mode 100644 index 0000000..bded402 --- /dev/null +++ b/ems/src/test/groovy/org/openremote/extension/ems/manager/EmsOptimisationServiceGopacsTest.groovy @@ -0,0 +1,574 @@ +/* + * 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.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 +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 + +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. 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 + * 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 { + + 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 + List createdRedispatchHandlers + Map storedAssets + 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(AssetStorageService) >> Stub(AssetStorageService) + getService(AssetPredictedDatapointService) >> Stub(AssetPredictedDatapointService) + getService(TimerService) >> Stub(TimerService) + getService(WebService) >> Stub(WebService) + getScheduledExecutor() >> Stub(ScheduledExecutorService) + } + + 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) { + def handler = + new RecordingGOPACSHandler(contractedEan, realm, assetId, handlerContainer) + createdHandlers << handler + 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, 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 + } + + private static PersistenceEvent event( + PersistenceEvent.Cause cause, EmsGOPACSAsset asset) { + 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. + * {@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, + Boolean oldRedispatchEnabled = null) { + storedAssets[asset.getId()] = asset + service.processAssetChange(event(cause, asset)) + asset.getAttributes().values().each { attribute -> + 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)) + } + } + + 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 + + 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"() { + 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 "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 "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 leaves the deployed handler alone"() { + given: "a deployed handler" + service.processAssetChange(event(PersistenceEvent.Cause.CREATE, gopacsAsset())) + def original = createdHandlers[0] + + when: "the asset is saved without changing the EAN, for instance a rename" + service.processAssetChange(event(PersistenceEvent.Cause.UPDATE, gopacsAsset())) + + 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() == 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 "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 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))) + 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 "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 "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)) + + 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))) + 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))) + 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 } + createdRedispatchHandlers.size() == 1 + createdRedispatchHandlers[0].stopCount == 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++ + } + } + + // 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++ + } + } +} 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 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..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 @@ -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,8 +260,99 @@ 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 pendingTaskCount(handler) == 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()) + pendingTaskCount(handler) == 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" + schedule(handler, {} as Runnable, 0L) + + then: "nothing is handed to the executor, so no task outlives the handler" + scheduledFutures.isEmpty() + pendingTaskCount(handler) == 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() + 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) { """