feat: add ConcurrentDownloadTimeoutEvent

Author: javier-godoy

pom.xml

@@ -301,6 +301,9 @@
 					<supportedPackagings>
 						<supportedPackaging>jar</supportedPackaging>
 					</supportedPackagings>
+					<systemProperties>
+						<com.vaadin.flow.server.pushMode>AUTOMATIC</com.vaadin.flow.server.pushMode>
+					</systemProperties>
 				</configuration>
 			</plugin>
 		</plugins>

src/main/java/com/flowingcode/vaadin/addons/gridexporter/ConcurrentDownloadTimeoutEvent.java

@@ -0,0 +1,58 @@
+package com.flowingcode.vaadin.addons.gridexporter;
+
+import java.util.EventObject;
+import java.util.Objects;
+
+/**
+ * An event that is fired when a concurrent download timeout occurs in the {@link GridExporter}.
+ * <p>
+ * This event allows the handler to determine whether the event propagation should be stopped,
+ * preventing other listeners from processing the event.
+ * </p>
+ *
+ * @param <T> the type of the GridExporter source
+ * @see GridExporter#setConcurrentDownloadTimeout(long, java.util.concurrent.TimeUnit)
+ */
+@SuppressWarnings("serial")
+public class ConcurrentDownloadTimeoutEvent extends EventObject {
+
+  private boolean propagationStopped;
+
+  /**
+   * Constructs a new ConcurrentDownloadTimeoutEvent.
+   *
+   * @param source the {@link GridExporter} that is the source of this event
+   * @throws IllegalArgumentException if source is null
+   */
+  public ConcurrentDownloadTimeoutEvent(GridExporter<?> source) {
+    super(Objects.requireNonNull(source));
+  }
+
+  /**
+   * Returns the source of this event.
+   *
+   * @return the {@code GridExporter} that is the source of this event
+   */
+  @Override
+  public GridExporter<?> getSource() {
+    return (GridExporter<?>) super.getSource();
+  }
+
+  /**
+   * Stops the propagation of this event. When propagation is stopped, other listeners will not be
+   * notified of this event.
+   */
+  public void stopPropagation() {
+    propagationStopped = true;
+  }
+
+  /**
+   * Checks if the propagation of this event has been stopped.
+   *
+   * @return {@code true} if the propagation has been stopped, {@code false} otherwise
+   * @see #stopPropagation()
+   */
+  public boolean isPropagationStopped() {
+    return propagationStopped;
+  }
+}

src/main/java/com/flowingcode/vaadin/addons/gridexporter/ConcurrentStreamResourceWriter.java

@@ -147,6 +147,16 @@ public float getCost(VaadinSession session) {
     return DEFAULT_COST;
   }
 
+  /**
+   * Callback method that is invoked when a timeout occurs while trying to acquire a permit for
+   * starting a download.
+   * <p>
+   * Implementations can use this method to perform any necessary actions in response to the
+   * timeout, such as logging a warning or notifying the user.
+   * </p>
+   */
+  protected abstract void onTimeout();
+
   /**
    * Handles {@code stream} (writes data to it) using {@code session} as a context.
    * <p>
@@ -185,6 +195,7 @@ public final void accept(OutputStream stream, VaadinSession session) throws IOEx
             semaphore.release(permits);
           }
         } else {
+          onTimeout();
           throw new InterruptedByTimeoutException();
         }
       } catch (InterruptedException e) {

src/main/java/com/flowingcode/vaadin/addons/gridexporter/GridExporter.java

@@ -32,11 +32,13 @@
 import com.vaadin.flow.data.renderer.BasicRenderer;
 import com.vaadin.flow.data.renderer.LitRenderer;
 import com.vaadin.flow.data.renderer.Renderer;
+import com.vaadin.flow.function.SerializableConsumer;
 import com.vaadin.flow.function.SerializableSupplier;
 import com.vaadin.flow.function.ValueProvider;
 import com.vaadin.flow.server.StreamResource;
 import com.vaadin.flow.server.StreamResourceWriter;
 import com.vaadin.flow.server.VaadinSession;
+import com.vaadin.flow.shared.Registration;
 import java.io.Serializable;
 import java.lang.reflect.Field;
 import java.lang.reflect.InvocationTargetException;
@@ -49,8 +51,10 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -76,6 +80,10 @@ public class GridExporter<T> implements Serializable {
 
   private static long concurrentDownloadTimeoutNanos = 0L;
   private float concurrentDownloadCost = DEFAULT_COST;
+  private final List<SerializableConsumer<ConcurrentDownloadTimeoutEvent>> instanceDownloadTimeoutListeners =
+      new CopyOnWriteArrayList<>();
+  private static final List<SerializableConsumer<ConcurrentDownloadTimeoutEvent>> globalDownloadTimeoutListeners =
+      new CopyOnWriteArrayList<>();
 
   static final String COLUMN_VALUE_PROVIDER_DATA = "column-value-provider-data";
   static final String COLUMN_EXPORTED_PROVIDER_DATA = "column-value-exported-data";
@@ -328,9 +336,69 @@ public long getTimeout() {
         return concurrentDownloadTimeoutNanos;
       }
 
+      @Override
+      protected void onTimeout() {
+        fireConcurrentDownloadTimeout();
+      }
+
+
     };
   }
 
+  /**
+   * Handles the timeout event by notifying all registered listeners.
+   * <p>
+   * This method is called when a timeout occurs during a concurrent download. It creates a
+   * {@link ConcurrentDownloadTimeoutEvent} and notifies all instance and global listeners. If any
+   * listener stops the event propagation, subsequent listeners will not be notified.
+   */
+  private void fireConcurrentDownloadTimeout() {
+    if (!instanceDownloadTimeoutListeners.isEmpty() || !globalDownloadTimeoutListeners.isEmpty()) {
+      grid.getUI().ifPresent(ui -> ui.access(() -> {
+        ConcurrentDownloadTimeoutEvent ev = new ConcurrentDownloadTimeoutEvent(GridExporter.this);
+        Stream.concat(instanceDownloadTimeoutListeners.stream(),
+            globalDownloadTimeoutListeners.stream()).forEach(listener -> {
+              if (!ev.isPropagationStopped()) {
+                listener.accept(ev);
+              }
+            });
+      }));
+    }
+  }
+
+  /**
+   * Adds a listener for concurrent download timeout events specific to this instance.
+   * <p>
+   * The listener will be called whenever a concurrent download timeout event occurs.
+   *
+   * @param listener the listener to be added
+   * @return a {@link Registration} object that can be used to remove the listener
+   */
+  public Registration addConcurrentDownloadTimeoutEvent(
+      SerializableConsumer<ConcurrentDownloadTimeoutEvent> listener) {
+    instanceDownloadTimeoutListeners.add(0, listener);
+    return () -> instanceDownloadTimeoutListeners.remove(listener);
+  }
+
+  /**
+   * Adds a global listener for concurrent download timeout events.
+   * <p>
+   * The listener will be called whenever a concurrent download timeout event occurs.
+   * <p>
+   * Note that instance-specific listeners take precedence over global listeners. If an instance
+   * listener stops the event propagation by calling
+   * {@link ConcurrentDownloadTimeoutEvent#stopPropagation() stopPropagation()}, the global
+   * listeners will not be notified.
+   *
+   * @param listener the listener to be added
+   * @return a {@link Registration} object that can be used to remove the listener
+   */
+  public static Registration addGlobalConcurrentDownloadTimeoutEvent(
+      SerializableConsumer<ConcurrentDownloadTimeoutEvent> listener) {
+    globalDownloadTimeoutListeners.add(0, listener);
+    return () -> globalDownloadTimeoutListeners.remove(listener);
+  }
+
   /**
    * Sets the limit for the {@linkplain #setConcurrentDownloadCost(float) cost of concurrent
    * downloads}. If all the downloads have a cost of {@link #DEFAULT_COST}, the limit represents the

src/test/java/com/flowingcode/vaadin/addons/gridexporter/GridExporterBigDatasetDemo.java

@@ -94,7 +94,8 @@ public GridExporterBigDatasetDemo() throws EncryptedDocumentException, IOExcepti
 
             Additionally, <code>setConcurrentDownloadTimeout</code> enforces a timeout for acquiring the necessary permits
             during a download operation. If the permits are not obtained within the specified timeframe, the download
-            request will be aborted, preventing prolonged waiting periods, especially during peak system loads.
+            request will be aborted and the <code>DownloadTimeoutEvent</code> listener will execute, preventing prolonged
+            waiting periods, especially during peak system loads.
             </div>""");
     add(concurrent);
     // #endif

src/test/java/com/flowingcode/vaadin/addons/gridexporter/VaadinServiceInitListenerImpl.java

@@ -1,5 +1,7 @@
 package com.flowingcode.vaadin.addons.gridexporter;
 
+import com.vaadin.flow.component.notification.Notification;
+import com.vaadin.flow.component.notification.NotificationVariant;
 import com.vaadin.flow.server.ServiceInitEvent;
 import com.vaadin.flow.server.VaadinServiceInitListener;
 import java.util.concurrent.TimeUnit;
@@ -10,6 +12,10 @@ public class VaadinServiceInitListenerImpl implements VaadinServiceInitListener
   public void serviceInit(ServiceInitEvent event) {
     GridExporter.setConcurrentDownloadLimit(10);
     GridExporter.setConcurrentDownloadTimeout(5, TimeUnit.SECONDS);
+    GridExporter.addGlobalConcurrentDownloadTimeoutEvent(ev -> {
+      Notification.show("System is busy. Please try again later.")
+          .addThemeVariants(NotificationVariant.LUMO_ERROR);
+    });
   }
 
 }

src/test/java/com/flowingcode/vaadin/addons/gridexporter/test/ConcurrentExportTests.java

@@ -4,6 +4,7 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertTrue;
 import com.flowingcode.vaadin.addons.gridexporter.ConfigurableConcurrentStreamResourceWriter;
 import com.flowingcode.vaadin.addons.gridexporter.GridExporter;
 import com.vaadin.flow.server.StreamResourceWriter;
@@ -29,17 +30,24 @@ public class ConcurrentExportTests {
 
   private static final int TEST_TIMEOUT = 5000;
 
-  private static Matcher<Throwable> interruptedByTimeout() {
+  private static Matcher<Throwable> throwsInterruptedByTimeout() {
     return Matchers.instanceOf(InterruptedByTimeoutException.class);
   }
 
   private class ConcurrentStreamResourceWriter
       extends ConfigurableConcurrentStreamResourceWriter {
 
+    private boolean interruptedByTimeout;
+
     public ConcurrentStreamResourceWriter(StreamResourceWriter delegate) {
       super(delegate);
     }
 
+    @Override
+    protected void onTimeout() {
+      interruptedByTimeout = true;
+    }
+
   }
 
   private CyclicBarrier barrier;
@@ -68,6 +76,8 @@ private interface MockDownload {
     MockDownload await() throws InterruptedException;
 
     MockDownload start();
+
+    boolean wasInterruptedByTimeout();
   }
 
   private MockDownload newDownload() {
@@ -143,6 +153,11 @@ public MockDownload withCost(float cost) {
         writer.setCost(cost);
         return this;
       }
+
+      @Override
+      public boolean wasInterruptedByTimeout() {
+        return writer.interruptedByTimeout;
+      }
     };
   }
 
@@ -244,7 +259,8 @@ public void testInterruptedByTimeout1()
     await(barrier);
 
     assertThat(q1.get(), nullValue());
-    assertThat(q2.get(), interruptedByTimeout());
+    assertThat(q2.get(), throwsInterruptedByTimeout());
+    assertTrue(q2.wasInterruptedByTimeout());
   }
 
 
@@ -262,7 +278,7 @@ public void testInterruptedByTimeout2()
 
     assertThat(q1.get(), nullValue());
     assertThat(q2.get(), nullValue());
-    assertThat(q3.get(), interruptedByTimeout());
+    assertThat(q3.get(), throwsInterruptedByTimeout());
   }
 
   @Test(timeout = TEST_TIMEOUT)
@@ -278,8 +294,8 @@ public void testInterruptedByTimeout3()
     await(barrier);
 
     assertThat(q1.get(), nullValue());
-    assertThat(q2.get(), interruptedByTimeout());
-    assertThat(q3.get(), interruptedByTimeout());
+    assertThat(q2.get(), throwsInterruptedByTimeout());
+    assertThat(q3.get(), throwsInterruptedByTimeout());
   }
 
 }