@Test public void testCheckLock() { long v = dynoLockClient.acquireLock(resource, 5000); Assert.assertTrue("Acquire lock did not succeed in time", v > 0); Assert.assertEquals(1, dynoLockClient.getLockedResources().size()); v = dynoLockClient.checkLock(resource); Assert.assertTrue("Check lock failed for acquired lock",v > 0); dynoLockClient.releaseLock(resource); Assert.assertTrue("Check lock failed for acquired lock", dynoLockClient.checkLock(resource) == 0); }
public long checkLock(String resource) { if (!checkResourceExists(resource)) { return 0; } else { long startTime = Instant.now().toEpochMilli(); CopyOnWriteArrayList<Long> resultTtls = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(quorum); votingHostsSelector.getVotingHosts().getEntireList().stream() .map(host -> new CheckAndRunHost(host, pool, "pttl", resource, resourceKeyMap.get(resource))) .forEach(checkAndRunHost -> CompletableFuture.supplyAsync(checkAndRunHost, service) .thenAccept(r -> { String result = r.getResult().toString(); if (result.equals("0") || result.equals("-2")) { logger.info("Lock not present on host"); } else { resultTtls.add(Long.valueOf(result)); latch.countDown(); } }) ); boolean latchValue = awaitLatch(latch, resource); if (latchValue) { long timeElapsed = Instant.now().toEpochMilli() - startTime; logger.info("Checked lock on {}", resource); return Collections.min(resultTtls) - timeElapsed; } else { logger.info("Timed out before we could check the lock"); return 0; } } }
DynoLockClient { public long checkLock(String resource) { if (!checkResourceExists(resource)) { return 0; } else { long startTime = Instant.now().toEpochMilli(); CopyOnWriteArrayList<Long> resultTtls = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(quorum); votingHostsSelector.getVotingHosts().getEntireList().stream() .map(host -> new CheckAndRunHost(host, pool, "pttl", resource, resourceKeyMap.get(resource))) .forEach(checkAndRunHost -> CompletableFuture.supplyAsync(checkAndRunHost, service) .thenAccept(r -> { String result = r.getResult().toString(); if (result.equals("0") || result.equals("-2")) { logger.info("Lock not present on host"); } else { resultTtls.add(Long.valueOf(result)); latch.countDown(); } }) ); boolean latchValue = awaitLatch(latch, resource); if (latchValue) { long timeElapsed = Instant.now().toEpochMilli() - startTime; logger.info("Checked lock on {}", resource); return Collections.min(resultTtls) - timeElapsed; } else { logger.info("Timed out before we could check the lock"); return 0; } } } }
DynoLockClient { public long checkLock(String resource) { if (!checkResourceExists(resource)) { return 0; } else { long startTime = Instant.now().toEpochMilli(); CopyOnWriteArrayList<Long> resultTtls = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(quorum); votingHostsSelector.getVotingHosts().getEntireList().stream() .map(host -> new CheckAndRunHost(host, pool, "pttl", resource, resourceKeyMap.get(resource))) .forEach(checkAndRunHost -> CompletableFuture.supplyAsync(checkAndRunHost, service) .thenAccept(r -> { String result = r.getResult().toString(); if (result.equals("0") || result.equals("-2")) { logger.info("Lock not present on host"); } else { resultTtls.add(Long.valueOf(result)); latch.countDown(); } }) ); boolean latchValue = awaitLatch(latch, resource); if (latchValue) { long timeElapsed = Instant.now().toEpochMilli() - startTime; logger.info("Checked lock on {}", resource); return Collections.min(resultTtls) - timeElapsed; } else { logger.info("Timed out before we could check the lock"); return 0; } } } DynoLockClient(ConnectionPool pool, VotingHostsSelector votingHostsSelector, long timeout, TimeUnit unit); }
DynoLockClient { public long checkLock(String resource) { if (!checkResourceExists(resource)) { return 0; } else { long startTime = Instant.now().toEpochMilli(); CopyOnWriteArrayList<Long> resultTtls = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(quorum); votingHostsSelector.getVotingHosts().getEntireList().stream() .map(host -> new CheckAndRunHost(host, pool, "pttl", resource, resourceKeyMap.get(resource))) .forEach(checkAndRunHost -> CompletableFuture.supplyAsync(checkAndRunHost, service) .thenAccept(r -> { String result = r.getResult().toString(); if (result.equals("0") || result.equals("-2")) { logger.info("Lock not present on host"); } else { resultTtls.add(Long.valueOf(result)); latch.countDown(); } }) ); boolean latchValue = awaitLatch(latch, resource); if (latchValue) { long timeElapsed = Instant.now().toEpochMilli() - startTime; logger.info("Checked lock on {}", resource); return Collections.min(resultTtls) - timeElapsed; } else { logger.info("Timed out before we could check the lock"); return 0; } } } 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 long checkLock(String resource) { if (!checkResourceExists(resource)) { return 0; } else { long startTime = Instant.now().toEpochMilli(); CopyOnWriteArrayList<Long> resultTtls = new CopyOnWriteArrayList<>(); CountDownLatch latch = new CountDownLatch(quorum); votingHostsSelector.getVotingHosts().getEntireList().stream() .map(host -> new CheckAndRunHost(host, pool, "pttl", resource, resourceKeyMap.get(resource))) .forEach(checkAndRunHost -> CompletableFuture.supplyAsync(checkAndRunHost, service) .thenAccept(r -> { String result = r.getResult().toString(); if (result.equals("0") || result.equals("-2")) { logger.info("Lock not present on host"); } else { resultTtls.add(Long.valueOf(result)); latch.countDown(); } }) ); boolean latchValue = awaitLatch(latch, resource); if (latchValue) { long timeElapsed = Instant.now().toEpochMilli() - startTime; logger.info("Checked lock on {}", resource); return Collections.min(resultTtls) - timeElapsed; } else { logger.info("Timed out before we could check the lock"); return 0; } } } 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();  }