@Test public void testLockClientConcurrent() { DynoLockClient[] cs = new DynoLockClient[] {constructDynoLockClient(), constructDynoLockClient(), constructDynoLockClient()}; CopyOnWriteArrayList<DynoLockClient> clients = new CopyOnWriteArrayList<>(cs); List<Long> ttls = Arrays.asList(new Long[]{1000L, 500L, 250L}); AtomicInteger count = new AtomicInteger(3); Collections.shuffle(ttls); ConcurrentLinkedDeque<Long> ttlQueue = new ConcurrentLinkedDeque<>(ttls); List<Long> resultList = Collections.synchronizedList(new ArrayList()); Supplier<Tuple<Long, Long>> acquireLock = () -> { long ttl = ttlQueue.poll(); long value = clients.get(count.decrementAndGet()).acquireLock(resource, ttl); resultList.add(value); return new Tuple<>(ttl, value); }; IntStream.range(0, ttls.size()).mapToObj(i -> CompletableFuture.supplyAsync(acquireLock) .thenAccept(t -> Assert.assertTrue(t._2() < t._1()))).forEach(f -> { try { f.get(); } catch (InterruptedException e) { Assert.fail("Interrupted during the test"); } catch (ExecutionException e) { e.printStackTrace(); Assert.fail(); } }); boolean lock = false; for(Long r: resultList) { if(r > 0) { if(lock) { Assert.fail("Lock did not work as expected " + Arrays.toString(resultList.toArray())); } lock = true; } } }
public boolean acquireLock(String resource, long ttlMS, Consumer<String> failure) { return acquireLockWithExtension(resource, ttlMS, (r) -> { releaseLock(r); failure.accept(r); }); }
DynoLockClient { public boolean acquireLock(String resource, long ttlMS, Consumer<String> failure) { return acquireLockWithExtension(resource, ttlMS, (r) -> { releaseLock(r); failure.accept(r); }); } }
DynoLockClient { public boolean acquireLock(String resource, long ttlMS, Consumer<String> failure) { return acquireLockWithExtension(resource, ttlMS, (r) -> { releaseLock(r); failure.accept(r); }); } DynoLockClient(ConnectionPool pool, VotingHostsSelector votingHostsSelector, long timeout, TimeUnit unit); }
DynoLockClient { public boolean acquireLock(String resource, long ttlMS, Consumer<String> failure) { return acquireLockWithExtension(resource, ttlMS, (r) -> { releaseLock(r); failure.accept(r); }); } DynoLockClient(ConnectionPool pool, VotingHostsSelector votingHostsSelector, long timeout, TimeUnit unit); void setTimeoutUnit(TimeUnit timeoutUnit); void setTimeout(long timeout); List<String> getLockedResources(); void releaseLock(String resource); boolean acquireLock(String resource, long ttlMS, Consumer<String> failure); long acquireLock(String resource, long ttlMS); long checkLock(String resource); long extendLock(String resource, long ttlMS); void cleanup(); void logLocks(); }
DynoLockClient { public boolean acquireLock(String resource, long ttlMS, Consumer<String> failure) { return acquireLockWithExtension(resource, ttlMS, (r) -> { releaseLock(r); failure.accept(r); }); } DynoLockClient(ConnectionPool pool, VotingHostsSelector votingHostsSelector, long timeout, TimeUnit unit); void setTimeoutUnit(TimeUnit timeoutUnit); void setTimeout(long timeout); List<String> getLockedResources(); void releaseLock(String resource); boolean acquireLock(String resource, long ttlMS, Consumer<String> failure); long acquireLock(String resource, long ttlMS); long checkLock(String resource); long extendLock(String resource, long ttlMS); void cleanup(); void logLocks();  }