You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tomee.apache.org by db...@apache.org on 2022/09/23 01:53:10 UTC
[tomee] branch TOMEE-4050 created (now 876bf3b9e1)
This is an automated email from the ASF dual-hosted git repository.
dblevins pushed a change to branch TOMEE-4050
in repository https://gitbox.apache.org/repos/asf/tomee.git
at 876bf3b9e1 Initialization and retry tests for CachingSupplier Required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
This branch includes the following new commits:
new 5e943d37fb Add greater/less operations on Duration
new 6e201ba508 Prototype CachingSupplier required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
new 876bf3b9e1 Initialization and retry tests for CachingSupplier Required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[tomee] 01/03: Add greater/less operations on Duration
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dblevins pushed a commit to branch TOMEE-4050
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 5e943d37fb26f971391c7fe9afe2d392f5b30cae
Author: David Blevins <db...@tomitribe.com>
AuthorDate: Wed Sep 21 13:48:03 2022 -0700
Add greater/less operations on Duration
---
.../java/org/apache/openejb/util/Duration.java | 21 +++++++
.../java/org/apache/openejb/util/DurationTest.java | 65 ++++++++++++++++++++++
2 files changed, 86 insertions(+)
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java b/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
index f028a16327..dd7b1c6af4 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
@@ -115,6 +115,7 @@ public class Duration {
this.unit = unit;
}
+
private static final class Normalize {
private final long a;
private final long b;
@@ -168,6 +169,26 @@ public class Duration {
return new Duration(n.a - n.b, n.base);
}
+ public boolean greaterThan(final Duration that) {
+ final Normalize n = new Normalize(this, that);
+ return n.a > n.b;
+ }
+
+ public boolean greaterOrEqualTo(final Duration that) {
+ final Normalize n = new Normalize(this, that);
+ return n.a >= n.b;
+ }
+
+ public boolean lessThan(final Duration that) {
+ final Normalize n = new Normalize(this, that);
+ return n.a < n.b;
+ }
+
+ public boolean lessOrEqualTo(final Duration that) {
+ final Normalize n = new Normalize(this, that);
+ return n.a <= n.b;
+ }
+
public static Duration parse(final String text) {
return new Duration(text);
}
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java b/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
index 2713c0d270..4f370dbac0 100644
--- a/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
+++ b/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
@@ -20,6 +20,7 @@ import junit.framework.TestCase;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
@@ -75,4 +76,68 @@ public class DurationTest extends TestCase {
assertEquals(new Duration(125, SECONDS), Duration.parse("2 minutes and 5 seconds"));
}
+
+ public void testAdd() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+
+ final Duration c = a.add(b);
+ assertEquals(c.getUnit(), SECONDS);
+ assertEquals(c.getTime(), 61);
+
+ final Duration d = b.add(a);
+ assertEquals(d.getUnit(), SECONDS);
+ assertEquals(d.getTime(), 61);
+ }
+
+ public void testSubtract() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+
+ final Duration c = a.subtract(b);
+ assertEquals(c.getUnit(), SECONDS);
+ assertEquals(c.getTime(), -59);
+
+ final Duration d = b.subtract(a);
+ assertEquals(d.getUnit(), SECONDS);
+ assertEquals(d.getTime(), 59);
+ }
+
+ public void testGreaterThan() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+ assertFalse(a.greaterThan(b));
+ assertFalse(a.greaterThan(a));
+ assertTrue(b.greaterThan(a));
+ }
+
+ public void testLessThan() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+ assertTrue(a.lessThan(b));
+ assertFalse(a.lessThan(a));
+ assertFalse(b.lessThan(a));
+ }
+
+ public void testGreaterOrEqual() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+ assertFalse(a.greaterOrEqualTo(b));
+ assertTrue(a.greaterOrEqualTo(a));
+ assertTrue(b.greaterOrEqualTo(a));
+ }
+
+ public void testLessOrEqual() {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = new Duration(1, MINUTES);
+
+ assertTrue(a.lessOrEqualTo(b));
+ assertTrue(a.lessOrEqualTo(a));
+ assertFalse(b.lessOrEqualTo(a));
+ }
}
[tomee] 02/03: Prototype CachingSupplier required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dblevins pushed a commit to branch TOMEE-4050
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 6e201ba5085a696aa3c0b0d8cd7af1790de73caa
Author: David Blevins <db...@tomitribe.com>
AuthorDate: Wed Sep 21 18:51:16 2022 -0700
Prototype CachingSupplier required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
---
.../org/apache/openejb/util/CachedSupplier.java | 237 +++++++++++++++++++++
.../java/org/apache/openejb/util/Duration.java | 14 ++
.../apache/openejb/util/CachedSupplierTest.java | 112 ++++++++++
.../java/org/apache/openejb/util/DurationTest.java | 17 ++
4 files changed, 380 insertions(+)
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java b/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java
new file mode 100644
index 0000000000..8eec3d6444
--- /dev/null
+++ b/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.util;
+
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+public class CachedSupplier<T> implements Supplier<T> {
+
+ private final Duration initialRetryDelay;
+ private final Duration maxRetryDelay;
+ private final Duration accessTimeout;
+ private final Duration refreshInterval;
+ private final Supplier<T> supplier;
+
+ private final AtomicReference<T> value = new AtomicReference<>();
+ private final AtomicReference<Accessor<T>> accessor = new AtomicReference<>();
+ private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory());
+
+ private CachedSupplier(final Supplier<T> supplier, final Duration initialRetryDelay,
+ final Duration maxRetryDelay, final Duration accessTimeout,
+ final Duration refreshInterval) {
+
+ Objects.requireNonNull(supplier, "supplier");
+ Objects.requireNonNull(initialRetryDelay, "initialRetryDelay");
+ Objects.requireNonNull(maxRetryDelay, "maxRetryDelay");
+ Objects.requireNonNull(accessTimeout, "accessTimeout");
+ Objects.requireNonNull(refreshInterval, "refreshInterval");
+
+ this.supplier = supplier;
+ this.initialRetryDelay = initialRetryDelay;
+ this.maxRetryDelay = maxRetryDelay;
+ this.accessTimeout = accessTimeout;
+ this.refreshInterval = refreshInterval;
+
+ /*
+ * This must be last as it starts running code
+ * that uses the above settings
+ */
+ this.accessor.set(new BlockTillInitialized());
+ }
+
+ @Override
+ public T get() {
+ final Accessor<T> accessor = this.accessor.get();
+ return accessor.get();
+ }
+
+ public interface Accessor<T> {
+ T get();
+ }
+
+ class BlockTillInitialized implements Accessor<T> {
+ final CountDownLatch initialized = new CountDownLatch(1);
+
+ public BlockTillInitialized() {
+ executor.execute(new Initialize(1, initialRetryDelay));
+ }
+
+ @Override
+ public T get() {
+ try {
+ initialized.await(accessTimeout.getTime(), accessTimeout.getUnit());
+ return value.get();
+ } catch (InterruptedException e) {
+ throw new TimeoutException();
+ }
+ }
+
+ class Initialize implements Runnable {
+ final int attempts;
+ final Duration delay;
+
+ public Initialize(final int attempts, final Duration delay) {
+ this.attempts = attempts;
+ this.delay = delay;
+ }
+
+ public Initialize retry() {
+ if (delay.greaterOrEqualTo(maxRetryDelay)) {
+ return new Initialize(attempts + 1, maxRetryDelay);
+ } else {
+ return new Initialize(attempts + 1, Duration.min(maxRetryDelay, delay.multiply(2)));
+ }
+ }
+
+ @Override
+ public void run() {
+ try {
+ final T t = supplier.get();
+ if (t != null) {
+ value.set(t);
+ accessor.set(new Initialized());
+ initialized.countDown();
+ return;
+ }
+ } catch (Throwable e) {
+ // TODO
+ e.printStackTrace();
+ }
+
+ /*
+ * Initialization didn't work. Let's try again
+ */
+ final Initialize retry = retry();
+ executor.schedule(retry, retry.delay.getTime(), retry.delay.getUnit());
+ }
+ }
+ }
+
+ class Initialized implements Accessor<T> {
+ public Initialized() {
+ executor.scheduleAtFixedRate(new Refresh(), refreshInterval.getTime(), refreshInterval.getTime(), refreshInterval.getUnit());
+ }
+
+ @Override
+ public T get() {
+ return value.get();
+ }
+
+ class Refresh implements Runnable {
+ @Override
+ public void run() {
+ try {
+ final T t = supplier.get();
+ if (t != null) {
+ value.set(t);
+ }
+ } catch (Throwable e) {
+ // TODO
+ e.printStackTrace();
+ }
+ }
+ }
+ }
+
+ public static <T> CachedSupplier<T> of(final Supplier<T> supplier) {
+ return new Builder<T>().supplier(supplier).build();
+ }
+
+ public static <T> Builder<T> builder(final Supplier<T> supplier) {
+ return new Builder<T>().supplier(supplier);
+ }
+
+ public static class TimeoutException extends RuntimeException {
+ // TODO
+ }
+
+ private static class DaemonThreadFactory implements ThreadFactory {
+ @Override
+ public Thread newThread(final Runnable r) {
+ final Thread thread = new Thread(r);
+ thread.setName(CachedSupplier.class.getSimpleName() + " Supplier");
+ thread.setDaemon(true);
+ return thread;
+ }
+ }
+
+ public static class Builder<T> {
+ private Duration initialRetryDelay = new Duration(2, TimeUnit.SECONDS);
+ private Duration maxRetryDelay = new Duration(1, TimeUnit.HOURS);
+ private Duration accessTimeout = new Duration(30, TimeUnit.SECONDS);
+ private Duration refreshInterval = new Duration(1, TimeUnit.DAYS);
+ private Supplier<T> supplier;
+
+
+ public Builder<T> initialRetryDelay(final Duration initialRetryDelay) {
+ this.initialRetryDelay = initialRetryDelay;
+ return this;
+ }
+
+ public Builder<T> initialRetryDelay(final int time, final TimeUnit unit) {
+ return initialRetryDelay(new Duration(time, unit));
+ }
+
+ public Builder<T> maxRetryDelay(final Duration maxRetryDelay) {
+ this.maxRetryDelay = maxRetryDelay;
+ return this;
+ }
+
+ public Builder<T> maxRetryDelay(final int time, final TimeUnit unit) {
+ return maxRetryDelay(new Duration(time, unit));
+ }
+
+ public Builder<T> accessTimeout(final Duration accessTimeout) {
+ this.accessTimeout = accessTimeout;
+ return this;
+ }
+
+ public Builder<T> accessTimeout(final int time, final TimeUnit unit) {
+ return accessTimeout(new Duration(time, unit));
+ }
+
+
+ public Builder<T> refreshInterval(final Duration refreshInterval) {
+ this.refreshInterval = refreshInterval;
+ return this;
+ }
+
+ public Builder<T> refreshInterval(final int time, final TimeUnit unit) {
+ return refreshInterval(new Duration(time, unit));
+ }
+
+ public Builder<T> supplier(final Supplier<T> supplier) {
+ this.supplier = supplier;
+ return this;
+ }
+
+ public CachedSupplier<T> build() {
+ return new CachedSupplier<>(supplier,
+ initialRetryDelay,
+ maxRetryDelay,
+ accessTimeout,
+ refreshInterval);
+ }
+ }
+}
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java b/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
index dd7b1c6af4..9de1d84094 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/util/Duration.java
@@ -169,6 +169,10 @@ public class Duration {
return new Duration(n.a - n.b, n.base);
}
+ public Duration multiply(final long n) {
+ return new Duration(getTime() * n, getUnit());
+ }
+
public boolean greaterThan(final Duration that) {
final Normalize n = new Normalize(this, that);
return n.a > n.b;
@@ -189,6 +193,16 @@ public class Duration {
return n.a <= n.b;
}
+ public static Duration max(final Duration a, final Duration b) {
+ final Normalize n = new Normalize(a, b);
+ return (n.a >= n.b) ? a : b;
+ }
+
+ public static Duration min(final Duration a, final Duration b) {
+ final Normalize n = new Normalize(a, b);
+ return (n.a <= n.b) ? a : b;
+ }
+
public static Duration parse(final String text) {
return new Duration(text);
}
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java b/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java
new file mode 100644
index 0000000000..e179bc3acc
--- /dev/null
+++ b/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.openejb.util;
+
+import org.junit.Test;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.*;
+
+public class CachedSupplierTest {
+
+ /**
+ * Supplier returns a valid result immediately and there
+ * are no delays on the first get.
+ *
+ * We also assert that calling get multiple times on the
+ * CachedSupplier return the cached result and do not get
+ * updated results before the refresh occurs.
+ */
+ @Test
+ public void happyPath() {
+ }
+
+ /**
+ * Supplier does not immediately return an initial instance, so we
+ * block till one is available. We assert that we blocked until get
+ * a valid result and no timeout or null is returned.
+ */
+ @Test
+ public void delayedInitialization() {
+ }
+
+ /**
+ * Supplier does not immediately return an initial instance
+ * and the timeout is reached. We assert a TimeoutException
+ * is thrown. We also assert that when the Supplier eventually
+ * does return a valid result we no longer get a TimeoutException
+ * or any blocking.
+ */
+ @Test
+ public void delayedInitializationTimeout() {
+ }
+
+ /**
+ * Supplier returns null on the first three calls to get. On the
+ * fourth retry a valid result is returned from get. We assert
+ * the number of times the supplier get is called as well as the
+ * time between each call to ensure exponential backoff is working
+ */
+ @Test
+ public void initializationRetry() {
+ }
+
+ /**
+ * Supplier returns null repeatedly on all initialization attempts.
+ * We assert that when the max retry time is reached all subsequent
+ * retries are at that same time interval and do not continue increasing
+ * exponentially.
+ */
+ @Test
+ public void initializationRetryTillMax() {
+ }
+
+ /**
+ * Suppler returns a valid result on initialization. We expect that even
+ * when we are not actively calling get() the value will be refreshed
+ * according to the refreshInterval. We wait for at least 3 refreshes
+ * to occur and assert the value we get is the most recent value returned
+ * from the supplier. We intentionally check for this expected value
+ * while the refresh is currently executing for the fourth time. We do
+ * that to ensure that there is no time values are null, even when we're
+ * concurrently trying to refresh it in the background.
+ */
+ @Test
+ public void refreshReliablyCalled() {
+ }
+
+
+ /**
+ * On the first refresh the Supplier returns null indicating there is
+ * no valid replacement. We assert that the previous valid value is
+ * still in use.
+ */
+ @Test
+ public void refreshFailedWithNull() {
+ }
+
+ /**
+ * On the first refresh the Supplier throws an exception, therefore there is
+ * no valid replacement. We assert that the previous valid value is
+ * still in use.
+ */
+ @Test
+ public void refreshFailedWithException() {
+ }
+
+}
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java b/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
index 4f370dbac0..f0258da294 100644
--- a/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
+++ b/container/openejb-core/src/test/java/org/apache/openejb/util/DurationTest.java
@@ -140,4 +140,21 @@ public class DurationTest extends TestCase {
assertTrue(a.lessOrEqualTo(a));
assertFalse(b.lessOrEqualTo(a));
}
+
+ public void testMultiply() {
+ {
+ final Duration a = new Duration(1, SECONDS);
+ final Duration b = a.multiply(3);
+
+ assertEquals(b.getUnit(), SECONDS);
+ assertEquals(b.getTime(), 3);
+ }
+ {
+ final Duration a = new Duration(3, MINUTES);
+ final Duration b = a.multiply(3);
+
+ assertEquals(b.getUnit(), MINUTES);
+ assertEquals(b.getTime(), 9);
+ }
+ }
}
[tomee] 03/03: Initialization and retry tests for CachingSupplier Required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
Posted by db...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dblevins pushed a commit to branch TOMEE-4050
in repository https://gitbox.apache.org/repos/asf/tomee.git
commit 876bf3b9e123e485a1a3488e8de5e1bed0e000dc
Author: David Blevins <db...@tomitribe.com>
AuthorDate: Thu Sep 22 18:50:16 2022 -0700
Initialization and retry tests for CachingSupplier
Required for TOMEE-4050: Retry and Refresh for MP JWT keys supplied via HTTP
---
.../org/apache/openejb/util/CachedSupplier.java | 6 +-
.../apache/openejb/util/CachedSupplierTest.java | 410 ++++++++++++++++++++-
2 files changed, 411 insertions(+), 5 deletions(-)
diff --git a/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java b/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java
index 8eec3d6444..45a47e8a35 100644
--- a/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java
+++ b/container/openejb-core/src/main/java/org/apache/openejb/util/CachedSupplier.java
@@ -80,8 +80,10 @@ public class CachedSupplier<T> implements Supplier<T> {
@Override
public T get() {
try {
- initialized.await(accessTimeout.getTime(), accessTimeout.getUnit());
- return value.get();
+ if (initialized.await(accessTimeout.getTime(), accessTimeout.getUnit())){
+ return value.get();
+ }
+ throw new TimeoutException();
} catch (InterruptedException e) {
throw new TimeoutException();
}
diff --git a/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java b/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java
index e179bc3acc..ecac615047 100644
--- a/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java
+++ b/container/openejb-core/src/test/java/org/apache/openejb/util/CachedSupplierTest.java
@@ -18,9 +18,23 @@ package org.apache.openejb.util;
import org.junit.Test;
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
-import static org.junit.Assert.*;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class CachedSupplierTest {
@@ -34,6 +48,17 @@ public class CachedSupplierTest {
*/
@Test
public void happyPath() {
+ final AtomicInteger count = new AtomicInteger();
+ final Supplier<Integer> supplier = count::incrementAndGet;
+ final CachedSupplier<Integer> cached = CachedSupplier.of(supplier);
+
+ Runner.threads(100)
+ .run(() -> assertEquals(1, (int) cached.get()))
+ .assertNoExceptions()
+ .assertTimesLessThan(5, MILLISECONDS);
+
+ // Assert the supplier was not called more than once
+ assertEquals(1, count.get());
}
/**
@@ -43,6 +68,32 @@ public class CachedSupplierTest {
*/
@Test
public void delayedInitialization() {
+ final CountDownLatch causeSomeDelays = new CountDownLatch(1);
+ final AtomicInteger count = new AtomicInteger();
+ final Supplier<Integer> supplier = () -> {
+ await(causeSomeDelays);
+ sleep(111);
+ return count.incrementAndGet();
+ };
+ final CachedSupplier<Integer> cached = CachedSupplier.of(supplier);
+
+ final Runner runner = Runner.threads(100);
+
+ // Run and expect at least 100 ms of delays
+ runner.pre(causeSomeDelays::countDown)
+ .run(() -> assertEquals(1, (int) cached.get()))
+ .assertNoExceptions()
+ .assertTimesGreaterThan(100, MILLISECONDS)
+ .assertTimesLessThan(200, MILLISECONDS);
+
+ // Everything is cached now, so runs should be quick
+ runner.pre(null)
+ .run(() -> assertEquals(1, (int) cached.get()))
+ .assertNoExceptions()
+ .assertTimesLessThan(5, MILLISECONDS);
+
+ // Assert the supplier was not called more than once
+ assertEquals(1, count.get());
}
/**
@@ -53,7 +104,51 @@ public class CachedSupplierTest {
* or any blocking.
*/
@Test
- public void delayedInitializationTimeout() {
+ public void delayedInitializationTimeout() throws InterruptedException {
+ final CountDownLatch causeSomeDelays = new CountDownLatch(1);
+ final CountDownLatch nearlyThere = new CountDownLatch(1);
+ final AtomicInteger count = new AtomicInteger();
+ final Supplier<Integer> supplier = () -> {
+ await(causeSomeDelays);
+ sleep(150);
+ nearlyThere.countDown();
+ sleep(50);
+ try {
+ return count.incrementAndGet();
+ } finally {
+ nearlyThere.countDown();
+ }
+ };
+
+ final CachedSupplier<Integer> cached = CachedSupplier.builder(supplier)
+ .accessTimeout(100, MILLISECONDS)
+ .build();
+
+ final Runner runner = Runner.threads(100);
+
+ runner.pre(causeSomeDelays::countDown)
+ .run(() -> assertEquals(1, (int) cached.get()))
+ .assertExceptions(CachedSupplier.TimeoutException.class)
+ .assertTimesGreaterThan(99, MILLISECONDS)
+ .assertTimesLessThan(150, MILLISECONDS);
+
+ // Wait for the supplier to get near completion
+ assertTrue(nearlyThere.await(1, MINUTES));
+
+ runner.pre(null);
+
+ // Calls should now block a bit, but ultimately succeed with no issues
+ runner.run(() -> assertEquals(1, (int) cached.get()))
+ .assertNoExceptions()
+ .assertTimesGreaterThan(30, MILLISECONDS);
+
+ // Calls should now succeed with no delay
+ runner.run(() -> assertEquals(1, (int) cached.get()))
+ .assertNoExceptions()
+ .assertTimesLessThan(5, MILLISECONDS);
+
+ // Assert the supplier was not called more than once
+ assertEquals(1, count.get());
}
/**
@@ -64,6 +159,43 @@ public class CachedSupplierTest {
*/
@Test
public void initializationRetry() {
+ final Long[] calls = new Long[10];
+ final AtomicInteger count = new AtomicInteger();
+ final Supplier<Integer> supplier = () -> {
+ final int i = count.incrementAndGet();
+ if (i < calls.length) {
+ calls[i] = System.nanoTime();
+ }
+
+ // Return null for the first three calls
+ // Then return the actual value
+ return i < 4 ? null : i;
+ };
+
+ final CachedSupplier<Integer> cached = CachedSupplier.builder(supplier)
+ .initialRetryDelay(500, MILLISECONDS)
+ .accessTimeout(1, MINUTES)
+ .build();
+
+ final Runner runner = Runner.threads(100);
+
+ runner.run(() -> assertEquals(4, (int) cached.get()))
+ .assertNoExceptions();
+
+ final Long[] tries = Stream.of(calls)
+ .filter(Objects::nonNull)
+ .toArray(Long[]::new);
+
+ assertEquals(4, tries.length);
+ assertEquals(4, count.get());
+
+ long first = NANOSECONDS.toSeconds(tries[1] - tries[0]);
+ long second = NANOSECONDS.toSeconds(tries[2] - tries[1]);
+ long third = NANOSECONDS.toSeconds(tries[3] - tries[2]);
+
+ assertEquals(1, first);
+ assertEquals(2, second);
+ assertEquals(4, third);
}
/**
@@ -74,6 +206,50 @@ public class CachedSupplierTest {
*/
@Test
public void initializationRetryTillMax() {
+ final Long[] calls = new Long[10];
+ final AtomicInteger count = new AtomicInteger();
+ final Supplier<Integer> supplier = () -> {
+ final int i = count.incrementAndGet();
+ if (i < calls.length) {
+ calls[i] = System.nanoTime();
+ }
+
+ // Return null for the first three calls
+ // Then return the actual value
+ return i < 7 ? null : i;
+ };
+
+ final CachedSupplier<Integer> cached = CachedSupplier.builder(supplier)
+ .initialRetryDelay(500, MILLISECONDS)
+ .maxRetryDelay(10, SECONDS)
+ .accessTimeout(1, MINUTES)
+ .build();
+
+ final Runner runner = Runner.threads(100);
+
+ runner.run(() -> assertEquals(7, (int) cached.get()))
+ .assertNoExceptions();
+
+ final Long[] tries = Stream.of(calls)
+ .filter(Objects::nonNull)
+ .toArray(Long[]::new);
+
+ assertEquals(7, tries.length);
+ assertEquals(7, count.get());
+
+ long first = NANOSECONDS.toSeconds(tries[1] - tries[0]);
+ long second = NANOSECONDS.toSeconds(tries[2] - tries[1]);
+ long third = NANOSECONDS.toSeconds(tries[3] - tries[2]);
+ long fourth = NANOSECONDS.toSeconds(tries[4] - tries[3]);
+ long fifth = NANOSECONDS.toSeconds(tries[5] - tries[4]);
+ long sixth = NANOSECONDS.toSeconds(tries[6] - tries[5]);
+
+ assertEquals(1, first);
+ assertEquals(2, second);
+ assertEquals(4, third);
+ assertEquals(8, fourth);
+ assertEquals(10, fifth);
+ assertEquals(10, sixth);
}
/**
@@ -90,7 +266,6 @@ public class CachedSupplierTest {
public void refreshReliablyCalled() {
}
-
/**
* On the first refresh the Supplier returns null indicating there is
* no valid replacement. We assert that the previous valid value is
@@ -109,4 +284,233 @@ public class CachedSupplierTest {
public void refreshFailedWithException() {
}
+ private void sleep(final int millis) {
+ try {
+ Thread.sleep(millis);
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private void await(final CountDownLatch latch) {
+ try {
+ latch.await();
+ } catch (InterruptedException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ static class Timer {
+ private final long start = System.nanoTime();
+
+ public static Timer start() {
+ return new Timer();
+ }
+
+ public Time time() {
+ return new Time(System.nanoTime() - start);
+ }
+
+ public static class Time {
+ private final long time;
+ private final String description;
+
+ public Time(final long timeInNanoseconds) {
+ this.time = timeInNanoseconds;
+ final long seconds = NANOSECONDS.toSeconds(this.time);
+ final long milliseconds = NANOSECONDS.toMillis(this.time) - SECONDS.toMillis(seconds);
+ final long nanoseconds = this.time - SECONDS.toNanos(seconds) - MILLISECONDS.toNanos(milliseconds);
+ this.description = String.format("%ss, %sms and %sns", seconds, milliseconds, nanoseconds);
+ }
+
+ public long getTime() {
+ return time;
+ }
+
+ public Time assertLessThan(final long time, final TimeUnit unit) {
+ final long expected = unit.toNanos(time);
+ final long actual = this.time;
+ assertTrue("Actual time: " + description, actual < expected);
+ return this;
+ }
+
+ public Time assertGreaterThan(final long time, final TimeUnit unit) {
+ final long expected = unit.toNanos(time);
+ final long actual = this.time;
+ assertTrue("Actual time: " + description, actual > expected);
+ return this;
+ }
+ }
+ }
+
+ public static class Runner {
+ private final int threads;
+ private final Executor executor;
+ private final Duration timeout = new Duration(1, MINUTES);
+ private Runnable before = null;
+
+ public Runner(final int threads) {
+ this.threads = threads;
+ this.executor = Executors.newFixedThreadPool(threads, new DaemonThreadFactory(Runner.class));
+ }
+
+ public static Runner threads(final int threads) {
+ return new Runner(threads);
+ }
+
+ public Runner pre(final Runnable runnable) {
+ this.before = runnable;
+ return this;
+ }
+
+ public Run run(final Runnable runnable) {
+ final Throwable[] failures = new Throwable[threads];
+ final Timer.Time[] times = new Timer.Time[threads];
+
+ /*
+ * You won't immediately understand these CountDownLatches (look down).
+ *
+ * Here's the deal: when you launch 100+ threads in a loop as we're
+ * about to do it can take 25+ milliseconds. By the time you get to
+ * your 99th thread, the previous 50 are all probably gone. The
+ * thread-creation overhead messes with all your timings and threads
+ * are executing somewhat serially with very little parallelism.
+ *
+ * The latches fix this by forcing all the threads to truly run
+ * at the same time.
+ *
+ * Imagine each thread is a runner in a race. What we want is
+ * each runner to get on the racetrack, into the starting
+ * position (ready.countDown) and wait diligently for the sound
+ * of the starting pistol (start.await) before they start running.
+ *
+ * When all runners are in position (ready.await), we fire the starting
+ * pistol (start.countDown). Awesome, they're all truly running at
+ * once and competing.
+ *
+ * As each runner finishes the race we have them call completed.countDown
+ * When all runners are finished the completed.await call will unblock
+ * and we exit this method with all results in hand.
+ *
+ * Seems like overkill, but after you've been burned by poor testing
+ * covering up thread safety issues you learn to do it right.
+ */
+ final CountDownLatch ready = new CountDownLatch(threads);
+ final CountDownLatch start = new CountDownLatch(1);
+ final CountDownLatch completed = new CountDownLatch(threads);
+
+ for (int submitted = 0; submitted < threads; submitted++) {
+ final int id = submitted;
+ executor.execute(new Runnable() {
+ @Override
+ public void run() {
+ ready.countDown();
+ try {
+ start.await();
+ } catch (InterruptedException e) {
+ return;
+ }
+
+ /*
+ * If there's anything we'd like to execute
+ * that shouldn't be included in the timings,
+ * do it now.
+ */
+ if (before != null) before.run();
+
+ /*
+ * Run, Forrest! Run!!
+ */
+ final Timer timer = Timer.start();
+ try {
+ runnable.run();
+ } catch (Throwable t) {
+ failures[id] = t;
+ } finally {
+ times[id] = timer.time();
+ completed.countDown();
+ }
+ }
+ });
+ }
+
+ // wait for the above threads to be ready
+ await(ready, "ready");
+
+ // fire the starting pistol
+ start.countDown();
+
+ // wait for them to finish the race
+ await(completed, "completed");
+
+ return new Run(threads, failures, times);
+ }
+
+ private void await(final CountDownLatch latch, final String state) {
+ try {
+ if (!latch.await(timeout.getTime(), timeout.getUnit())) {
+ fail(String.format("%s of %s threads not %s after %s",
+ state,
+ threads - latch.getCount(),
+ threads,
+ timeout
+ ));
+ }
+ } catch (InterruptedException e) {
+ fail(String.format("Interrupted while waiting %s state", "ready"));
+ }
+ }
+
+ public static class Run {
+ final int threads;
+ final Throwable[] exceptions;
+ final Timer.Time[] times;
+
+
+ public Run(final int threads, final Throwable[] exceptions, final Timer.Time[] times) {
+ this.threads = threads;
+ this.exceptions = exceptions;
+ this.times = times;
+ }
+
+ public Run assertNoExceptions() {
+ final long failed = Stream.of(exceptions)
+ .filter(Objects::nonNull)
+ .peek(Throwable::printStackTrace)
+ .count();
+ if (failed > 0) {
+ final long succeeded = threads - failed;
+ fail(String.format("Succeeded: %s, Failed: %s", succeeded, failed));
+ }
+ return this;
+ }
+
+ public Run assertExceptions(final Class<? extends Throwable> expected) {
+ for (final Throwable actual : exceptions) {
+ assertNotNull(actual);
+ try {
+ assertEquals(expected, actual.getClass());
+ } catch (AssertionError e) {
+ actual.printStackTrace();
+ throw e;
+ }
+ }
+ return this;
+ }
+
+ public Run assertTimesLessThan(final long time, final TimeUnit unit) {
+ for (final Timer.Time t : times) {
+ t.assertLessThan(time, unit);
+ }
+ return this;
+ }
+
+ public Run assertTimesGreaterThan(final long time, final TimeUnit unit) {
+ for (final Timer.Time t : times) {
+ t.assertGreaterThan(time, unit);
+ }
+ return this;
+ }
+ }
+ }
}