MongoLockService implements LockService { @Override public boolean acquireLock() { if (getLockCollection().count() == 0) { throw new IllegalStateException( "Cannot acquire lock because lock table has not been initialized. Initialize MongoLockService properly, or call destroyLock"); } final Document lockDocument = new LockEntryDocBuilder().lock().build(); final Document result = getLockCollection() .findOneAndUpdate(new Document(LockEntryDocBuilder.LOCK_ENTRY, true), new Document("$set", lockDocument)); return result != null && result.getBoolean(LockEntryDocBuilder.LOCKED, true) == false; } MongoLockService(MongoDatabase database); @Override boolean acquireLock(); @Override boolean releaseLock(); @Override LockEntry getLockInformation(); @Override void destroyLock(); static final String MIGRAMONGO_LOCK_COLLECTION; }
@Test public void onlyOneThreadShouldAcquireLockWhenRunningConcurrently() throws Exception { final int numberOfWorkers = 10; final ExecutorService executor = Executors.newFixedThreadPool(numberOfWorkers); final AtomicInteger locksAcquired = new AtomicInteger(0); final AtomicInteger locksRejected = new AtomicInteger(0); for (int i = 0; i < numberOfWorkers; i++) { executor.submit( () -> lockService.acquireLock() ? locksAcquired.incrementAndGet() : locksRejected.incrementAndGet()); } executor.shutdown(); executor.awaitTermination(5, TimeUnit.SECONDS); assertThat(locksAcquired.get(), is(1)); assertThat(locksRejected.get(), is(numberOfWorkers - 1)); }
@Test public void onlyOneThreadShouldAcquireLockWhenRunningConcurrentlyAndLockObjectsAreCreatedConcurrently() throws Exception { final int numberOfWorkers = 10; final ExecutorService executor = Executors.newFixedThreadPool(numberOfWorkers); final AtomicInteger locksAcquired = new AtomicInteger(0); final AtomicInteger locksRejected = new AtomicInteger(0); for (int i = 0; i < numberOfWorkers; i++) { executor.submit(() -> { final LockService thisLockService = new MongoLockService(database); return thisLockService.acquireLock() ? locksAcquired.incrementAndGet() : locksRejected.incrementAndGet(); }); } executor.shutdown(); executor.awaitTermination(5, TimeUnit.SECONDS); assertThat(locksAcquired.get(), is(1)); assertThat(locksRejected.get(), is(numberOfWorkers - 1)); }
@Test public void shouldThrowIllegalStateExceptionIfLockServiceIsNotInitialized() { database.getCollection(MongoLockService.MIGRAMONGO_LOCK_COLLECTION).drop(); assertThrows(IllegalStateException.class, () -> lockService.acquireLock()); }
@Test public void shouldAcquireLock() throws Exception { assertThat(lockService.acquireLock(), is(true)); }
@Test public void shouldNotAcquireLockTwice() throws Exception { lockService.acquireLock(); assertThat(lockService.acquireLock(), is(false)); }