demus/openhab-core
1
0
Fork 0
mirror of https://github.com/danieldemus/openhab-core.git synced 2026-07-29 12:34:22 +02:00
Code Issues 1 Packages Projects Releases Wiki Activity

Clean retry queue when a thing is updated/removed in YAML thing provider (#5361)

Browse Source 
* [YAML thing provider] Clean retry queue when a thing is updated/removed

Signed-off-by: Laurent Garnier <lg.hc@free.fr>
This commit is contained in:
lolodomo
2026-05-29 22:16:38 +02:00
committed by GitHub
parent 4fbee78009
commit 64cb8b8145
2 changed files with 257 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bundles/org.openhab.core.model.yaml/src/main/java/org/openhab/core/model/yaml/internal/things/YamlThingProvider.java
+56 -15
View File
@@ -102,30 +102,38 @@ public class YamlThingProvider extends AbstractProvider<Thing>
private final Map<String, Collection<Thing>> thingsMap = new ConcurrentHashMap<>();
private final List<QueueContent> queue = new CopyOnWriteArrayList<>();
private final Object queueLock = new Object();
private final Runnable lazyRetryRunnable = new Runnable() {
@Override
public void run() {
threadStopping = false;
logger.debug("Starting lazy retry thread");
while (!queue.isEmpty()) {
for (QueueContent qc : queue) {
if (retryCreateThing(qc.thingHandlerFactory, qc.thingTypeUID, qc.configuration, qc.thingUID,
qc.bridgeUID)) {
queue.remove(qc);
boolean stop = false;
do {
// Wait 1s before retrying
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
stop = true;
threadStopping = true;
}
if (!stop) {
synchronized (queueLock) {
queue.removeIf(qc -> retryCreateThing(qc.thingHandlerFactory, qc.thingTypeUID, qc.configuration,
qc.thingUID, qc.bridgeUID));
stop = queue.isEmpty();
threadStopping = stop == true;
}
}
if (!queue.isEmpty()) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
}
}
} while (!stop);
logger.debug("Lazy retry thread ran out of work. Good bye.");
}
};
private @Nullable Thread lazyRetryThread;
protected @Nullable Thread lazyRetryThread;
private boolean threadStopping;
private record QueueContent(ThingHandlerFactory thingHandlerFactory, ThingTypeUID thingTypeUID,
Configuration configuration, ThingUID thingUID, @Nullable ThingUID bridgeUID) {
@@ -146,7 +154,9 @@ public class YamlThingProvider extends AbstractProvider<Thing>
@Deactivate
public void deactivate() {
queue.clear();
synchronized (queueLock) {
queue.clear();
}
thingsMap.clear();
loadedXmlThingTypes.clear();
}
@@ -196,6 +206,11 @@ public class YamlThingProvider extends AbstractProvider<Thing>
@Override
public void updatedModel(String modelName, Collection<YamlThingDTO> elements) {
boolean isolated = isIsolatedModel(modelName);
if (!isolated) {
elements.stream().map(this::buildThingUID).filter(Objects::nonNull).forEach(uid -> {
removeFromRetryQueue(uid);
});
}
List<Thing> updated = elements.stream().map(t -> mapThing(t, isolated)).filter(Objects::nonNull).toList();
Collection<Thing> modelThings = Objects
.requireNonNull(thingsMap.computeIfAbsent(modelName, k -> new ArrayList<>()));
@@ -222,6 +237,9 @@ public class YamlThingProvider extends AbstractProvider<Thing>
boolean isolated = isIsolatedModel(modelName);
Collection<Thing> modelThings = thingsMap.getOrDefault(modelName, new ArrayList<>());
elements.stream().map(this::buildThingUID).filter(Objects::nonNull).forEach(uid -> {
if (!isolated) {
removeFromRetryQueue(uid);
}
modelThings.stream().filter(th -> th.getUID().equals(uid)).findFirst().ifPresentOrElse(oldThing -> {
modelThings.remove(oldThing);
logger.debug("model {} removed thing {}", modelName, uid);
@@ -539,8 +557,21 @@ public class YamlThingProvider extends AbstractProvider<Thing>
private void queueRetryThingCreation(ThingHandlerFactory handlerFactory, ThingTypeUID thingTypeUID,
Configuration configuration, ThingUID thingUID, @Nullable ThingUID bridgeUID) {
queue.add(new QueueContent(handlerFactory, thingTypeUID, configuration, thingUID, bridgeUID));
synchronized (queueLock) {
queue.add(new QueueContent(handlerFactory, thingTypeUID, configuration, thingUID, bridgeUID));
}
Thread thread = lazyRetryThread;
boolean stopping = threadStopping;
if (thread != null && thread.isAlive() && stopping) {
// Wait for the thread to terminate
while (thread.isAlive()) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
if (thread == null || !thread.isAlive()) {
thread = new Thread(lazyRetryRunnable);
lazyRetryThread = thread;
@@ -548,6 +579,16 @@ public class YamlThingProvider extends AbstractProvider<Thing>
}
}
private void removeFromRetryQueue(ThingUID thingUID) {
synchronized (queueLock) {
queue.removeIf(qc -> thingUID.equals(qc.thingUID));
}
}
public int getRetryQueueSize() {
return queue.size();
}
private Configuration processThingConfiguration(ThingTypeUID thingTypeUID, ThingUID thingUID,
Configuration configuration) {
Set<String> thingStringParams = !configuration.keySet().isEmpty()
bundles/org.openhab.core.model.yaml/src/test/java/org/openhab/core/model/yaml/internal/things/YamlThingProviderTest.java
+201 -2
View File
@@ -27,6 +27,7 @@ import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@@ -145,6 +146,7 @@ public class YamlThingProviderTest {
private @NonNullByDefault({}) YamlModelRepositoryImpl modelRepository;
private @NonNullByDefault({}) YamlThingProvider thingProvider;
private @NonNullByDefault({}) TestThingChangeListener thingListener;
private @NonNullByDefault({}) NtpThingHandlerFactory thingHandlerFactory;
@BeforeEach
public void setup() {
@@ -152,7 +154,7 @@ public class YamlThingProviderTest {
when(watchServiceMock.getWatchPath()).thenReturn(watchPath);
when(bundleResolver.resolveBundle(any())).thenReturn(bundle);
when(bundle.getSymbolicName()).thenReturn("org.openhab.binding.ntp");
when(bundle.getSymbolicName()).thenReturn(NTP_BUNDLE_SYMBOLIC_NAME);
URI uriThingType = null;
URI uriChannelType = null;
@@ -229,7 +231,7 @@ public class YamlThingProviderTest {
when(localeProvider.getLocale()).thenReturn(Locale.FRENCH);
NtpThingHandlerFactory thingHandlerFactory = new NtpThingHandlerFactory();
thingHandlerFactory = new NtpThingHandlerFactory(0);
thingProvider = new YamlThingProvider(bundleResolver, thingTypeRegistry, channelTypeRegistry,
configDescriptionRegistry, localeProvider);
thingProvider.onReadyMarkerAdded(new ReadyMarker(XML_THING_TYPE, NTP_BUNDLE_SYMBOLIC_NAME));
@@ -307,6 +309,9 @@ public class YamlThingProviderTest {
assertEquals("dd-MM-yyyy", channel.getConfiguration().get("DateTimeFormat"));
assertEquals("A parameter that is not in the channel config description.",
channel.getConfiguration().get("other"));
assertNull(thingProvider.lazyRetryThread);
assertEquals(0, thingProvider.getRetryQueueSize());
}
@Test
@@ -370,6 +375,9 @@ public class YamlThingProviderTest {
// Parameter DateTimeFormat injected with default value
assertThat(channel.getConfiguration().keySet(), contains("DateTimeFormat"));
assertEquals(DEFAULT_DATE_TIME_FORMAT, channel.getConfiguration().get("DateTimeFormat"));
assertNull(thingProvider.lazyRetryThread);
assertEquals(0, thingProvider.getRetryQueueSize());
}
@Test
@@ -439,6 +447,9 @@ public class YamlThingProviderTest {
assertEquals("dd-MM-yyyy", channel.getConfiguration().get("DateTimeFormat"));
assertEquals("A parameter that is not in the channel config description.",
channel.getConfiguration().get("other"));
assertNull(thingProvider.lazyRetryThread);
assertEquals(0, thingProvider.getRetryQueueSize());
}
}
@@ -497,6 +508,9 @@ public class YamlThingProviderTest {
assertEquals(0, channel.getProperties().size());
// default value not injected for parameter DateTimeFormat
assertThat(channel.getConfiguration().keySet(), hasSize(0));
assertNull(thingProvider.lazyRetryThread);
assertEquals(0, thingProvider.getRetryQueueSize());
}
}
@@ -527,7 +541,189 @@ public class YamlThingProviderTest {
assertEquals("123.45", channel.getConfiguration().get("DateTimeFormat"));
}
@Test
public void testUsageOfRetryJob() throws IOException {
thingProvider.removeThingHandlerFactory(thingHandlerFactory);
thingHandlerFactory = new NtpThingHandlerFactory(2);
thingProvider.addThingHandlerFactory(thingHandlerFactory);
Files.copy(SOURCE_PATH.resolve("thing.yaml"), fullModelPath);
modelRepository.processWatchEvent(WatchService.Kind.CREATE, fullModelPath);
assertEquals(1, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(1, thingProvider.getRetryQueueSize());
assertNotNull(thingProvider.lazyRetryThread);
assertTrue(thingProvider.lazyRetryThread.isAlive());
try {
Thread.sleep(2500);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
fail("Test interrupted while waiting for retry job to complete", e);
}
assertEquals(3, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(0, thingProvider.getRetryQueueSize());
assertNotNull(thingProvider.lazyRetryThread);
assertFalse(thingProvider.lazyRetryThread.isAlive());
assertFalse(YamlModelUtils.isIsolatedModel(MODEL_NAME));
assertThat(thingListener.things, is(aMapWithSize(1)));
assertThat(thingListener.things, hasKey("ntp:ntp:local"));
assertThat(thingProvider.getAllFromModel(MODEL_NAME), hasSize(1));
Collection<Thing> things = thingProvider.getAll();
assertThat(things, hasSize(1));
Thing thing = things.iterator().next();
assertFalse(thing instanceof Bridge);
assertEquals(NTP_THING_UID, thing.getUID());
assertNull(thing.getBridgeUID());
assertEquals(NTP_THING_TYPE_UID, thing.getThingTypeUID());
assertEquals("NTP Local Server", thing.getLabel());
assertEquals("Paris", thing.getLocation());
assertEquals(NTP_THING_TYPE_EQUIPMENT_TAG, thing.getSemanticEquipmentTag());
assertEquals(0, thing.getProperties().size());
// 2 parameters injected with default value
assertThat(thing.getConfiguration().keySet(),
containsInAnyOrder("hostname", "refreshInterval", "refreshNtp", "serverPort", "timeZone", "other"));
}
@Test
public void testRemoveFromRetryQueueWhenCalledFromUpdatedModel() throws IOException {
thingProvider.removeThingHandlerFactory(thingHandlerFactory);
thingHandlerFactory = new NtpThingHandlerFactory(1);
thingProvider.addThingHandlerFactory(thingHandlerFactory);
Files.copy(SOURCE_PATH.resolve("thing.yaml"), fullModelPath);
modelRepository.processWatchEvent(WatchService.Kind.CREATE, fullModelPath);
assertEquals(1, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(1, thingProvider.getRetryQueueSize());
assertNotNull(thingProvider.lazyRetryThread);
assertTrue(thingProvider.lazyRetryThread.isAlive());
Files.copy(SOURCE_PATH.resolve("thingWithEmptyConfig.yaml"), fullModelPath,
StandardCopyOption.REPLACE_EXISTING);
modelRepository.processWatchEvent(WatchService.Kind.MODIFY, fullModelPath);
assertEquals(2, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(0, thingProvider.getRetryQueueSize());
// Wait to be sure the job is terminated (it sleeps 1s)
try {
Thread.sleep(1200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
fail("Test interrupted while waiting for retry job to complete", e);
}
assertEquals(2, thingHandlerFactory.getNbCallsToCreateThing());
assertNotNull(thingProvider.lazyRetryThread);
assertFalse(thingProvider.lazyRetryThread.isAlive());
assertFalse(YamlModelUtils.isIsolatedModel(MODEL_NAME));
assertThat(thingListener.things, is(aMapWithSize(1)));
assertThat(thingListener.things, hasKey("ntp:ntp:local"));
assertThat(thingProvider.getAllFromModel(MODEL_NAME), hasSize(1));
Collection<Thing> things = thingProvider.getAll();
assertThat(things, hasSize(1));
Thing thing = things.iterator().next();
assertFalse(thing instanceof Bridge);
assertEquals(NTP_THING_UID, thing.getUID());
assertNull(thing.getBridgeUID());
assertEquals(NTP_THING_TYPE_UID, thing.getThingTypeUID());
assertEquals(NTP_THING_TYPE_LABEL, thing.getLabel());
assertNull(thing.getLocation());
assertEquals(NTP_THING_TYPE_EQUIPMENT_TAG, thing.getSemanticEquipmentTag());
assertEquals(0, thing.getProperties().size());
// 4 parameters injected with default value
assertThat(thing.getConfiguration().keySet(),
containsInAnyOrder("hostname", "refreshInterval", "refreshNtp", "serverPort"));
}
@Test
public void testRemoveFromRetryQueueWhenCalledFromRemovedModel() throws IOException {
thingProvider.removeThingHandlerFactory(thingHandlerFactory);
thingHandlerFactory = new NtpThingHandlerFactory(5);
thingProvider.addThingHandlerFactory(thingHandlerFactory);
Files.copy(SOURCE_PATH.resolve("thing.yaml"), fullModelPath);
modelRepository.processWatchEvent(WatchService.Kind.CREATE, fullModelPath);
assertEquals(1, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(1, thingProvider.getRetryQueueSize());
assertNotNull(thingProvider.lazyRetryThread);
assertTrue(thingProvider.lazyRetryThread.isAlive());
Files.delete(fullModelPath);
modelRepository.processWatchEvent(WatchService.Kind.DELETE, fullModelPath);
assertEquals(1, thingHandlerFactory.getNbCallsToCreateThing());
assertEquals(0, thingProvider.getRetryQueueSize());
// Wait to be sure the job is terminated (it sleeps 1s)
try {
Thread.sleep(1200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
fail("Test interrupted while waiting for retry job to complete", e);
}
assertEquals(1, thingHandlerFactory.getNbCallsToCreateThing());
assertNotNull(thingProvider.lazyRetryThread);
assertFalse(thingProvider.lazyRetryThread.isAlive());
assertThat(thingProvider.getAllFromModel(MODEL_NAME), hasSize(0));
assertThat(thingProvider.getAll(), hasSize(0));
}
@Test
public void testCreateIsolatedModelNotStartingRetryJob() throws IOException {
thingProvider.removeThingHandlerFactory(thingHandlerFactory);
thingHandlerFactory = new NtpThingHandlerFactory(4);
thingProvider.addThingHandlerFactory(thingHandlerFactory);
Files.copy(SOURCE_PATH.resolve("thingWithEmptyConfig.yaml"), fullModelPath);
List<String> errors = new ArrayList<>();
List<String> warnings = new ArrayList<>();
try (FileInputStream inputStream = new FileInputStream(fullModelPath.toFile())) {
String name = modelRepository.createIsolatedModel(inputStream, errors, warnings);
assertNotNull(name);
assertEquals(0, errors.size());
assertEquals(0, warnings.size());
assertTrue(YamlModelUtils.isIsolatedModel(name));
assertThat(thingListener.things, is(aMapWithSize(0)));
assertThat(thingProvider.getAll(), hasSize(0)); // No thing for the registry
Collection<Thing> things = thingProvider.getAllFromModel(name);
assertThat(things, hasSize(1));
Thing thing = things.iterator().next();
assertFalse(thing instanceof Bridge);
assertEquals(NTP_THING_UID, thing.getUID());
assertNull(thing.getBridgeUID());
assertEquals(NTP_THING_TYPE_UID, thing.getThingTypeUID());
assertEquals(NTP_THING_TYPE_LABEL, thing.getLabel());
assertNull(thing.getLocation());
assertNull(thing.getSemanticEquipmentTag()); // Not created by the thing handler factory
assertEquals(0, thing.getProperties().size());
assertThat(thing.getConfiguration().keySet(), hasSize(0));
assertNull(thingProvider.lazyRetryThread);
assertEquals(0, thingProvider.getRetryQueueSize());
}
}
private class NtpThingHandlerFactory implements ThingHandlerFactory {
private int nbCallsBeforeSucceededCreatingThing;
private int nbCallsToCreateThing = 0;
public NtpThingHandlerFactory(int nbCallsBeforeSucceededCreatingThing) {
this.nbCallsBeforeSucceededCreatingThing = nbCallsBeforeSucceededCreatingThing;
}
public int getNbCallsToCreateThing() {
return nbCallsToCreateThing;
}
@Override
public boolean supportsThingType(ThingTypeUID thingTypeUID) {
return NTP_THING_TYPE_UID.equals(thingTypeUID);
@@ -545,6 +741,9 @@ public class YamlThingProviderTest {
@Override
public @Nullable Thing createThing(ThingTypeUID thingTypeUID, Configuration configuration,
@Nullable ThingUID thingUID, @Nullable ThingUID bridgeUID) {
if (++nbCallsToCreateThing <= nbCallsBeforeSucceededCreatingThing) {
return null;
}
ThingUID effectiveUID = thingUID != null ? thingUID : ThingFactory.generateRandomThingUID(thingTypeUID);
ThingType thingType = thingTypeRegistry.getThingType(thingTypeUID);
if (thingType != null) {