Merge branch 'master' into java8-main

Author: rbygrave

metrics-ebean/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <artifactId>avaje-metrics-parent</artifactId>
     <groupId>io.avaje</groupId>
-    <version>9.3-SNAPSHOT</version>
+    <version>9.3-RC1</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

metrics-graphite/pom.xml

@@ -4,7 +4,7 @@
   <parent>
     <artifactId>avaje-metrics-parent</artifactId>
     <groupId>io.avaje</groupId>
-    <version>9.3-SNAPSHOT</version>
+    <version>9.3-RC1</version>
   </parent>
 
   <artifactId>avaje-metrics-graphite</artifactId>

metrics/pom.xml

@@ -3,7 +3,7 @@
   <parent>
     <groupId>io.avaje</groupId>
     <artifactId>avaje-metrics-parent</artifactId>
-    <version>9.3-SNAPSHOT</version>
+    <version>9.3-RC1</version>
   </parent>
 
   <artifactId>avaje-metrics</artifactId>
@@ -20,6 +20,12 @@
       <version>1.7.36</version>
     </dependency>
 
+    <dependency>
+      <groupId>io.avaje</groupId>
+      <artifactId>avaje-applog</artifactId>
+      <version>1.0</version>
+    </dependency>
+
     <dependency>
       <groupId>io.avaje</groupId>
       <artifactId>junit</artifactId>

metrics/src/main/java/io/avaje/metrics/DScheduledTask.java

@@ -0,0 +1,112 @@
+package io.avaje.metrics;
+
+import io.avaje.applog.AppLog;
+
+import java.util.Objects;
+import java.util.concurrent.*;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static java.lang.System.Logger.Level.ERROR;
+
+final class DScheduledTask implements ScheduledTask {
+
+  private static final System.Logger log = AppLog.getLogger(DScheduledTask.class);
+
+  static final class DBuilder implements Builder {
+    private int initial = 60;
+    private int delay = 60;
+    private TimeUnit timeUnit = TimeUnit.SECONDS;
+
+    private Runnable task;
+
+    @Override
+    public DBuilder schedule(int initial, int delay, TimeUnit timeUnit) {
+      this.initial = initial;
+      this.delay = delay;
+      this.timeUnit = timeUnit;
+      return this;
+    }
+
+    @Override
+    public DBuilder task(Runnable task) {
+      this.task = task;
+      return this;
+    }
+
+    @Override
+    public DScheduledTask build() {
+      Objects.requireNonNull(task, "task is required");
+      return new DScheduledTask(task, initial, delay, timeUnit);
+    }
+  }
+
+
+  private final ReentrantLock activeLock = new ReentrantLock();
+  private final Runnable task;
+  private final int initial;
+  private final int delay;
+  private final TimeUnit timeUnit;
+
+  private final ScheduledExecutorService executor;
+  private ScheduledFuture<?> backgroundTask;
+
+  DScheduledTask(Runnable task, int initial, int delay, TimeUnit timeUnit) {
+    this.task = task;
+    this.initial = initial;
+    this.delay = delay;
+    this.timeUnit = timeUnit;
+    this.executor = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory("schTask"));
+  }
+
+  @Override
+  public void start() {
+    this.backgroundTask = executor.scheduleWithFixedDelay(this::runTask, initial, delay, timeUnit);
+  }
+
+  @Override
+  public boolean cancel(boolean mayInterruptIfRunning) {
+    return this.backgroundTask.cancel(mayInterruptIfRunning);
+  }
+
+  /**
+   * Wait for the task to complete if it is actively running .
+   */
+  @Override
+  public void waitIfRunning(long timeout, TimeUnit timeUnit) {
+    try {
+      if (activeLock.tryLock(timeout, timeUnit)) {
+        activeLock.unlock();
+      }
+    } catch (InterruptedException e) {
+      Thread.currentThread().interrupt();
+      log.log(ERROR, "interrupted while waiting for task to complete", e);
+    }
+  }
+
+  private void runTask() {
+    activeLock.lock();
+    try {
+      task.run();
+    } catch (Throwable e) {
+      log.log(ERROR, "Error stopping task", e);
+    } finally {
+      activeLock.unlock();
+    }
+  }
+
+  private static final class DaemonThreadFactory implements ThreadFactory {
+
+    private final String name;
+
+    DaemonThreadFactory(String name) {
+      this.name = name;
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+      Thread t = new Thread(r, name);
+      t.setDaemon(true);
+      return t;
+    }
+  }
+}

metrics/src/main/java/io/avaje/metrics/ScheduledTask.java

@@ -0,0 +1,59 @@
+package io.avaje.metrics;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A ScheduledTask that can run periodically, be cancelled, and
+ * also aware of when its running (for use with Lambda).
+ * <p>
+ * The ScheduledTask will use a Daemon thread and expect to just
+ * stop on JVM shutdown.
+ */
+public interface ScheduledTask {
+
+  /**
+   * Return the Builder for a ScheduledTask.
+   */
+  static Builder builder() {
+    return new DScheduledTask.DBuilder();
+  }
+
+  /**
+   * Start the scheduled task.
+   */
+  void start();
+
+  /**
+   * Cancel the scheduled task.
+   */
+  boolean cancel(boolean mayInterruptIfRunning);
+
+  /**
+   * If the task is actively running wait for the task to complete.
+   */
+  void waitIfRunning(long timeout, TimeUnit timeUnit);
+
+  /**
+   * The builder for a ScheduledTask.
+   */
+  interface Builder {
+
+    /**
+     * Specify the schedule to run the task.
+     * @param initial The initial delay
+     * @param delay The delay between task execution
+     * @param timeUnit The timeunit of the scheduled delay
+     */
+    Builder schedule(int initial, int delay, TimeUnit timeUnit);
+
+    /**
+     * Specify the task to execute periodically according to the schedule.
+     */
+    Builder task(Runnable task);
+
+    /**
+     * Build the scheduled task.
+     */
+    ScheduledTask build();
+  }
+}

metrics/src/test/java/io/avaje/metrics/ScheduledTaskTest.java

@@ -0,0 +1,44 @@
+package io.avaje.metrics;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ScheduledTaskTest {
+
+  @Test
+  void runIt() throws InterruptedException {
+
+    ScheduledTask task = ScheduledTask.builder()
+      .schedule(1, 1, TimeUnit.MILLISECONDS)
+      .task(ScheduledTaskTest::hello)
+      .build();
+
+    assertThat(counter.get()).isEqualTo(0);
+    System.out.println("start...");
+    task.start();
+    Thread.sleep(5);
+    assertThat(counter.get()).isGreaterThan(0);
+
+    task.waitIfRunning(10, TimeUnit.SECONDS);
+    Thread.sleep(30);
+
+    System.out.println("cancel...");
+    task.cancel(false);
+    long after0 = counter.get();
+    assertThat(after0).isGreaterThan(1);
+    Thread.sleep(10);
+    long after1 = counter.get();
+    assertThat(after1).isEqualTo(after0);
+    System.out.println("done");
+  }
+
+  private static final AtomicLong counter = new AtomicLong();
+
+  private static void hello() {
+    System.out.println("hi " + counter.incrementAndGet());
+  }
+}

pom.xml

@@ -9,7 +9,7 @@
 
   <groupId>io.avaje</groupId>
   <artifactId>avaje-metrics-parent</artifactId>
-  <version>9.3-SNAPSHOT</version>
+  <version>9.3-RC1</version>
   <packaging>pom</packaging>
 
   <url>https://avaje-metrics.github.io</url>