WIP hfs watchBody 7

runtime/filesystem-http/src/main/java/io/aklivity/zilla/runtime/filesystem/http/HttpPath.java

@@ -21,7 +21,6 @@
 import static java.net.http.HttpClient.Version.HTTP_2;
 import static java.nio.file.StandardWatchEventKinds.ENTRY_MODIFY;
 import static java.util.Objects.requireNonNull;
-import static org.agrona.LangUtil.rethrowUnchecked;
 
 import java.io.File;
 import java.io.IOException;
@@ -39,6 +38,9 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class HttpPath implements Path
 {
@@ -50,9 +52,10 @@ public class HttpPath implements Path
 
     private final HttpFileSystem fs;
     private final URI location;
+    private final ScheduledExecutorService executor;
+    private int pollSeconds; // TODO: Ati - HttpFileSystemConfiguration -> final
 
     private byte[] body;
-    private boolean longPolling;
     private String etag;
     private HttpWatchService.HttpWatchKey watchKey;
     private CompletableFuture<Void> future;
@@ -65,16 +68,21 @@ public class HttpPath implements Path
         {
             throw new IllegalArgumentException(String.format("invalid protocol: %s", location.getScheme()));
         }
+        System.out.println("HP constructor location " + location); // TODO: Ati
         this.fs = fs;
         this.location = location;
-        this.longPolling = true;
+        this.executor = Executors.newScheduledThreadPool(2);
+        // TODO: Ati - HttpFileSystemConfiguration
+        //this.pollSeconds = 30;
+        this.pollSeconds = 3;
     }
 
     HttpPath()
     {
         this.fs = null;
         this.location = null;
-        this.longPolling = true;
+        this.executor = null;
+        this.pollSeconds = 0;
     }
 
     @Override
@@ -243,6 +251,7 @@ public WatchKey register(
         WatchService watcher,
         WatchEvent.Kind<?>... events) throws IOException
     {
+        System.out.println("HP register"); // TODO: Ati
         return register(watcher, events, new WatchEvent.Modifier[0]);
     }
 
@@ -288,6 +297,13 @@ public int hashCode()
         return Objects.hashCode(location);
     }
 
+    // TODO: Ati - HttpFileSystemConfiguration
+    public void pollSeconds(
+        int pollSeconds)
+    {
+        this.pollSeconds = pollSeconds;
+    }
+
     byte[] readBody()
     {
         System.out.println("HP readBody"); // TODO: Ati
@@ -328,12 +344,30 @@ else if (response.statusCode() == HTTP_NOT_MODIFIED)
         catch (Exception ex)
         {
             System.out.println("HP readBody exception " + ex);  // TODO: Ati
-            rethrowUnchecked(ex);
+            body = new byte[0];
         }
         return body;
     }
 
-    void watchBody()
+    void watch()
+    {
+        scheduleWatchBody(this.pollSeconds);
+    }
+
+    private void scheduleWatchBody(
+        int pollSeconds)
+    {
+        if (pollSeconds == 0)
+        {
+            watchBody();
+        }
+        else
+        {
+            executor.schedule(this::watchBody, pollSeconds, TimeUnit.SECONDS);
+        }
+    }
+
+    private void watchBody()
     {
         HttpRequest.Builder requestBuilder = HttpRequest.newBuilder()
             .GET()
@@ -355,19 +389,17 @@ private void handleResponse(
         System.out.println("HP handleResponse response: " + response); // TODO: Ati
         System.out.println("HP handleResponse response.headers: " + response.headers()); // TODO: Ati
         int statusCode = response.statusCode();
-        //int pollSeconds = 0;
+        int pollSeconds = 1;
         if (statusCode == 404)
         {
             body = EMPTY_BODY;
             watchKey.addEvent(ENTRY_MODIFY, this);
-            longPolling = false;
-            //pollSeconds = this.pollSeconds;
+            pollSeconds = this.pollSeconds;
         }
         else if (statusCode >= 500 && statusCode <= 599)
         {
             body = null;
-            longPolling = false;
-            //pollSeconds = this.pollSeconds;
+            pollSeconds = this.pollSeconds;
         }
         else
         {
@@ -384,9 +416,8 @@ else if (statusCode >= 500 && statusCode <= 599)
                 }
                 else if (response.statusCode() != 304)
                 {
-                    this.body = response.body();
-                    longPolling = false;
-                    //pollSeconds = this.pollSeconds;
+                    body = response.body();
+                    pollSeconds = this.pollSeconds;
                 }
             }
             else
@@ -401,23 +432,27 @@ else if (response.statusCode() != 304)
                     watchKey.addEvent(ENTRY_MODIFY, this);
                     //addEvent(path);
                 }
-                //pollSeconds = this.pollSeconds;
+                pollSeconds = this.pollSeconds;
             }
         }
-        watchKey.watchBody();
-    }
-
-    boolean longPolling()
-    {
-        return longPolling;
+        scheduleWatchBody(pollSeconds);
     }
 
     void cancel()
     {
         System.out.println("HP cancel"); // TODO: Ati
+        body = null;
+        etag = null;
         future.cancel(true);
     }
 
+    // required for testing
+    public void shutdown()
+    {
+        System.out.println("HP shutdown"); // TODO: Ati
+        executor.shutdownNow();
+    }
+
     private Void handleException(
         Throwable throwable)
     {

runtime/filesystem-http/src/main/java/io/aklivity/zilla/runtime/filesystem/http/HttpWatchService.java

@@ -28,9 +28,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Objects;
-import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 public class HttpWatchService implements WatchService
@@ -39,22 +37,18 @@ public class HttpWatchService implements WatchService
 
     private final WatchKey closeKey = new HttpWatchKey(CLOSE_PATH);
 
-    private final ScheduledExecutorService executor;
     private final List<HttpWatchKey> watchKeys;
     private final LinkedBlockingQueue<WatchKey> pendingKeys;
     private final MessageDigest md5;
 
-    private int pollSeconds;
     private volatile boolean closed;
 
     public HttpWatchService()
     {
-        this.executor = Executors.newScheduledThreadPool(2);
+
         this.watchKeys = Collections.synchronizedList(new LinkedList<>());
         this.pendingKeys = new LinkedBlockingQueue<>();
         this.md5 = initMessageDigest("MD5");
-        // TODO: Ati - HttpFileSystemConfiguration
-        this.pollSeconds = 30;
         this.closed = false;
     }
 
@@ -98,13 +92,6 @@ public WatchKey take() throws InterruptedException
         return key;
     }
 
-    // TODO: Ati - HttpFileSystemConfiguration
-    public void pollSeconds(
-        int pollSeconds)
-    {
-        this.pollSeconds = pollSeconds;
-    }
-
     private void checkOpen()
     {
         if (closed)
@@ -147,7 +134,7 @@ HttpWatchKey register(
         }
         System.out.printf("HWS register path: %s\n", path); // TODO: Ati
         HttpWatchKey watchKey = new HttpWatchKey(path);
-        watchKey.watchBody();
+        watchKey.watch();
         watchKeys.add(watchKey);
         return watchKey;
     }
@@ -213,6 +200,7 @@ public boolean reset()
         @Override
         public void cancel()
         {
+            System.out.println("HWK cancel"); // TODO: Ati
             watchKeys.remove(this);
             path.cancel();
             valid = false;
@@ -254,18 +242,11 @@ public int hashCode()
             return Objects.hashCode(path);
         }
 
-        void watchBody()
+        void watch()
         {
             if (valid)
             {
-                if (path.longPolling())
-                {
-                    path.watchBody();
-                }
-                else
-                {
-                    executor.schedule(path::watchBody, pollSeconds, TimeUnit.SECONDS);
-                }
+                path.watch();
             }
         }
 

runtime/filesystem-http/src/test/java/io/aklivity/zilla/runtime/filesystem/http/HttpFileSystemIT.java

@@ -203,8 +203,8 @@ public void shouldWatch() throws Exception
         // GIVEN
         String url = "http://localhost:8080/hello.txt";
         Path path = Path.of(new URI(url));
+        ((HttpPath) path).pollSeconds(1); // TODO: Ati
         HttpWatchService watchService = (HttpWatchService) path.getFileSystem().newWatchService();
-        watchService.pollSeconds(1); // TODO: Ati
 
         // WHEN
         k3po.start();
@@ -216,6 +216,7 @@ public void shouldWatch() throws Exception
         WatchKey key2 = watchService.take();
         List<WatchEvent<?>> events2 = key2.pollEvents();
         watchService.close();
+        ((HttpPath) path).shutdown();
         k3po.finish();
 
         // THEN
@@ -237,7 +238,7 @@ public void shouldWatchRead() throws Exception
         String url = "http://localhost:8080/hello.txt";
         Path path = Path.of(new URI(url));
         HttpWatchService watchService = (HttpWatchService) path.getFileSystem().newWatchService();
-        watchService.pollSeconds(1); // TODO: Ati
+        ((HttpPath) path).pollSeconds(1); // TODO: Ati
 
         // WHEN
         k3po.start();
@@ -252,6 +253,7 @@ public void shouldWatchRead() throws Exception
         k3po.notifyBarrier("SECOND_READ");
         String body2 = Files.readString(path);
         watchService.close();
+        ((HttpPath) path).shutdown();
         k3po.finish();
 
         // THEN