AsyncTimeLockUnlocker implements TimeLockUnlocker, AutoCloseable { @Override @SuppressWarnings("FutureReturnValueIgnored") public void enqueue(Set<LockToken> tokens) { autobatcher.apply(tokens); } private  AsyncTimeLockUnlocker(DisruptorAutobatcher<Set<LockToken>, Void> autobatcher); static AsyncTimeLockUnlocker create(TimelockService timelockService); @Override @SuppressWarnings("FutureReturnValueIgnored") void enqueue(Set<LockToken> tokens); @Override void close();  }
@Test(timeout = 2_000) public void enqueueDoesNotBlock() { doAnswer(invocation -> { Uninterruptibles.sleepUninterruptibly(30, TimeUnit.SECONDS); return null; }).when(timelockService).tryUnlock(any()); unlocker.enqueue(ImmutableSet.copyOf(tokenList)); }
@Test public void enqueueingTokensEnqueuedSinglyAllEventuallyCaptured() { setupTokenCollectingTimeLock(); tokenList.forEach(token -> unlocker.enqueue(ImmutableSet.of(token))); verifyTryUnlockAttemptedAtLeastOnce(); assertAllTokensEventuallyUnlocked(); }
@Test public void enqueueingTokensEnqueuedInGroupsAllEventuallyCaptured() { setupTokenCollectingTimeLock(); Lists.partition(tokenList, 7).forEach(tokens -> unlocker.enqueue(ImmutableSet.copyOf(tokens))); verifyTryUnlockAttemptedAtLeastOnce(); assertAllTokensEventuallyUnlocked(); }
@SuppressWarnings("unchecked") @Test public void noParallelCallsMadeFromTimelockPointOfView() { AtomicBoolean timelockInUse = new AtomicBoolean(false); AtomicBoolean concurrentlyInvoked = new AtomicBoolean(false); doAnswer(invocation -> { if (timelockInUse.get()) { concurrentlyInvoked.set(true); } timelockInUse.set(true); unlockedTokens.addAll((Set<LockToken>) invocation.getArguments()[0]); timelockInUse.set(false); return null; }).when(timelockService).tryUnlock(any()); tokenList.forEach(token -> unlocker.enqueue(ImmutableSet.of(token))); verifyTryUnlockAttemptedAtLeastOnce(); assertAllTokensEventuallyUnlocked(); assertThat(concurrentlyInvoked.get()).as("TimeLock was, at some point, called concurrently").isFalse(); }