diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java index ad0e7e594c..d04d946f69 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java @@ -22,6 +22,7 @@ package org.springframework.integration.metadata; * invoked when changes occur in the metadata store. * * @author Marius Bogoevici + * @since 4.2 */ public interface ListenableMetadataStore extends ConcurrentMetadataStore { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java index ec53c1dd87..9c2003b7fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java @@ -20,6 +20,7 @@ package org.springframework.integration.metadata; * A callback to be invoked whenever a value changes in the data store. * * @author Marius Bogoevici + * @since 4.2 */ public interface MetadataStoreListener { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/locks/ExpirableLockRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/ExpirableLockRegistry.java new file mode 100644 index 0000000000..2c723b796b --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/locks/ExpirableLockRegistry.java @@ -0,0 +1,35 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.support.locks; + +/** + * A {@link LockRegistry} implementing this interface supports the removal of aged locks + * that are not currently locked. + * + * @author Gary Russell + * @since 4.2 + * + */ +public interface ExpirableLockRegistry extends LockRegistry { + + /** + * Remove locks last acquired more than 'age' ago that are not currently locked. + * @param age the time since the lock was last obtained. + * @throws IllegalStateException if the registry configuration does not support this feature. + */ + void expireUnusedOlderThan(long age); + +} diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java index 43c4c7dfbc..b02c66c3de 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java @@ -29,10 +29,11 @@ import org.springframework.util.ObjectUtils; * Mainly useful in conjunction with retrying matcherss such as {@link EventuallyMatcher} * * @author Marius Bogoevici + * @since 4.2 */ public class EqualsResultMatcher extends DiagnosingMatcher { - private Evaluator evaluator; + private final Evaluator evaluator; public EqualsResultMatcher(Evaluator evaluator) { this.evaluator = evaluator; diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java index 80eca5c0d5..079cccc65d 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java @@ -29,6 +29,7 @@ import org.hamcrest.Matcher; * (Copied from {@code org.springframework.xd.test.fixtures.EventuallyMatcher}) * * @author Eric Bottard + * @since 4.2 */ public class EventuallyMatcher extends DiagnosingMatcher { diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java new file mode 100644 index 0000000000..7a6ca00000 --- /dev/null +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/lock/ZookeeperLockRegistry.java @@ -0,0 +1,262 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.zookeeper.lock; + +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Condition; +import java.util.concurrent.locks.Lock; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.locks.InterProcessMutex; + +import org.springframework.integration.support.locks.ExpirableLockRegistry; +import org.springframework.messaging.MessagingException; +import org.springframework.util.Assert; + +/** + * {@link ExpirableLockRegistry} implementation using Zookeeper, or more specifically, + * Curator {@link InterProcessMutex}. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class ZookeeperLockRegistry implements ExpirableLockRegistry { + + private static final String DEFAULT_ROOT = "/SpringIntegration-LockRegistry"; + + private final CuratorFramework client; + + private final KeyToPathStrategy keyToPath; + + private final Map locks = new HashMap(); + + private final boolean trackingTime; + + /** + * Construct a lock registry using the default {@link KeyToPathStrategy} which + * simple appends the key to '/SpringIntegration-LockRegistry/'. + * @param client the {@link CuratorFramework}. + */ + public ZookeeperLockRegistry(CuratorFramework client) { + this(client, DEFAULT_ROOT); + } + + /** + * Construct a lock registry using the default {@link KeyToPathStrategy} which + * simple appends the key to {@code '/'}. + * @param client the {@link CuratorFramework}. + * @param path the path root (no trailing /). + */ + public ZookeeperLockRegistry(CuratorFramework client, String root) { + this(client, new DefaultKeyToPathStrategy(root)); + } + + /** + * Construct a lock registry using the supplied {@link KeyToPathStrategy}. + * @param client the {@link CuratorFramework}. + * @param keyToPath the implementation of {@link KeyToPathStrategy}. + */ + public ZookeeperLockRegistry(CuratorFramework client, KeyToPathStrategy keyToPath) { + Assert.notNull(client, "'client' cannot be null"); + Assert.notNull(client, "'keyToPath' cannot be null"); + this.client = client; + this.keyToPath = keyToPath; + this.trackingTime = !keyToPath.bounded(); + } + + @Override + public Lock obtain(Object lockKey) { + Assert.isInstanceOf(String.class, lockKey); + String path = this.keyToPath.pathFor((String) lockKey); + ZkLock lock = this.locks.get(path); + if (lock == null) { + synchronized (this.locks) { + lock = this.locks.get(path); + if (lock == null) { + lock = new ZkLock(this.client, path); + this.locks.put(path, lock); + } + if (this.trackingTime) { + lock.setLastUsed(System.currentTimeMillis()); + } + } + } + return lock; + } + + /** + * Remove locks last acquired more than 'age' ago that are not currently locked. + * Expiry is not supported if the {@link KeyToPathStrategy} is bounded (returns a finite + * number of paths). With such a {@link KeyToPathStrategy}, the overhead of tracking when + * a lock is obtained is avoided. + * @param age the time since the lock was last obtained. + */ + @Override + public void expireUnusedOlderThan(long age) { + if (!this.trackingTime) { + throw new IllegalStateException("Ths KeyToPathStrategy is bounded; expiry is not supported"); + } + synchronized(this.locks) { + Iterator> iterator = this.locks.entrySet().iterator(); + long now = System.currentTimeMillis(); + while(iterator.hasNext()) { + Entry entry = iterator.next(); + ZkLock lock = entry.getValue(); + if (now - lock.getLastUsed() > age + && !lock.isAcquiredInThisProcess()) { + iterator.remove(); + } + } + } + } + + /** + * Strategy to convert a lock key (e.g. aggregation correlation id) to a + * Zookeeper path. + * + */ + public interface KeyToPathStrategy { + + /** + * Return the path for the key. + * @param key the key. + * @return the path. + */ + String pathFor(String key); + + /** + * @return true if this strategy returns a bounded number of locks, removing + * the need for removing LRU locks. + */ + boolean bounded(); + + } + + private static class DefaultKeyToPathStrategy implements KeyToPathStrategy { + + private final String root; + + public DefaultKeyToPathStrategy(String rootPath) { + Assert.notNull(rootPath, "'rootPath' cannot be null"); + if (!rootPath.endsWith("/")) { + this.root = rootPath + "/"; + } + else { + this.root = rootPath; + } + } + + @Override + public String pathFor(String key) { + return root + key; + } + + @Override + public boolean bounded() { + return false; + } + + } + + private static class ZkLock implements Lock { + + private final InterProcessMutex mutex; + + private final String path; + + private long lastUsed; + + public ZkLock(CuratorFramework client, String path) { + this.mutex = new InterProcessMutex(client, path); + this.path = path; + } + + public long getLastUsed() { + return this.lastUsed; + } + + public void setLastUsed(long lastUsed) { + this.lastUsed = lastUsed; + } + + @Override + public void lock() { + try { + this.mutex.acquire(); + } + catch (Exception e) { + throw new RuntimeException("Failed to aquire mutex at " + this.path, e); + } + } + + @Override + public void lockInterruptibly() throws InterruptedException { + boolean locked = false; + // this is a bit ugly, but... + while (!locked) { + locked = tryLock(1, TimeUnit.SECONDS); + } + + } + + @Override + public boolean tryLock() { + try { + return tryLock(0, TimeUnit.MICROSECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return false; + } + } + + @Override + public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { + try { + return this.mutex.acquire(time, unit); + } + catch (Exception e) { + throw new MessagingException("Failed to aquire mutex at " + this.path, e); + } + } + + @Override + public void unlock() { + try { + this.mutex.release(); + } + catch (Exception e) { + throw new MessagingException("Failed to release mutex at " + this.path, e); + } + } + + @Override + public Condition newCondition() { + throw new UnsupportedOperationException("Conditions are not supported"); + } + + public boolean isAcquiredInThisProcess() { + return this.mutex.isAcquiredInThisProcess(); + } + + } + +} diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java index cd7c585b26..02a75797ed 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java @@ -42,6 +42,7 @@ import org.springframework.util.Assert; * the names of which are stored as keys. * * @author Marius Bogoevici + * @since 4.2 */ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLifecycle { @@ -320,9 +321,9 @@ public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLif private static class LocalChildData { - private String value; + private final String value; - private int version; + private final int version; public LocalChildData(String value, int version) { this.value = value; diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java index 9d2be5d3f3..1bafbc8bce 100644 --- a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java @@ -18,6 +18,7 @@ package org.springframework.integration.zookeeper.metadata; /** * @author Marius Bogoevici + * @since 4.2 */ @SuppressWarnings("serial") public class ZookeeperMetadataStoreException extends RuntimeException { diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java new file mode 100644 index 0000000000..b7a81dde43 --- /dev/null +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/ZookeeperTestSupport.java @@ -0,0 +1,92 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.zookeeper; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.CuratorFrameworkFactory; +import org.apache.curator.retry.BoundedExponentialBackoffRetry; +import org.apache.curator.test.TestingServer; +import org.apache.curator.utils.CloseableUtils; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; + +/** + * @author Marius Bogoevici + * @author Gary Russell + * @since 4.2 + * + */ +public class ZookeeperTestSupport { + + private static final Log logger = LogFactory.getLog(ZookeeperTestSupport.class); + + protected final Log log = LogFactory.getLog(this.getClass()); + + protected static TestingServer testingServer; + + protected CuratorFramework client; + + @BeforeClass + public static void setUpClass() throws Exception { + testingServer = new TestingServer(true); + } + + @AfterClass + public static void tearDownClass() throws Exception { + try { + testingServer.stop(); + } + catch (IOException e) { + logger.warn("Exception thrown while shutting down ZooKeeper: ", e); + } + testingServer.getTempDirectory().delete(); + } + + @Before + public void setUp() throws Exception{ + client = createNewClient(); + } + + @After + public void tearDown() throws Exception { + CloseableUtils.closeQuietly(this.client); + } + + protected CuratorFramework createNewClient() throws InterruptedException { + CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(), + new BoundedExponentialBackoffRetry(100, 1000, 3)); + client.start(); + client.blockUntilConnected(10000, TimeUnit.SECONDS); + return client; + } + + protected void closeClient(CuratorFramework client) { + try { + CloseableUtils.closeQuietly(client); + } + catch (Exception e) { + log.warn("Exception thrown while closing client: ", e); + } + } + +} diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java new file mode 100644 index 0000000000..13129c5b07 --- /dev/null +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/lock/ZkLockRegistryTests.java @@ -0,0 +1,313 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.zookeeper.lock; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; + +import org.junit.Test; + +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.zookeeper.ZookeeperTestSupport; +import org.springframework.integration.zookeeper.lock.ZookeeperLockRegistry.KeyToPathStrategy; +import org.springframework.messaging.MessagingException; + + + +/** + * @author Gary Russell + * @since 4.2 + * + */ +public class ZkLockRegistryTests extends ZookeeperTestSupport { + + @Test + public void testLock() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo"); + lock.lock(); + try { + assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + } + finally { + lock.unlock(); + } + } + + Thread.sleep(10); + registry.expireUnusedOlderThan(0); + assertEquals(0, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + } + + @Test + public void testLockInterruptibly() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo"); + lock.lockInterruptibly(); + try { + assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + } + finally { + lock.unlock(); + } + } + } + + @Test + public void testReentrantLock() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lock(); + try { + Lock lock2 = registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lock(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testReentrantLockInterruptibly() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = registry.obtain("foo"); + assertSame(lock1, lock2); + lock2.lockInterruptibly(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testTwoLocks() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + for (int i = 0; i < 10; i++) { + Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + try { + Lock lock2 = registry.obtain("bar"); + assertNotSame(lock1, lock2); + lock2.lockInterruptibly(); + lock2.unlock(); + } + finally { + lock1.unlock(); + } + } + } + + @Test + public void testTwoThreadsSecondFailsToGetLock() throws Exception { + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + final Lock lock1 = registry.obtain("foo"); + lock1.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + Lock lock2 = registry.obtain("foo"); + locked.set(lock2.tryLock(200, TimeUnit.MILLISECONDS)); + latch.countDown(); + try { + lock2.unlock(); + } + catch (MessagingException e) { + return e.getCause(); + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + Object ise = result.get(10, TimeUnit.SECONDS); + assertThat(ise, instanceOf(IllegalMonitorStateException.class)); + assertThat(((Exception) ise).getMessage(), containsString("You do not own")); + } + + @Test + public void testTwoThreads() throws Exception { + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + final Lock lock1 = registry.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + } + + @Test + public void testTwoThreadsDifferentRegistries() throws Exception { + final ZookeeperLockRegistry registry1 = new ZookeeperLockRegistry(this.client); + final ZookeeperLockRegistry registry2 = new ZookeeperLockRegistry(this.client); + final Lock lock1 = registry1.obtain("foo"); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch1 = new CountDownLatch(1); + final CountDownLatch latch2 = new CountDownLatch(1); + final CountDownLatch latch3 = new CountDownLatch(1); + lock1.lockInterruptibly(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + + @Override + public void run() { + Lock lock2 = registry2.obtain("foo"); + try { + latch1.countDown(); + lock2.lockInterruptibly(); + latch2.await(10, TimeUnit.SECONDS); + locked.set(true); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + finally { + lock2.unlock(); + latch3.countDown(); + } + } + }); + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock1.unlock(); + latch2.countDown(); + assertTrue(latch3.await(10, TimeUnit.SECONDS)); + assertTrue(locked.get()); + } + + @Test + public void testTwoThreadsWrongOneUnlocks() throws Exception { + final ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client); + final Lock lock = registry.obtain("foo"); + lock.lockInterruptibly(); + final AtomicBoolean locked = new AtomicBoolean(); + final CountDownLatch latch = new CountDownLatch(1); + Future result = Executors.newSingleThreadExecutor().submit(new Callable() { + + @Override + public Object call() throws Exception { + try { + lock.unlock(); + } + catch (Exception e) { + latch.countDown(); + return e.getCause(); + } + return null; + } + }); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertFalse(locked.get()); + lock.unlock(); + Object imse = result.get(10, TimeUnit.SECONDS); + assertThat(imse, instanceOf(IllegalMonitorStateException.class)); + assertThat(((Exception) imse).getMessage(), containsString("You do not own")); + } + + @Test + public void testLockWithBoundedStrategy() throws Exception { + ZookeeperLockRegistry registry = new ZookeeperLockRegistry(this.client, new KeyToPathStrategy() { + + @Override + public String pathFor(String key) { + return "/SpringIntegration-LockRegistry/singleLock"; + } + + @Override + public boolean bounded() { + return true; + } + }); + for (int i = 0; i < 10; i++) { + Lock lock = registry.obtain("foo"); + lock.lock(); + try { + assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + } + finally { + lock.unlock(); + } + } + + Thread.sleep(10); + try { + registry.expireUnusedOlderThan(0); + fail("expected exception"); + } + catch (IllegalStateException e) { + assertThat(e.getMessage(), containsString("expiry is not supported")); + } + assertEquals(1, TestUtils.getPropertyValue(registry, "locks", Map.class).size()); + } + +} diff --git a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java index b0922f0143..55526006aa 100644 --- a/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java @@ -25,7 +25,6 @@ import static org.junit.Assert.fail; import static org.springframework.integration.test.matcher.EqualsResultMatcher.equalsResult; import static org.springframework.integration.test.matcher.EventuallyMatcher.eventually; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -34,71 +33,44 @@ import java.util.Map; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.apache.curator.framework.CuratorFramework; -import org.apache.curator.framework.CuratorFrameworkFactory; -import org.apache.curator.retry.BoundedExponentialBackoffRetry; -import org.apache.curator.test.TestingServer; import org.apache.curator.utils.CloseableUtils; import org.hamcrest.collection.IsIterableContainingInOrder; import org.junit.After; -import org.junit.AfterClass; import org.junit.Before; -import org.junit.BeforeClass; import org.junit.Test; import org.mockito.Mockito; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.integration.metadata.MetadataStoreListenerAdapter; import org.springframework.integration.metadata.MetadataStoreListener; +import org.springframework.integration.metadata.MetadataStoreListenerAdapter; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.integration.test.matcher.EqualsResultMatcher.Evaluator; +import org.springframework.integration.zookeeper.ZookeeperTestSupport; /** * @author Marius Bogoevici + * @since 4.2 */ -public class ZookeeperMetadataStoreTests { - - private static final Log log = LogFactory.getLog(ZookeeperMetadataStore.class); - - private static TestingServer testingServer; - - private CuratorFramework client; +public class ZookeeperMetadataStoreTests extends ZookeeperTestSupport { private ZookeeperMetadataStore metadataStore; - @BeforeClass - public static void setUpClass() throws Exception { - testingServer = new TestingServer(true); - } - - @AfterClass - public static void tearDownClass() throws Exception { - try { - testingServer.stop(); - } - catch (IOException e) { - log.warn("Exception thrown while shutting down ZooKeeper: ", e); - } - testingServer.getTempDirectory().delete(); - } - + @Override @Before - public void setUp() throws Exception{ - client = createNewClient(); - metadataStore = new ZookeeperMetadataStore(client); - metadataStore.start(); + public void setUp() throws Exception { + super.setUp(); + this.metadataStore = new ZookeeperMetadataStore(client); + this.metadataStore.start(); } + @Override @After public void tearDown() throws Exception { this.metadataStore.stop(); this.client.delete().deletingChildrenIfNeeded().forPath(this.metadataStore.getRoot()); - CloseableUtils.closeQuietly(this.client); } - @Test public void testGetNonExistingKeyValue() { String retrievedValue = metadataStore.get("does-not-exist"); @@ -413,12 +385,4 @@ public class ZookeeperMetadataStoreTests { } } - private CuratorFramework createNewClient() throws InterruptedException { - CuratorFramework client = CuratorFrameworkFactory.newClient(testingServer.getConnectString(), - new BoundedExponentialBackoffRetry(100, 1000, 3)); - client.start(); - client.blockUntilConnected(10000, TimeUnit.SECONDS); - return client; - } - }