mirror of https://github.com/apache/jclouds.git
fix issue #1205: removed copied in tests from Suppliers.memoize as current code is no longer a derivative of that
This commit is contained in:
Adrian Cole
parent
c395d90928
commit
e27d9e94fe
|
|
@ -29,59 +29,51 @@ import java.util.concurrent.atomic.AtomicReference;
|
|||
import org.jclouds.rest.AuthorizationException;
|
||||
|
||||
import com.google.common.base.Objects;
|
||||
import com.google.common.base.Optional;
|
||||
import com.google.common.base.Supplier;
|
||||
import com.google.common.cache.CacheBuilder;
|
||||
import com.google.common.cache.CacheLoader;
|
||||
import com.google.common.cache.LoadingCache;
|
||||
import com.google.common.collect.ForwardingObject;
|
||||
import com.google.common.util.concurrent.UncheckedExecutionException;
|
||||
|
||||
/**
|
||||
* This will retry the supplier if it encounters a timeout exception, but not if it encounters an
|
||||
* AuthorizationException.
|
||||
* <p/>
|
||||
* A shared exception reference is used so that anyone who encounters an authorizationexception will
|
||||
* be short-circuited. This prevents accounts from being locked out.
|
||||
* A shared exception reference is used so that anyone who encounters an authorizationexception will be short-circuited.
|
||||
* This prevents accounts from being locked out.
|
||||
*
|
||||
* <h3>details</h3>
|
||||
* http://code.google.com/p/google-guice/issues/detail?id=483 guice doesn't remember when singleton
|
||||
* providers throw exceptions. in this case, if the supplier fails with an authorization exception,
|
||||
* it is called again for each provider method that depends on it. To short-circuit this, we
|
||||
* remember the last exception trusting that guice is single-threaded.
|
||||
* http://code.google.com/p/google-guice/issues/detail?id=483 guice doesn't remember when singleton providers throw
|
||||
* exceptions. in this case, if the supplier fails with an authorization exception, it is called again for each provider
|
||||
* method that depends on it. To short-circuit this, we remember the last exception trusting that guice is
|
||||
* single-threaded.
|
||||
*
|
||||
* Note this implementation is folded into the same class, vs being decorated as stacktraces are
|
||||
* exceptionally long and difficult to grok otherwise. We use {@link LoadingCache} to deal with
|
||||
* concurrency issues related to the supplier.
|
||||
* Note this implementation is folded into the same class, vs being decorated as stacktraces are exceptionally long and
|
||||
* difficult to grok otherwise. We use {@link LoadingCache} to deal with concurrency issues related to the supplier.
|
||||
*
|
||||
* @author Adrian Cole
|
||||
*/
|
||||
public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T> extends ForwardingObject implements
|
||||
Supplier<T> {
|
||||
Supplier<T> {
|
||||
|
||||
static class NullValueException extends RuntimeException {
|
||||
|
||||
}
|
||||
|
||||
static class SetAndThrowAuthorizationExceptionSupplierBackedLoader<V> extends CacheLoader<String, V> {
|
||||
static class SetAndThrowAuthorizationExceptionSupplierBackedLoader<V> extends CacheLoader<String, Optional<V>> {
|
||||
|
||||
private final Supplier<V> delegate;
|
||||
private final AtomicReference<AuthorizationException> authException;
|
||||
|
||||
public SetAndThrowAuthorizationExceptionSupplierBackedLoader(Supplier<V> delegate,
|
||||
AtomicReference<AuthorizationException> authException) {
|
||||
AtomicReference<AuthorizationException> authException) {
|
||||
this.delegate = checkNotNull(delegate, "delegate");
|
||||
this.authException = checkNotNull(authException, "authException");
|
||||
}
|
||||
|
||||
@Override
|
||||
public V load(String key) {
|
||||
public Optional<V> load(String key) {
|
||||
if (authException.get() != null)
|
||||
throw authException.get();
|
||||
try {
|
||||
V value = delegate.get();
|
||||
if (value == null)
|
||||
throw new NullValueException();
|
||||
return value;
|
||||
return Optional.fromNullable(delegate.get());
|
||||
} catch (Exception e) {
|
||||
AuthorizationException aex = getFirstThrowableOfType(e, AuthorizationException.class);
|
||||
if (aex != null) {
|
||||
|
|
@ -102,21 +94,21 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T> ext
|
|||
private final Supplier<T> delegate;
|
||||
private final long duration;
|
||||
private final TimeUnit unit;
|
||||
private final LoadingCache<String, T> cache;
|
||||
private final LoadingCache<String, Optional<T>> cache;
|
||||
|
||||
public static <T> MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T> create(
|
||||
AtomicReference<AuthorizationException> authException, Supplier<T> delegate, long duration, TimeUnit unit) {
|
||||
AtomicReference<AuthorizationException> authException, Supplier<T> delegate, long duration, TimeUnit unit) {
|
||||
return new MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T>(authException, delegate, duration,
|
||||
unit);
|
||||
unit);
|
||||
}
|
||||
|
||||
MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier(AtomicReference<AuthorizationException> authException,
|
||||
Supplier<T> delegate, long duration, TimeUnit unit) {
|
||||
Supplier<T> delegate, long duration, TimeUnit unit) {
|
||||
this.delegate = delegate;
|
||||
this.duration = duration;
|
||||
this.unit = unit;
|
||||
this.cache = CacheBuilder.newBuilder().expireAfterWrite(duration, unit)
|
||||
.build(new SetAndThrowAuthorizationExceptionSupplierBackedLoader<T>(delegate, authException));
|
||||
.build(new SetAndThrowAuthorizationExceptionSupplierBackedLoader<T>(delegate, authException));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -127,14 +119,7 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T> ext
|
|||
@Override
|
||||
public T get() {
|
||||
try {
|
||||
T returnVal = cache.get("FOO");
|
||||
return returnVal;
|
||||
} catch (UncheckedExecutionException e) {
|
||||
NullValueException nullV = getFirstThrowableOfType(e, NullValueException.class);
|
||||
if (nullV != null) {
|
||||
return null;
|
||||
}
|
||||
throw propagate(e.getCause());
|
||||
return cache.get("FOO").orNull();
|
||||
} catch (ExecutionException e) {
|
||||
throw propagate(e.getCause());
|
||||
}
|
||||
|
|
@ -143,7 +128,7 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier<T> ext
|
|||
@Override
|
||||
public String toString() {
|
||||
return Objects.toStringHelper(this).add("delegate", delegate).add("duration", duration).add("unit", unit)
|
||||
.toString();
|
||||
.toString();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,52 +18,38 @@
|
|||
*/
|
||||
package org.jclouds.rest.suppliers;
|
||||
|
||||
import static com.google.common.base.Preconditions.checkNotNull;
|
||||
import static com.google.common.base.Suppliers.ofInstance;
|
||||
import static com.google.common.util.concurrent.Atomics.newReference;
|
||||
import static org.testng.Assert.assertEquals;
|
||||
import static org.testng.Assert.assertSame;
|
||||
|
||||
import java.io.ByteArrayInputStream;
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.ObjectInputStream;
|
||||
import java.io.ObjectOutputStream;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.jclouds.rest.AuthorizationException;
|
||||
import org.jclouds.rest.suppliers.MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier.SetAndThrowAuthorizationExceptionSupplierBackedLoader;
|
||||
import org.testng.annotations.Test;
|
||||
|
||||
import com.google.common.base.Function;
|
||||
import com.google.common.base.Supplier;
|
||||
import com.google.common.base.Suppliers;
|
||||
import com.google.common.util.concurrent.Atomics;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Adrian Cole
|
||||
*/
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
@Test(groups = "unit", testName = "MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplierTest")
|
||||
public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplierTest {
|
||||
@Test
|
||||
public void testLoaderNormal() {
|
||||
AtomicReference<AuthorizationException> authException = Atomics.newReference();
|
||||
assertEquals(new SetAndThrowAuthorizationExceptionSupplierBackedLoader<String>(Suppliers.ofInstance("foo"),
|
||||
authException).load("KEY"), "foo");
|
||||
AtomicReference<AuthorizationException> authException = newReference();
|
||||
assertEquals(new SetAndThrowAuthorizationExceptionSupplierBackedLoader<String>(ofInstance("foo"),
|
||||
authException).load("KEY").get(), "foo");
|
||||
assertEquals(authException.get(), null);
|
||||
}
|
||||
|
||||
@Test(expectedExceptions = AuthorizationException.class)
|
||||
public void testLoaderThrowsAuthorizationExceptionAndAlsoSetsExceptionType() {
|
||||
AtomicReference<AuthorizationException> authException = Atomics.newReference();
|
||||
AtomicReference<AuthorizationException> authException = newReference();
|
||||
try {
|
||||
new SetAndThrowAuthorizationExceptionSupplierBackedLoader<String>(new Supplier<String>() {
|
||||
|
||||
@Override
|
||||
public String get() {
|
||||
throw new AuthorizationException();
|
||||
}
|
||||
|
|
@ -75,11 +61,9 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplierTest {
|
|||
|
||||
@Test(expectedExceptions = AuthorizationException.class)
|
||||
public void testLoaderThrowsAuthorizationExceptionAndAlsoSetsExceptionTypeWhenNested() {
|
||||
AtomicReference<AuthorizationException> authException = Atomics.newReference();
|
||||
AtomicReference<AuthorizationException> authException = newReference();
|
||||
try {
|
||||
new SetAndThrowAuthorizationExceptionSupplierBackedLoader<String>(new Supplier<String>() {
|
||||
|
||||
@Override
|
||||
public String get() {
|
||||
throw new RuntimeException(new ExecutionException(new AuthorizationException()));
|
||||
}
|
||||
|
|
@ -91,11 +75,9 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplierTest {
|
|||
|
||||
@Test(expectedExceptions = RuntimeException.class)
|
||||
public void testLoaderThrowsOriginalExceptionAndAlsoSetsExceptionTypeWhenNestedAndNotAuthorizationException() {
|
||||
AtomicReference<AuthorizationException> authException = Atomics.newReference();
|
||||
AtomicReference<AuthorizationException> authException = newReference();
|
||||
try {
|
||||
new SetAndThrowAuthorizationExceptionSupplierBackedLoader<String>(new Supplier<String>() {
|
||||
|
||||
@Override
|
||||
public String get() {
|
||||
throw new RuntimeException(new IllegalArgumentException("foo"));
|
||||
}
|
||||
|
|
@ -104,163 +86,4 @@ public class MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplierTest {
|
|||
assertEquals(authException.get().getClass(), RuntimeException.class);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMemoizeKeepsValueForFullDurationWhenDelegateCallIsSlow() {
|
||||
final long SLEEP_TIME = 250;
|
||||
final long EXPIRATION_TIME = 200;
|
||||
|
||||
Supplier<Integer> slowSupplier = new CountingSupplier() {
|
||||
|
||||
@Override
|
||||
public Integer get() {
|
||||
try {
|
||||
Thread.sleep(SLEEP_TIME);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
return super.get();
|
||||
}
|
||||
};
|
||||
|
||||
Supplier<Integer> memoizedSupplier = new MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier(
|
||||
new AtomicReference<AuthorizationException>(), slowSupplier, EXPIRATION_TIME, TimeUnit.MILLISECONDS);
|
||||
|
||||
assertEquals(memoizedSupplier.get(), (Integer) 10);
|
||||
assertEquals(memoizedSupplier.get(), (Integer) 10);
|
||||
}
|
||||
|
||||
// =================================
|
||||
//
|
||||
// TODO Everything below this point is taken from SuppliersTest, to test our version of the
|
||||
// Suppliers2.memoizeWithExpiration
|
||||
// It should be deleted when we can switch back to using the google
|
||||
// Supplier.memoizeWithExpiration.
|
||||
|
||||
private static class CountingSupplier implements Supplier<Integer> {
|
||||
transient int calls = 0;
|
||||
|
||||
@Override
|
||||
public Integer get() {
|
||||
calls++;
|
||||
return calls * 10;
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMemoizeWithExpiration() throws InterruptedException {
|
||||
CountingSupplier countingSupplier = new CountingSupplier();
|
||||
|
||||
Supplier<Integer> memoizedSupplier = new MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier(
|
||||
new AtomicReference<AuthorizationException>(), countingSupplier, 75, TimeUnit.MILLISECONDS);
|
||||
|
||||
checkExpiration(countingSupplier, memoizedSupplier);
|
||||
}
|
||||
|
||||
private void checkExpiration(CountingSupplier countingSupplier, Supplier<Integer> memoizedSupplier)
|
||||
throws InterruptedException {
|
||||
// the underlying supplier hasn't executed yet
|
||||
assertEquals(0, countingSupplier.calls);
|
||||
|
||||
assertEquals(10, (int) memoizedSupplier.get());
|
||||
// now it has
|
||||
assertEquals(1, countingSupplier.calls);
|
||||
|
||||
assertEquals(10, (int) memoizedSupplier.get());
|
||||
// it still should only have executed once due to memoization
|
||||
assertEquals(1, countingSupplier.calls);
|
||||
|
||||
Thread.sleep(150);
|
||||
|
||||
assertEquals(20, (int) memoizedSupplier.get());
|
||||
// old value expired
|
||||
assertEquals(2, countingSupplier.calls);
|
||||
|
||||
assertEquals(20, (int) memoizedSupplier.get());
|
||||
// it still should only have executed twice due to memoization
|
||||
assertEquals(2, countingSupplier.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExpiringMemoizedSupplierThreadSafe() throws Throwable {
|
||||
Function<Supplier<Boolean>, Supplier<Boolean>> memoizer = new Function<Supplier<Boolean>, Supplier<Boolean>>() {
|
||||
@Override
|
||||
public Supplier<Boolean> apply(Supplier<Boolean> supplier) {
|
||||
return new MemoizedRetryOnTimeOutButNotOnAuthorizationExceptionSupplier(
|
||||
new AtomicReference<AuthorizationException>(), supplier, Long.MAX_VALUE, TimeUnit.NANOSECONDS);
|
||||
}
|
||||
};
|
||||
supplierThreadSafe(memoizer);
|
||||
}
|
||||
|
||||
private void supplierThreadSafe(Function<Supplier<Boolean>, Supplier<Boolean>> memoizer) throws Throwable {
|
||||
final AtomicInteger count = new AtomicInteger(0);
|
||||
final AtomicReference<Throwable> thrown = Atomics.newReference(null);
|
||||
final int numThreads = 3;
|
||||
final Thread[] threads = new Thread[numThreads];
|
||||
final long timeout = TimeUnit.SECONDS.toNanos(60);
|
||||
|
||||
final Supplier<Boolean> supplier = new Supplier<Boolean>() {
|
||||
boolean isWaiting(Thread thread) {
|
||||
switch (thread.getState()) {
|
||||
case BLOCKED:
|
||||
case WAITING:
|
||||
case TIMED_WAITING:
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
int waitingThreads() {
|
||||
int waitingThreads = 0;
|
||||
for (Thread thread : threads) {
|
||||
if (isWaiting(thread)) {
|
||||
waitingThreads++;
|
||||
}
|
||||
}
|
||||
return waitingThreads;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Boolean get() {
|
||||
// Check that this method is called exactly once, by the first
|
||||
// thread to synchronize.
|
||||
long t0 = System.nanoTime();
|
||||
while (waitingThreads() != numThreads - 1) {
|
||||
if (System.nanoTime() - t0 > timeout) {
|
||||
thrown.set(new TimeoutException("timed out waiting for other threads to block"
|
||||
+ " synchronizing on supplier"));
|
||||
break;
|
||||
}
|
||||
Thread.yield();
|
||||
}
|
||||
count.getAndIncrement();
|
||||
return Boolean.TRUE;
|
||||
}
|
||||
};
|
||||
|
||||
final Supplier<Boolean> memoizedSupplier = memoizer.apply(supplier);
|
||||
|
||||
for (int i = 0; i < numThreads; i++) {
|
||||
threads[i] = new Thread() {
|
||||
@Override
|
||||
public void run() {
|
||||
assertSame(Boolean.TRUE, memoizedSupplier.get());
|
||||
}
|
||||
};
|
||||
}
|
||||
for (Thread t : threads) {
|
||||
t.start();
|
||||
}
|
||||
for (Thread t : threads) {
|
||||
t.join();
|
||||
}
|
||||
|
||||
if (thrown.get() != null) {
|
||||
throw thrown.get();
|
||||
}
|
||||
assertEquals(1, count.get());
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue