diff --git a/src/main/java/org/phoebus/recceiver/RecceiverApplication.java b/src/main/java/org/phoebus/recceiver/RecceiverApplication.java index eed70b2..86af9d0 100644 --- a/src/main/java/org/phoebus/recceiver/RecceiverApplication.java +++ b/src/main/java/org/phoebus/recceiver/RecceiverApplication.java @@ -93,8 +93,8 @@ public class RecceiverApplication implements CommandLineRunner { @Value("${recceiver.cf.clean-on-startup:false}") private String cfCleanOnStartup; - @Value("${recceiver.cf.delete-on-shutdown:true}") - private String cfDeleteOnShutdown; + @Value("${recceiver.cf.clean-on-shutdown:true}") + private String cfCleanOnShutdown; @Value("${recceiver.cf.alias-enabled:true}") private String cfAliasEnabled; @@ -236,7 +236,7 @@ private Map buildProcessorOptions() { options.put("recceiver.cf.password", cfPassword); options.put("recceiver.cf.delete-on-clean", cfDeleteOnClean); options.put("recceiver.cf.clean-on-startup", cfCleanOnStartup); - options.put("recceiver.cf.delete-on-shutdown", cfDeleteOnShutdown); + options.put("recceiver.cf.clean-on-shutdown", cfCleanOnShutdown); options.put("recceiver.cf.alias-enabled", cfAliasEnabled); options.put("recceiver.cf.infotags", cfInfoTags); options.put("recceiver.cf.environment-vars", cfEnvironmentVars); diff --git a/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java b/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java index 8c57723..c1f52ce 100644 --- a/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java +++ b/src/main/java/org/phoebus/recceiver/spi/cf/CFRecSyncProcessor.java @@ -10,6 +10,7 @@ import java.time.OffsetDateTime; import java.util.ArrayList; import java.util.Collection; +import java.util.EnumSet; import java.util.HashSet; import java.util.LinkedHashMap; import java.util.LinkedHashSet; @@ -18,6 +19,7 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Logger; +import java.util.stream.Collectors; /** * Publishes channel records from IOCs to a ChannelFinder service. @@ -39,19 +41,23 @@ *
  • Connection Loss ({@link #connectionClosed(ConnectionClosedEvent)}): * Immediately marks all IOC channels as inactive (or deletes them, depending on * {@link #deleteOnClean}).
    • - *
  • Shutdown ({@link #destroy()}): Iterates over all tracked IOCs and - * either deletes or marks inactive, per {@link #deleteOnShutdown} preference.
    • + *
  • Shutdown ({@link #destroy()}): If {@link #cleanOnShutdown} is true, either + * deletes all of this recceiver's channels (if {@link #deleteOnClean} is true) or marks its + * Active and Inactive channels Unknown. Otherwise leaves ChannelFinder untouched.
    • * *

    * *

    Cleanup Preferences (Booleans): *

    *

    * @@ -85,7 +91,7 @@ public class CFRecSyncProcessor implements RecSyncProcessor { private static final String PASSWORD_KEY = "recceiver.cf.password"; private static final String DELETE_ON_CLEAN_KEY = "recceiver.cf.delete-on-clean"; private static final String CLEAN_ON_STARTUP_KEY = "recceiver.cf.clean-on-startup"; - private static final String DELETE_ON_SHUTDOWN_KEY = "recceiver.cf.delete-on-shutdown"; + private static final String CLEAN_ON_SHUTDOWN_KEY = "recceiver.cf.clean-on-shutdown"; private static final String ALIAS_ENABLED_KEY = "recceiver.cf.alias-enabled"; private static final String INFOTAGS_KEY = "recceiver.cf.infotags"; private static final String ENVIRONMENT_VARS_KEY = "recceiver.cf.environment-vars"; @@ -113,9 +119,6 @@ public class CFRecSyncProcessor implements RecSyncProcessor { public static final String ALIAS_PROPERTY = "alias"; public static final String CA_PORT_PROPERTY = "caPort"; public static final String PVA_PORT_PROPERTY = "pvaPort"; - public static final String ACTIVE = "Active"; - public static final String INACTIVE = "Inactive"; - public static final String UNDEFINED = "Undefined"; public static final String DEFAULT_OWNER = "cfstore"; private static final String RECCEIVER_ID = resolveRecceiverId(); @@ -136,14 +139,13 @@ public class CFRecSyncProcessor implements RecSyncProcessor { private ChannelFinderStore store; private boolean deleteOnClean; private boolean cleanOnStartup; - private boolean deleteOnShutdown; + private boolean cleanOnShutdown; private boolean aliasEnabled; private Set infoTagWhitelist = Set.of(); private Map environmentVarMappings = Map.of(); // IOC connection registry: tracks all IOCs that have connected to this processor. // Key = iocId (e.g., "host:port"), Value = IocInfo metadata - // Used during shutdown to clean up channels belonging to tracked IOCs. private final Map iocRegistry = new ConcurrentHashMap<>(); // Tracks recId -> channel names so delete messages (which only carry recId) can be resolved. private final Map iocChannelState = new ConcurrentHashMap<>(); @@ -172,7 +174,7 @@ public RecSyncProcessor init(String name, Map options) { this.store = createStore(serviceUrl, username, password); this.deleteOnClean = Boolean.parseBoolean(options.getOrDefault(DELETE_ON_CLEAN_KEY, "false")); this.cleanOnStartup = Boolean.parseBoolean(options.getOrDefault(CLEAN_ON_STARTUP_KEY, "false")); - this.deleteOnShutdown = Boolean.parseBoolean(options.getOrDefault(DELETE_ON_SHUTDOWN_KEY, "true")); + this.cleanOnShutdown = Boolean.parseBoolean(options.getOrDefault(CLEAN_ON_SHUTDOWN_KEY, "true")); this.aliasEnabled = Boolean.parseBoolean(options.getOrDefault(ALIAS_ENABLED_KEY, "false")); this.infoTagWhitelist = parseInfoTags(options.get(INFOTAGS_KEY)); // Parse once at init so commit path only performs lookups. @@ -186,49 +188,63 @@ public RecSyncProcessor init(String name, Map options) { requiredProperties.addAll(environmentVarMappings.values()); this.store.ensureProperties(requiredProperties, propertyOwner); - cleanupExistingActiveChannelsForRecceiver(); + if (cleanOnStartup) { + cleanupExistingActiveChannelsForRecceiver(); + } return this; } /** - * On startup, cleans up any active channels previously published by this recceiver instance. - * - *

    Finds all channels tagged with this recceiver's ID and filters for those with pvStatus=Active. - * If {@code cleanOnStartup} is true, deletes them. Otherwise, marks them as inactive with an - * updated timestamp. This prevents stale channels from lingering when the recceiver restarts.

    + * On startup, cleans up any no inactive channels previously published by this recceiver instance. */ private void cleanupExistingActiveChannelsForRecceiver() { Map existing = store.findByProperty(RECCEIVER_ID_PROPERTY, RECCEIVER_ID); - if (existing.isEmpty()) { - return; - } + cleanChannels(existing, PvStatus.INACTIVE); + } - Map activeChannels = new LinkedHashMap<>(); - for (Map.Entry channelEntry : existing.entrySet()) { - if (ACTIVE.equals(channelEntry.getValue().property(PV_STATUS_PROPERTY))) { - activeChannels.put(channelEntry.getKey(), channelEntry.getValue()); - } - } - if (activeChannels.isEmpty()) { + /** + * Returns the channels whose pvStatus is NOT IN {@code excludedStatuses}. + * + * @param channels channels to filter, keyed by name. + * @param excludedStatuses statuses to skip; empty to keep every channel. + */ + private static Map filterChannelsExcluding(Map channels, + Set excludedStatuses) { + Set excluded = excludedStatuses.stream() + .map(PvStatus::value) + .collect(Collectors.toCollection(HashSet::new)); + return channels.entrySet().parallelStream() + .filter(entry -> !excluded.contains(entry.getValue().property(PV_STATUS_PROPERTY))) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (a, b) -> a, LinkedHashMap::new)); + } + + /** + * Applies the configured cleanup to the given channels: deletes them if {@link #deleteOnClean} + * is true, otherwise sets their pvStatus to {@code statusIfKept} and refreshes their timestamp. + * + * @param channels channels to clean, keyed by name. + * @param statusIfKept pvStatus to write when channels are kept rather than deleted. + */ + private void cleanChannels(Map channels, PvStatus statusIfKept) { + if (channels.isEmpty()) { return; } - - // Startup cleanup mode: cleanOnStartup=true => delete, false => mark inactive. - if (cleanOnStartup) { - store.deleteChannels(activeChannels.keySet()); + if (deleteOnClean) { + store.deleteChannels(channels.keySet()); return; } - + Map filteredChannels = filterChannelsExcluding(channels, + EnumSet.of(statusIfKept)); String now = OffsetDateTime.now().toString(); - Collection inactiveChannels = new ArrayList<>(); - for (ChannelFinderChannel channel : activeChannels.values()) { + Collection updated = new ArrayList<>(); + for (ChannelFinderChannel channel : filteredChannels.values()) { Map properties = new LinkedHashMap<>(channel.properties()); - properties.put(PV_STATUS_PROPERTY, INACTIVE); + properties.put(PV_STATUS_PROPERTY, statusIfKept.value()); properties.put(TIME_PROPERTY, now); - inactiveChannels.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); + updated.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); } - store.updateChannels(inactiveChannels); + store.updateChannels(updated); } protected ChannelFinderStore createStore(String serviceUrl, String username, String password) { @@ -240,8 +256,7 @@ protected ChannelFinderStore createStore(String serviceUrl, String username, Str /** * Processes a transaction from an IOC, updating channel records in ChannelFinder. * - *

    On initial=true (first commit): Registers the IOC in the shutdown - * registry so it can be cleaned up during shutdown.

    + *

    On initial=true (first commit): Registers the IOC in the IOC registry.

    * *

    On connected=false: Immediately triggers cleanup for this IOC's channels * by calling {@link #connectionClosed(ConnectionClosedEvent)}. This allows graceful disconnect @@ -271,7 +286,7 @@ public void commit(RecSyncTransaction transaction) { String iocId = resolveIocId(transaction); - // Register this IOC in the registry for shutdown cleanup tracking. + // Register this IOC in the in-memory IOC registry. if (transaction.initial()) { if (iocId != null && !iocId.isBlank()) { Map clientInfos = transaction.clientInfos(); @@ -286,7 +301,7 @@ public void commit(RecSyncTransaction transaction) { OffsetDateTime.now().toString() ); iocRegistry.put(iocId, iocInfo); - log.info(() -> "[CFRecSyncProcessor] registered IOC in shutdown registry: " + iocId); + log.info(() -> "[CFRecSyncProcessor] registered IOC: " + iocId); } } @@ -379,25 +394,7 @@ private void cleanupDeletedRecordChannels(String iocId, Set deletedRecI return; } - Map existing = store.findByNames(channelNamesToCleanup); - if (existing.isEmpty()) { - return; - } - - if (deleteOnClean) { - store.deleteChannels(existing.keySet()); - return; - } - - String now = OffsetDateTime.now().toString(); - Collection inactiveChannels = new ArrayList<>(); - for (ChannelFinderChannel channel : existing.values()) { - Map properties = new LinkedHashMap<>(channel.properties()); - properties.put(PV_STATUS_PROPERTY, INACTIVE); - properties.put(TIME_PROPERTY, now); - inactiveChannels.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); - } - store.updateChannels(inactiveChannels); + cleanChannels(store.findByNames(channelNamesToCleanup), PvStatus.INACTIVE); } private Collection buildChannels(RecSyncTransaction transaction) { @@ -470,7 +467,7 @@ private static Map buildStandardProperties(AddRecordMessage addR Map properties = new LinkedHashMap<>(); properties.put(HOST_NAME_PROPERTY, context.hostName()); properties.put(IOC_NAME_PROPERTY, context.iocName()); - properties.put(PV_STATUS_PROPERTY, ACTIVE); + properties.put(PV_STATUS_PROPERTY, PvStatus.ACTIVE.value()); properties.put(TIME_PROPERTY, context.timestamp()); properties.put(IOC_ID_PROPERTY, context.iocId()); properties.put(IOC_IP_PROPERTY, context.iocIp()); @@ -603,65 +600,30 @@ public void connectionClosed(ConnectionClosedEvent event) { return; } - if (deleteOnClean) { - store.deleteChannels(existing.keySet()); - iocRegistry.remove(iocId); - iocChannelState.remove(iocId); - return; - } - - String now = OffsetDateTime.now().toString(); - Collection inactiveChannels = new ArrayList<>(); - for (ChannelFinderChannel channel : existing.values()) { - Map properties = new LinkedHashMap<>(channel.properties()); - properties.put(PV_STATUS_PROPERTY, INACTIVE); - properties.put(TIME_PROPERTY, now); - inactiveChannels.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); - } - store.updateChannels(inactiveChannels); + cleanChannels(existing, PvStatus.INACTIVE); iocRegistry.remove(iocId); iocChannelState.remove(iocId); } @Override public void destroy() { - System.err.println("[CFRecSyncProcessor] destroy: cleaning up tracked IOCs on shutdown"); - log.info("[CFRecSyncProcessor] destroy: cleaning up tracked IOCs on shutdown"); - try { - cleanupTrackedIocs(); - } catch (Exception e) { - System.err.println("[CFRecSyncProcessor] error during shutdown cleanup: " + e.getMessage()); - log.severe("[CFRecSyncProcessor] error during shutdown cleanup: " + e.getMessage()); - } - } - - /** - * Performs cleanup of all IOCs tracked in the iocRegistry. - * Called during application shutdown via Spring @PreDestroy lifecycle. - * If deleteOnShutdown is true, deletes all channels belonging to tracked IOCs. - * Otherwise, marks them as inactive. - */ - private void cleanupTrackedIocs() { - if (iocRegistry.isEmpty()) { - System.err.println("[CFRecSyncProcessor] no IOCs to clean up on shutdown"); - log.info("[CFRecSyncProcessor] no IOCs to clean up on shutdown"); + if (!cleanOnShutdown) { + log.info("[CFRecSyncProcessor] destroy: clean-on-shutdown disabled, leaving ChannelFinder untouched"); return; } - - String msg = "[CFRecSyncProcessor] cleaning up " + iocRegistry.size() + " tracked IOCs"; - System.err.println(msg); - log.info(msg); - - // Iterate over a copy of the registry keys to avoid ConcurrentModificationException - Set iocIds = new HashSet<>(iocRegistry.keySet()); - for (String iocId : iocIds) { - try { - cleanupIoc(iocId); - } catch (Exception e) { - String errMsg = "[CFRecSyncProcessor] error cleaning up IOC " + iocId + ": " + e.getMessage(); - System.err.println(errMsg); - log.warning(() -> errMsg); - } + try { + Map channels = store.findByProperty(RECCEIVER_ID_PROPERTY, RECCEIVER_ID); + log.info(() -> "[CFRecSyncProcessor] destroy: " + + (deleteOnClean ? "deleting " : "marking " + PvStatus.UNKNOWN + " ") + + channels.size() + " channels for recceiver " + RECCEIVER_ID); + cleanChannels(channels, PvStatus.UNKNOWN); + } catch (Exception e) { + String errMsg = "[CFRecSyncProcessor] error during shutdown cleanup: " + e.getMessage(); + System.err.println(errMsg); + log.severe(() -> errMsg); + } finally { + iocRegistry.clear(); + iocChannelState.clear(); } } @@ -669,50 +631,6 @@ protected int iocRegistrySize() { return iocRegistry.size(); } - /** - * Cleans up channels for a specific IOC. - * @param iocId the IOC ID to clean up (e.g., "host:port") - */ - private void cleanupIoc(String iocId) { - Map existing = store.findByProperty(IOC_ID_PROPERTY, iocId); - if (existing.isEmpty()) { - String msg = "[CFRecSyncProcessor] no channels found for IOC " + iocId; - System.err.println(msg); - log.info(msg); - iocRegistry.remove(iocId); - iocChannelState.remove(iocId); - return; - } - - String foundMsg = "[CFRecSyncProcessor] found " + existing.size() + " channels for IOC " + iocId; - System.err.println(foundMsg); - log.info(foundMsg); - - if (deleteOnShutdown) { - String deleteMsg = "[CFRecSyncProcessor] deleting " + existing.size() + " channels for IOC " + iocId; - System.err.println(deleteMsg); - log.info(deleteMsg); - store.deleteChannels(existing.keySet()); - } else { - String inactiveMsg = "[CFRecSyncProcessor] marking " + existing.size() + " channels as undefined for IOC " + iocId; - System.err.println(inactiveMsg); - log.info(inactiveMsg); - String now = OffsetDateTime.now().toString(); - Collection inactiveChannels = new ArrayList<>(); - for (ChannelFinderChannel channel : existing.values()) { - Map properties = new LinkedHashMap<>(channel.properties()); - properties.put(PV_STATUS_PROPERTY, UNDEFINED); - properties.put(TIME_PROPERTY, now); - inactiveChannels.add(new ChannelFinderChannel(channel.name(), channel.owner(), properties)); - } - store.updateChannels(inactiveChannels); - } - - iocRegistry.remove(iocId); - iocChannelState.remove(iocId); - } - - private static String resolveIocId(String sourceAddress) { if (sourceAddress == null || sourceAddress.isBlank()) { return null; @@ -861,10 +779,8 @@ private record ChannelContext( } /** - * Metadata about a connected IOC, tracked for shutdown cleanup. + * Metadata about a connected IOC, kept in the in-memory IOC registry. * - *

    Allows the processor to know which IOCs to clean up when the application shuts down, - * even if some IOCs are not currently connected.

    * *

    Fields include source address, host, port, IOC name, host name, owner, and connection timestamp.

    */ diff --git a/src/main/java/org/phoebus/recceiver/spi/cf/PvStatus.java b/src/main/java/org/phoebus/recceiver/spi/cf/PvStatus.java new file mode 100644 index 0000000..f1b6dca --- /dev/null +++ b/src/main/java/org/phoebus/recceiver/spi/cf/PvStatus.java @@ -0,0 +1,29 @@ +package org.phoebus.recceiver.spi.cf; + +/** + * Values recceiver writes to a channel's {@code pvStatus} property in ChannelFinder. + */ +public enum PvStatus { + /** The IOC publishing the channel is connected. */ + ACTIVE("Active"), + /** The IOC disconnected, the record was removed, or recceiver cleaned up on startup. */ + INACTIVE("Inactive"), + /** Recceiver shut down; the IOC's actual state is not known. */ + UNKNOWN("Unknown"); + + private final String value; + + PvStatus(String value) { + this.value = value; + } + + /** @return the string stored in ChannelFinder for this status. */ + public String value() { + return value; + } + + @Override + public String toString() { + return value; + } +} \ No newline at end of file diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index b5a42bf..f272862 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -66,15 +66,21 @@ recceiver.cf.service-url=http://localhost:9090/ChannelFinder recceiver.cf.username=admin recceiver.cf.password=adminPass -# On connection close: true => delete IOC channels, false => mark IOC channels Inactive. +# For every cleanup and IOC disconnection: +# true => delete channels, false => keep channels and update their pvStatus. recceiver.cf.delete-on-clean=false -# On processor init/startup for this recceiver's currently Active channels: -# true => delete channels, false => mark channels Inactive. +# On startup, clean this recceiver's channels: +# delete-on-clean=true => delete all of them +# delete-on-clean=false => set pvStatus=Inactive +# false => startup does not touch ChannelFinder recceiver.cf.clean-on-startup=false -# On application shutdown: true => delete IOC channels (default), false => mark as Inactive. -recceiver.cf.delete-on-shutdown=false +# On application shutdown, clean this recceiver's channels: +# delete-on-clean=true => delete all of them (any pvStatus) +# delete-on-clean=false => set pvStatus=Unknown +# false => shutdown does not touch ChannelFinder. +recceiver.cf.clean-on-shutdown=true # Publish alias records as channels with property alias=. recceiver.cf.alias-enabled=true diff --git a/src/test/java/org/phoebus/recceiver/CFRecSyncProcessorTest.java b/src/test/java/org/phoebus/recceiver/CFRecSyncProcessorTest.java index 7aa7ad8..ceaffa1 100644 --- a/src/test/java/org/phoebus/recceiver/CFRecSyncProcessorTest.java +++ b/src/test/java/org/phoebus/recceiver/CFRecSyncProcessorTest.java @@ -9,6 +9,7 @@ import org.phoebus.recceiver.spi.cf.CFRecSyncProcessor; import org.phoebus.recceiver.spi.cf.ChannelFinderChannel; import org.phoebus.recceiver.spi.cf.ChannelFinderStore; +import org.phoebus.recceiver.spi.cf.PvStatus; import java.util.ArrayList; import java.util.Collection; @@ -20,6 +21,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -134,58 +136,90 @@ void initEnsuresEnvironmentVariableMappedProperties() { assertEquals(13, store.ensuredProperties.size()); } - @Test - void initCleanOnStartupFalseMarksActiveChannelsInactive() { - TestStore store = new TestStore(); - store.findByPropertyResult = Map.of( + private static Map activeAndInactiveChannels() { + return Map.of( "pv:active", new ChannelFinderChannel("pv:active", "owner1", Map.of( - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE, + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value(), CFRecSyncProcessor.TIME_PROPERTY, "old-time" )), "pv:inactive", new ChannelFinderChannel("pv:inactive", "owner1", Map.of( - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.INACTIVE, + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.INACTIVE.value(), CFRecSyncProcessor.TIME_PROPERTY, "old-time" )) ); + } + + @Test + void initDoesNothingWhenCleanOnStartupFalse() { + TestStore store = new TestStore(); + store.findByPropertyResult = activeAndInactiveChannels(); TestProcessor processor = new TestProcessor(store); RecSyncProcessor initialized = processor.init("cf", Map.of( "recceiver.cf.enabled", "true", "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", - "recceiver.cf.clean-on-startup", "false" + "recceiver.cf.clean-on-startup", "false", + "recceiver.cf.delete-on-clean", "true" + )); + + assertNotNull(initialized); + assertEquals(0, store.updatedChannels.size()); + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void initDoesNothingByDefault() { + TestStore store = new TestStore(); + store.findByPropertyResult = activeAndInactiveChannels(); + TestProcessor processor = new TestProcessor(store); + + RecSyncProcessor initialized = processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder" + )); + + assertNotNull(initialized); + assertEquals(0, store.updatedChannels.size()); + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void initCleanOnStartupMarksActiveChannelsInactiveWhenDeleteOnCleanFalse() { + TestStore store = new TestStore(); + store.findByPropertyResult = activeAndInactiveChannels(); + TestProcessor processor = new TestProcessor(store); + + RecSyncProcessor initialized = processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-startup", "true", + "recceiver.cf.delete-on-clean", "false" )); assertNotNull(initialized); assertEquals(1, store.updatedChannels.size()); assertEquals("pv:active", store.updatedChannels.get(0).name()); - assertEquals(CFRecSyncProcessor.INACTIVE, + assertEquals(PvStatus.INACTIVE.value(), store.updatedChannels.get(0).property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertEquals(0, store.deletedChannelNames.size()); } @Test - void initCleanOnStartupTrueDeletesActiveChannels() { + void initCleanOnStartupDeletesAllRecceiverChannelsWhenDeleteOnCleanTrue() { TestStore store = new TestStore(); - store.findByPropertyResult = Map.of( - "pv:active", new ChannelFinderChannel("pv:active", "owner1", Map.of( - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE - )), - "pv:inactive", new ChannelFinderChannel("pv:inactive", "owner1", Map.of( - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.INACTIVE - )) - ); + store.findByPropertyResult = activeAndInactiveChannels(); TestProcessor processor = new TestProcessor(store); RecSyncProcessor initialized = processor.init("cf", Map.of( "recceiver.cf.enabled", "true", "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", - "recceiver.cf.clean-on-startup", "true" + "recceiver.cf.clean-on-startup", "true", + "recceiver.cf.delete-on-clean", "true" )); assertNotNull(initialized); assertEquals(0, store.updatedChannels.size()); - assertEquals(1, store.deletedChannelNames.size()); - assertTrue(store.deletedChannelNames.contains("pv:active")); + assertEquals(Set.of("pv:active", "pv:inactive"), Set.copyOf(store.deletedChannelNames)); } @Test @@ -246,7 +280,7 @@ void commitPublishesStandardPropertiesFromTransaction() { assertEquals("alice", channel.owner()); assertEquals("host-a", channel.property(CFRecSyncProcessor.HOST_NAME_PROPERTY)); assertEquals("ioc-a", channel.property(CFRecSyncProcessor.IOC_NAME_PROPERTY)); - assertEquals(CFRecSyncProcessor.ACTIVE, channel.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertEquals(PvStatus.ACTIVE.value(), channel.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertEquals("127.0.0.1:5070", channel.property(CFRecSyncProcessor.IOC_ID_PROPERTY)); assertEquals("127.0.0.1", channel.property(CFRecSyncProcessor.IOC_IP_PROPERTY)); assertEquals("ai", channel.property(CFRecSyncProcessor.RECORD_TYPE_PROPERTY)); @@ -303,7 +337,7 @@ void commitPublishesAliasChannelsWhenEnabled() { assertEquals("host-a", alias.property(CFRecSyncProcessor.HOST_NAME_PROPERTY)); assertEquals("ioc-a", alias.property(CFRecSyncProcessor.IOC_NAME_PROPERTY)); assertEquals("127.0.0.1:5070", alias.property(CFRecSyncProcessor.IOC_ID_PROPERTY)); - assertEquals(CFRecSyncProcessor.ACTIVE, alias.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertEquals(PvStatus.ACTIVE.value(), alias.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertEquals("ai", alias.property(CFRecSyncProcessor.RECORD_TYPE_PROPERTY)); } @@ -413,25 +447,31 @@ void commitPublishesMappedEnvironmentVariableProperties() { } @Test - void initParsesDeleteOnShutdownConfiguration() { + void initIgnoresRemovedDeleteOnShutdownOption() { TestStore store = new TestStore(); + store.findByPropertyResult = Map.of( + "pv:one", new ChannelFinderChannel("pv:one", "owner1", Map.of( + CFRecSyncProcessor.IOC_ID_PROPERTY, "127.0.0.1:5070", + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value() + )) + ); TestProcessor processor = new TestProcessor(store); - // Test default value (true) RecSyncProcessor initialized = processor.init("cf", Map.of( - "recceiver.cf.enabled", "true", - "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder" - )); - assertNotNull(initialized); - - // Test explicit false value - TestProcessor processorFalse = new TestProcessor(store); - initialized = processorFalse.init("cf", Map.of( "recceiver.cf.enabled", "true", "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", - "recceiver.cf.delete-on-shutdown", "false" + "recceiver.cf.delete-on-shutdown", "true" )); assertNotNull(initialized); + + processor.commit(initialTransactionFor("pv:one")); + store.updatedChannels.clear(); + processor.destroy(); + + assertEquals(0, store.deletedChannelNames.size()); + assertEquals(1, store.updatedChannels.size()); + assertEquals(PvStatus.UNKNOWN.value(), + store.updatedChannels.getFirst().property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); } @Test @@ -462,91 +502,243 @@ void registersIocInShutdownRegistryOnInitialCommit() { assertEquals(1, processor.getIocRegistrySize()); } - @Test - void destroyMarksChannelsUndefinedWhenDeleteOnShutdownFalse() { + private static RecSyncTransaction initialTransactionFor(String recordName) { + return new RecSyncTransaction( + "/127.0.0.1:5070", + List.of(new AddRecordMessage(10, 0, 3, 6, "ai", recordName)), + List.of(), + List.of(), + Map.of( + CFRecSyncProcessor.IOCNAME_KEY, "ioc-a", + CFRecSyncProcessor.HOSTNAME_KEY, "host-a" + ), + true, + true + ); + } + + private static TestStore storeWithOneActiveIocChannel() { TestStore store = new TestStore(); store.findByPropertyResult = Map.of( "pv:one", new ChannelFinderChannel("pv:one", "owner1", Map.of( CFRecSyncProcessor.IOC_ID_PROPERTY, "127.0.0.1:5070", - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value() )) ); + return store; + } + @Test + void destroyDoesNothingWhenCleanOnShutdownFalse() { + TestStore store = storeWithOneActiveIocChannel(); TestProcessor processor = new TestProcessor(store); processor.init("cf", Map.of( "recceiver.cf.enabled", "true", "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", - "recceiver.cf.delete-on-shutdown", "false" + "recceiver.cf.clean-on-shutdown", "false", + "recceiver.cf.delete-on-clean", "true" )); + processor.commit(initialTransactionFor("pv:one")); + store.updatedChannels.clear(); - // Register an IOC - RecSyncTransaction transaction = new RecSyncTransaction( - "/127.0.0.1:5070", - List.of(new AddRecordMessage(10, 0, 3, 6, "ai", "pv:one")), - List.of(), - List.of(), - Map.of( - CFRecSyncProcessor.IOCNAME_KEY, "ioc-a", - CFRecSyncProcessor.HOSTNAME_KEY, "host-a" - ), - true, - true - ); - processor.commit(transaction); + processor.destroy(); + assertEquals(0, store.updatedChannels.size()); + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void destroyMarksChannelsUnknownByDefault() { + TestStore store = storeWithOneActiveIocChannel(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder" + )); + processor.commit(initialTransactionFor("pv:one")); + store.updatedChannels.clear(); + + processor.destroy(); + + assertEquals(1, store.updatedChannels.size()); + assertEquals(PvStatus.UNKNOWN.value(), + store.updatedChannels.getFirst().property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void destroyMarksChannelsUnknownWhenCleanOnShutdownAndDeleteOnCleanFalse() { + TestStore store = storeWithOneActiveIocChannel(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-shutdown", "true", + "recceiver.cf.delete-on-clean", "false" + )); + processor.commit(initialTransactionFor("pv:one")); store.updatedChannels.clear(); - // Destroy should mark channels as undefined, not delete them processor.destroy(); - // Verify channels were updated (not deleted) assertEquals(1, store.updatedChannels.size()); ChannelFinderChannel updated = store.updatedChannels.get(0); - assertEquals(CFRecSyncProcessor.UNDEFINED, updated.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertEquals(PvStatus.UNKNOWN.value(), updated.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertEquals(0, store.deletedChannelNames.size()); + assertEquals(0, processor.getIocRegistrySize()); } @Test - void destroyDeletesChannelsWhenDeleteOnShutdownTrue() { - TestStore store = new TestStore(); - store.findByPropertyResult = Map.of( - "pv:one", new ChannelFinderChannel("pv:one", "owner1", Map.of( - CFRecSyncProcessor.IOC_ID_PROPERTY, "127.0.0.1:5070", - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE + void destroyDeletesChannelsWhenCleanOnShutdownAndDeleteOnCleanTrue() { + TestStore store = storeWithOneActiveIocChannel(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-shutdown", "true", + "recceiver.cf.delete-on-clean", "true" + )); + processor.commit(initialTransactionFor("pv:one")); + store.updatedChannels.clear(); + + processor.destroy(); + + assertEquals(0, store.updatedChannels.size()); + assertEquals(1, store.deletedChannelNames.size()); + assertTrue(store.deletedChannelNames.contains("pv:one")); + assertEquals(0, processor.getIocRegistrySize()); + } + + private static Map channelsInEveryStatus() { + return Map.of( + "pv:active", new ChannelFinderChannel("pv:active", "owner1", Map.of( + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value(), + CFRecSyncProcessor.TIME_PROPERTY, "old-time" + )), + "pv:inactive", new ChannelFinderChannel("pv:inactive", "owner1", Map.of( + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.INACTIVE.value(), + CFRecSyncProcessor.TIME_PROPERTY, "old-time" + )), + "pv:unknown", new ChannelFinderChannel("pv:unknown", "owner1", Map.of( + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.UNKNOWN.value(), + CFRecSyncProcessor.TIME_PROPERTY, "old-time" + )), + "pv:custom", new ChannelFinderChannel("pv:custom", "owner1", Map.of( + CFRecSyncProcessor.PV_STATUS_PROPERTY, "Custom", + CFRecSyncProcessor.TIME_PROPERTY, "old-time" )) ); + } + + private static Map byName(List channels) { + Map result = new LinkedHashMap<>(); + for (ChannelFinderChannel channel : channels) { + result.put(channel.name(), channel); + } + return result; + } + @Test + void initCleanOnStartupMarksEveryChannelNotAlreadyInactiveAsInactive() { + TestStore store = new TestStore(); + store.findByPropertyResult = channelsInEveryStatus(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-startup", "true", + "recceiver.cf.delete-on-clean", "false" + )); + + Map updated = byName(store.updatedChannels); + // Every status (including Unknown, unexpected or missing) becomes Inactive; + // channels already Inactive are not rewritten. + assertEquals(Set.of("pv:active", "pv:unknown", "pv:custom"), updated.keySet()); + for (ChannelFinderChannel channel : updated.values()) { + assertEquals(PvStatus.INACTIVE.value(), channel.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + } + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void destroyCleansRecceiverChannelsEvenWhenNoIocTrackedInMemory() { + // e.g. channels left behind by a previous run that crashed + TestStore store = new TestStore(); + store.findByPropertyResult = activeAndInactiveChannels(); TestProcessor processor = new TestProcessor(store); processor.init("cf", Map.of( "recceiver.cf.enabled", "true", "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", - "recceiver.cf.delete-on-shutdown", "true" + "recceiver.cf.clean-on-shutdown", "true" )); + assertEquals(0, processor.getIocRegistrySize()); - // Register an IOC - RecSyncTransaction transaction = new RecSyncTransaction( - "/127.0.0.1:5070", - List.of(new AddRecordMessage(10, 0, 3, 6, "ai", "pv:one")), - List.of(), - List.of(), - Map.of( - CFRecSyncProcessor.IOCNAME_KEY, "ioc-a", - CFRecSyncProcessor.HOSTNAME_KEY, "host-a" - ), - true, - true - ); - processor.commit(transaction); + processor.destroy(); - store.updatedChannels.clear(); + assertEquals(2, store.updatedChannels.size()); + } + + @Test + void destroyMarksEveryChannelNotAlreadyUnknownAsUnknownWhenDeleteOnCleanFalse() { + TestStore store = new TestStore(); + store.findByPropertyResult = channelsInEveryStatus(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-shutdown", "true", + "recceiver.cf.delete-on-clean", "false" + )); + + processor.destroy(); + + Map updatedByName = byName(store.updatedChannels); + // Every status (including unexpected or missing ones) becomes Unknown; + + + // channels already Unknown are not rewritten. + assertEquals(Set.of("pv:active", "pv:inactive", "pv:custom"), updatedByName.keySet()); + for (ChannelFinderChannel channel : updatedByName.values()) { + assertEquals(PvStatus.UNKNOWN.value(), channel.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertNotEquals("old-time", channel.property(CFRecSyncProcessor.TIME_PROPERTY)); + } + assertEquals(0, store.deletedChannelNames.size()); + } + + @Test + void destroyDeletesAllRecceiverChannelsWhenDeleteOnCleanTrue() { + TestStore store = new TestStore(); + store.findByPropertyResult = channelsInEveryStatus(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-shutdown", "true", + "recceiver.cf.delete-on-clean", "true" + )); - // Destroy should delete channels processor.destroy(); - // Verify channels were deleted, not updated + assertEquals(Set.of("pv:active", "pv:inactive", "pv:unknown", "pv:custom"), + Set.copyOf(store.deletedChannelNames)); assertEquals(0, store.updatedChannels.size()); - assertEquals(1, store.deletedChannelNames.size()); - assertTrue(store.deletedChannelNames.contains("pv:one")); + } + + @Test + void destroyQueriesByRecceiverId() { + TestStore store = storeWithOneActiveIocChannel(); + TestProcessor processor = new TestProcessor(store); + processor.init("cf", Map.of( + "recceiver.cf.enabled", "true", + "recceiver.cf.service-url", "http://localhost:9090/ChannelFinder", + "recceiver.cf.clean-on-shutdown", "true" + )); + processor.commit(initialTransactionFor("pv:one")); + store.findByPropertyNames.clear(); + + processor.destroy(); + + assertEquals(List.of(CFRecSyncProcessor.RECCEIVER_ID_PROPERTY), store.findByPropertyNames); } @Test @@ -629,7 +821,7 @@ void commitMarksRecordInactiveWhenDeletedAndDeleteOnCleanFalse() { store.updatedChannels.clear(); store.findByNamesResult = Map.of( "pv:one", new ChannelFinderChannel("pv:one", "owner", Map.of( - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE, + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value(), CFRecSyncProcessor.TIME_PROPERTY, "old-time" )) ); @@ -648,7 +840,7 @@ void commitMarksRecordInactiveWhenDeletedAndDeleteOnCleanFalse() { assertEquals(1, store.updatedChannels.size()); ChannelFinderChannel updated = store.updatedChannels.get(0); assertEquals("pv:one", updated.name()); - assertEquals(CFRecSyncProcessor.INACTIVE, updated.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); + assertEquals(PvStatus.INACTIVE.value(), updated.property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertNotNull(updated.property(CFRecSyncProcessor.TIME_PROPERTY)); assertEquals(0, store.deletedChannelNames.size()); } @@ -659,7 +851,7 @@ void commitConnectedFalseMarksIocChannelsInactiveWhenDeleteOnCleanFalse() { store.findByPropertyResult = Map.of( "pv:one", new ChannelFinderChannel("pv:one", "owner1", Map.of( CFRecSyncProcessor.IOC_ID_PROPERTY, "127.0.0.1:5070", - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value() )) ); TestProcessor processor = new TestProcessor(store); @@ -684,7 +876,7 @@ void commitConnectedFalseMarksIocChannelsInactiveWhenDeleteOnCleanFalse() { processor.commit(disconnected); assertEquals(1, store.updatedChannels.size()); - assertEquals(CFRecSyncProcessor.INACTIVE, + assertEquals(PvStatus.INACTIVE.value(), store.updatedChannels.get(0).property(CFRecSyncProcessor.PV_STATUS_PROPERTY)); assertEquals(0, store.deletedChannelNames.size()); assertEquals(0, processor.getIocRegistrySize()); @@ -696,7 +888,7 @@ void commitConnectedFalseDeletesIocChannelsWhenDeleteOnCleanTrue() { store.findByPropertyResult = Map.of( "pv:one", new ChannelFinderChannel("pv:one", "owner1", Map.of( CFRecSyncProcessor.IOC_ID_PROPERTY, "127.0.0.1:5070", - CFRecSyncProcessor.PV_STATUS_PROPERTY, CFRecSyncProcessor.ACTIVE + CFRecSyncProcessor.PV_STATUS_PROPERTY, PvStatus.ACTIVE.value() )) ); TestProcessor processor = new TestProcessor(store); @@ -751,6 +943,7 @@ private static final class TestStore implements ChannelFinderStore { private final List deletedChannelNames = new ArrayList<>(); private Map findByNamesResult = Map.of(); private Map findByPropertyResult = Map.of(); + private final List findByPropertyNames = new ArrayList<>(); @Override public Map findByNames(Set channelNames) { @@ -766,6 +959,7 @@ public Map findByNames(Set channelNames) { @Override public Map findByProperty(String propertyName, String propertyValue) { + findByPropertyNames.add(propertyName); return findByPropertyResult; }