Deliver log in chunks

Each chunk is not taking longer than 30s to deliver (configurable via
OSGi)
This is due to the CDN not allowing streaming servlets (i.e. will always
allow for it to finish before delivering it to the client).

This closes #810

Author: kwin

accesscontroltool-bundle/src/main/java/biz/netcentric/cq/tools/actool/api/AcInstallationService.java

@@ -44,6 +44,15 @@ public interface AcInstallationService {
      */
     public boolean attachLogListener(String jobId, BiConsumer<InstallationLogLevel, String> listener, Consumer<Boolean> finishListener);
 
+    /**
+     * Checks if the asynchronous installation job with the given ID is running.
+     * @param jobId
+     * @return {@code true} if the job with the given ID is running, {@code false} otherwise
+     * @since 3.6.1
+     */
+    public boolean isRunning(String jobId);
+
+    
     /** Applies the full configuration as stored at the path configured at PID biz.netcentric.cq.tools.actool.impl.AcInstallationServiceImpl
      * to the repository.
      * 

accesscontroltool-bundle/src/main/java/biz/netcentric/cq/tools/actool/api/package-info.java

@@ -1,4 +1,4 @@
-@Version("3.1.0")
+@Version("3.2.0")
 package biz.netcentric.cq.tools.actool.api;
 
 /*-

accesscontroltool-bundle/src/main/java/biz/netcentric/cq/tools/actool/impl/AcInstallationServiceImpl.java

@@ -225,7 +225,7 @@ public JobResult process(Job job) {
     @Override
     public boolean attachLogListener(String jobId, BiConsumer<InstallationLogLevel, String> messageListener, Consumer<Boolean> finishListener) {
         Job job = jobManager.getJobById(jobId);
-        if (job == null) {
+        if (!isRunning(jobId)) {
             // finished job or unknown job not to distinguish, as only failed jobs are kept in the history
             LOG.debug("No job found with id {}", jobId);
             return false;
@@ -241,6 +241,13 @@ public boolean attachLogListener(String jobId, BiConsumer<InstallationLogLevel,
         return true;
     }
 
+    
+    @Override
+    public boolean isRunning(String jobId) {
+        Job job = jobManager.getJobById(jobId);
+        return job != null;
+    }
+
     @Override
     public InstallationLog apply() {
         return apply((String)null);

accesscontroltool-bundle/src/main/java/biz/netcentric/cq/tools/actool/ui/AcToolUiService.java

@@ -123,12 +123,20 @@ public class AcToolUiService {
 
         @AttributeDefinition(name="Write access", description="Principal names allowed to modify users/groups and permissions in the system via ACTool configuration files. Only leveraged for Touch UI but not for Web Console Plugin.")
         String[] writeAccessPrincipalNames() default { "administrators", "admin" };
+
+        @AttributeDefinition(name="Log stream request maximum runtime", description="Maximum time in milliseconds until the log stream request should be answered. Must be less than the timeout of the CDN (60 seconds for Fastly used by AEMaaCS).")
+        long maxLogStreamRequestRuntimeInMs() default 30 * 1000; // 30 seconds
     }
 
     private final Map<String, String> countryCodePerName;
 
     private final Configuration config;
 
+    /**
+     * Singleton message queue for the currently running asynchronous installation.
+     */
+    private BlockingQueue<String> messages;
+
     @Activate
     public AcToolUiService(Configuration config) {
         this.config = config;
@@ -289,6 +297,23 @@ protected void doPost(final HttpServletRequest req, final HttpServletResponse re
         }
         try {
             String jobId = acInstallationService.applyAsynchronously(builder.build());
+            messages = new LinkedBlockingQueue<>();
+            acInstallationService.attachLogListener(jobId, (level, message) -> {
+                try {
+                    if (!reqParams.showLogVerbose && level == InstallationLogLevel.TRACE) {
+                        return;
+                    }
+                    messages.put(level + ": " + message);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            }, success -> {
+                try {
+                    messages.put("FINISHED: " + success);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            });
             // return full URL to the server-sent event stream for this installation
             resp.setContentType("text/plain");
             String streamLogUrl = req.getRequestURI() + "/" + SUFFIX_STREAM_LOG + "?jobId=" + URLEncoder.encode(jobId, StandardCharsets.UTF_8.toString()) + "&" + PARAM_SHOW_LOG_VERBOSE + "=" + reqParams.showLogVerbose;
@@ -299,6 +324,7 @@ protected void doPost(final HttpServletRequest req, final HttpServletResponse re
             return;
         }
 
+        
     }
 
     /**
@@ -387,7 +413,9 @@ void downloadLog(HttpServletRequest req, final HttpServletResponse resp) throws
         }
     }
 
-    /** Server-sent events emitted for an asynchronous execution log
+    /** 
+     * Server-sent events emitted for an asynchronous execution log. As the Fastly CDN used with AEM as a Cloud Service does not have <a href="https://docs.fastly.com/en/guides/streaming-miss">streaming</a>
+     * enabled, this is delivering the events in chunks.
      * @throws IOException 
      * 
      * @see <a href="https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events">Server-sent events</a>
@@ -401,35 +429,20 @@ private void streamLog(HttpServletRequest req, HttpServletResponse resp) throws
             resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
             return;
         }
-        final boolean isVerbose = Boolean.parseBoolean(req.getParameter(AcToolUiService.PARAM_SHOW_LOG_VERBOSE));
-        BlockingQueue<String> messages = new LinkedBlockingQueue<>();
-        boolean isActive = acInstallationService.attachLogListener(jobId, (level, message) -> {
-            try {
-                if (!isVerbose && level == InstallationLogLevel.TRACE) {
-                    return;
-                }
-                messages.put(level + ": " + message);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-            }
-        }, success -> {
-            try {
-                messages.put("FINISHED: " + success);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-            }
-        });
-        if (!isActive) {
-            LOG.debug("Haven't found job with id {}, probably already finished processing it", jobId);
+        // is it fully consumed and no longer running?
+        if (!acInstallationService.isRunning(jobId) && (messages == null || messages.isEmpty())) {
+            LOG.debug("Asynchronous installation with id {} is not running anymore and no messages are available", jobId);
             resp.setStatus(HttpServletResponse.SC_NO_CONTENT);
             return;
         }
+        long startTime = System.currentTimeMillis();
         try {
+            boolean isActive = true;
             while (isActive) {
                 String message = messages.poll(30, java.util.concurrent.TimeUnit.SECONDS);
                 if (message == null) {
-                    // just sent keep-alives
-                    message = ".";
+                    message = "...";
+                    isActive = false;
                 }
                 if (message.startsWith("FINISHED: ")) {
                     isActive = false;
@@ -438,6 +451,11 @@ private void streamLog(HttpServletRequest req, HttpServletResponse resp) throws
                     resp.getWriter().println();
                     resp.getWriter().flush();
                 }
+                // overall runtime above threshold
+                if (System.currentTimeMillis() - startTime > config.maxLogStreamRequestRuntimeInMs()) {
+                    LOG.debug("Reached maximum runtime of {} ms for log stream request, finishing response", config.maxLogStreamRequestRuntimeInMs());
+                    isActive = false;
+                }
             }
         } catch(InterruptedException e) {
             Thread.currentThread().interrupt();

pom.xml

@@ -244,6 +244,12 @@
                     <groupId>biz.aQute.bnd</groupId>
                     <artifactId>bnd-baseline-maven-plugin</artifactId>
                     <version>${bnd.version}</version>
+                    <configuration>
+                        <diffignores>
+                            <!-- ignore Bundle-Version header for baseline, i.e. no enforcement of specific bundle versions -->
+                            <diffignore>Bundle-Version</diffignore>
+                        </diffignores>
+                    </configuration>
                 </plugin>
                 <plugin>
                     <groupId>biz.aQute.bnd</groupId>