tests and error handling improvements

Signed-off-by: Kavindu Dodanduwa <[email protected]>

Author: Kavindu-Dodan

Diff for: providers/flagd/src/main/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/file/FileConnector.java

@@ -45,7 +45,7 @@ public void init() throws IOException {
                 // initial read
                 String flagData = new String(Files.readAllBytes(filePath), StandardCharsets.UTF_8);
                 if (!queue.offer(new StreamPayload(StreamPayloadType.DATA, flagData))) {
-                    throw new RuntimeException("Unable to write to queue. Queue is full.");
+                    log.warn("Unable to offer file content to queue: queue is full");
                 }
 
                 long lastTS = Files.getLastModifiedTime(filePath).toMillis();
@@ -58,7 +58,7 @@ public void init() throws IOException {
                         lastTS = currentTS;
                         flagData = new String(Files.readAllBytes(filePath), StandardCharsets.UTF_8);
                         if (!queue.offer(new StreamPayload(StreamPayloadType.DATA, flagData))) {
-                            throw new RuntimeException("Unable to write to queue. Queue is full.");
+                            log.warn("Unable to offer file content to queue: queue is full");
                         }
                     }
 
@@ -67,7 +67,9 @@ public void init() throws IOException {
 
                 log.info("Shutting down file connector.");
             } catch (Throwable t) {
-                log.error("Error from file connector", t);
+                log.error("Error from file connector. File connector will exit", t);
+                // note - queue offer response is intentionally ignored
+                queue.offer(new StreamPayload(StreamPayloadType.ERROR, t.toString()));
             }
         });
 

Diff for: providers/flagd/src/test/java/dev/openfeature/contrib/providers/flagd/resolver/process/storage/connector/file/FileConnectorTest.java

@@ -4,15 +4,18 @@
 import dev.openfeature.contrib.providers.flagd.resolver.process.storage.connector.StreamPayloadType;
 import org.junit.jupiter.api.Test;
 
+import java.io.File;
 import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
 import java.time.Duration;
 import java.util.concurrent.BlockingQueue;
 
 import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.VALID_LONG;
 import static dev.openfeature.contrib.providers.flagd.resolver.process.TestUtils.getResourcePath;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 
 class FileConnectorTest {
@@ -39,12 +42,65 @@ void readAndExposeFeatureFlagsFromSource() throws IOException {
     }
 
     @Test
-    void throwsErrorIfInvalidFile(){
+    void emitErrorStateForInvalidPath() throws IOException {
         // given
         final FileConnector connector = new FileConnector("INVALID_PATH");
 
+        // when
+        connector.init();
+
+        // then
+        final BlockingQueue<StreamPayload> stream = connector.getStream();
+
+        // Must emit an error within considerable time
+        final StreamPayload[] payload = new StreamPayload[1];
+        assertTimeoutPreemptively(Duration.ofMillis(200), () -> {
+            payload[0] = stream.take();
+        });
+
+        assertNotNull(payload[0].getData());
+        assertEquals(StreamPayloadType.ERROR, payload[0].getType());
+    }
+
+    @Test
+    void watchForFileUpdatesAndEmitThem() throws IOException {
+        final String initial = "{\"flags\":{\"myBoolFlag\":{\"state\":\"ENABLED\",\"variants\":{\"on\":true," +
+                "\"off\":false},\"defaultVariant\":\"on\"}}}";
+
+        final String updatedFlags = "{\"flags\":{\"myBoolFlag\":{\"state\":\"ENABLED\",\"variants\":{\"on\":true," +
+                "\"off\":false},\"defaultVariant\":\"off\"}}}";
+
+        // given
+        final File tmpFlagFile = File.createTempFile("flagd", "flags");
+        final Path flagSourcePath = tmpFlagFile.toPath();
+
+        Files.write(flagSourcePath, initial.getBytes(), StandardOpenOption.WRITE);
+        final FileConnector connector = new FileConnector(flagSourcePath.toString());
+
+        // when
+        connector.init();
+
         // then
-        assertThrows(IOException.class, connector::init);
+        final BlockingQueue<StreamPayload> stream = connector.getStream();
+
+        final StreamPayload[] payload = new StreamPayload[1];
+
+        // first validate the initial payload
+        assertTimeoutPreemptively(Duration.ofMillis(200), () -> {
+            payload[0] = stream.take();
+        });
+
+        assertEquals(initial, payload[0].getData());
+
+        // then update the flags
+        Files.write(flagSourcePath, updatedFlags.getBytes(), StandardOpenOption.WRITE);
+
+        // finally wait for updated payload
+        assertTimeoutPreemptively(Duration.ofSeconds(10), () -> {
+            payload[0] = stream.take();
+        });
+
+        assertEquals(updatedFlags, payload[0].getData());
     }
 
 }