From 8782d097cf7465efdb69c204ef2f0142b7b1a7cf Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Tue, 16 Jun 2015 17:14:20 -0400 Subject: [PATCH] INT-3691 Add ZookeeperMetadataStore JIRA: https://jira.spring.io/browse/INT-3691 - create new core interfaces `ListenableMetadataStore` and `MessageStoreListener`; - create new module `spring-integration-zookeeper`; - add `ZookeeperMetadataStore` implementation Polishing Polishing - moved classes into core/test where necessary, including conversion methods - properly renamed `MetadataStoreListenerAdapter` --- build.gradle | 18 + .../metadata/ListenableMetadataStore.java | 41 ++ .../metadata/MetadataStoreListener.java | 48 ++ .../MetadataStoreListenerAdapter.java | 40 ++ .../support/utils/IntegrationUtils.java | 39 +- .../test/matcher/EqualsResultMatcher.java | 57 +++ .../test/matcher/EventuallyMatcher.java | 85 ++++ .../metadata/ZookeeperMetadataStore.java | 390 ++++++++++++++++ .../ZookeeperMetadataStoreException.java | 33 ++ .../zookeeper/metadata/package-info.java | 5 + .../metadata/ZookeeperMetadataStoreTests.java | 422 ++++++++++++++++++ .../src/test/resources/log4j.properties | 8 + 12 files changed, 1185 insertions(+), 1 deletion(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListenerAdapter.java create mode 100644 spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java create mode 100644 spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java create mode 100644 spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java create mode 100644 spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java create mode 100644 spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/package-info.java create mode 100644 spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java create mode 100644 spring-integration-zookeeper/src/test/resources/log4j.properties diff --git a/build.gradle b/build.gradle index 08fc41d0d4..2e85dcde8c 100644 --- a/build.gradle +++ b/build.gradle @@ -86,6 +86,7 @@ subprojects { subproject -> commonsDbcpVersion = '1.4' commonsIoVersion = '2.4' commonsNetVersion = '3.3' + curatorVersion = '2.8.0' derbyVersion = '10.11.1.1' eclipseLinkVersion = '2.4.2' ftpServerVersion = '1.0.6' @@ -135,6 +136,7 @@ subprojects { subproject -> springWsVersion = '2.2.1.RELEASE' xmlUnitVersion = '1.5' xstreamVersion = '1.4.7' + zookeeperVersion = '3.4.6' } eclipse { @@ -681,6 +683,22 @@ project('spring-integration-xmpp') { } } +project('spring-integration-zookeeper') { + description = 'Spring Integration Zookeeper Support' + dependencies { + compile project(":spring-integration-core") + compile ("org.apache.zookeeper:zookeeper:$zookeeperVersion") + + compile("org.apache.curator:curator-recipes:$curatorVersion") { + exclude group: 'org.apache.zookeeper', module: 'zookeeper' + exclude group: 'org.jboss.netty', module: 'netty' + } + + testCompile "org.springframework.integration:spring-integration-test" + testCompile "org.apache.curator:curator-test:$curatorVersion" + } +} + project("spring-integration-bom") { description = "Spring Integration (Bill of Materials)" 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 new file mode 100644 index 0000000000..ad0e7e594c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/ListenableMetadataStore.java @@ -0,0 +1,41 @@ +/* + * Copyright 2002-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.metadata; + +/** + * + * {@link ConcurrentMetadataStore} with the ability of registering {@link MetadataStoreListener} callbacks, to be + * invoked when changes occur in the metadata store. + * + * @author Marius Bogoevici + */ +public interface ListenableMetadataStore extends ConcurrentMetadataStore { + + /** + * Registers a listener with the metadata store + * + * @param callback the callback to be registered + */ + void addListener(MetadataStoreListener callback); + + /** + * Unregisters a listener + * + * @param callback the callback to be unregistered + */ + void removeListener(MetadataStoreListener callback); +} 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 new file mode 100644 index 0000000000..ec53c1dd87 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListener.java @@ -0,0 +1,48 @@ +/* + * 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.metadata; + +/** + * A callback to be invoked whenever a value changes in the data store. + * + * @author Marius Bogoevici + */ +public interface MetadataStoreListener { + + /** + * Invoked when a key is added to the store + * + * @param key the key being added + * @param value the value being added + */ + void onAdd(String key, String value); + + /** + * Invoked when a key is removed from the store + * + * @param key the key being removed + * @param oldValue the value being removed + */ + void onRemove(String key, String oldValue); + + /** + * Invoked when a key is updated into the store + * @param key the key being updated + * @param newValue the new value + */ + void onUpdate(String key, String newValue); +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListenerAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListenerAdapter.java new file mode 100644 index 0000000000..0b659f5814 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/metadata/MetadataStoreListenerAdapter.java @@ -0,0 +1,40 @@ +/* + * 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.metadata; + +/** + * Base implementation for a {@link MetadataStoreListener}. Subclasses may override any of the methods. + * + * @author Marius Bogoevici + */ +public abstract class MetadataStoreListenerAdapter implements MetadataStoreListener { + + @Override + public void onAdd(String key, String value) { + + } + + @Override + public void onRemove(String key, String oldValue) { + + } + + @Override + public void onUpdate(String key, String newValue) { + + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java index f349e33133..3a962bccff 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/utils/IntegrationUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-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. @@ -15,6 +15,8 @@ */ package org.springframework.integration.support.utils; +import java.io.UnsupportedEncodingException; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -23,11 +25,13 @@ import org.springframework.core.convert.ConversionService; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * General utility methods. * * @author Gary Russell + * @author Marius Bogoevici * @since 4.0 * */ @@ -98,4 +102,37 @@ public class IntegrationUtils { return beanFactory.getBean(beanName, type); } + /** + * Utility method for null-safe conversion from String to byte[] + * + * @param value the String to be converted + * @param encoding the encoding + * @return the byte[] corresponding to the given String and encoding, null if provided String argument was null + * @throws IllegalArgumentException if the encoding is not supported + */ + public static byte[] stringToBytes(String value, String encoding) { + try { + return value != null ? value.getBytes(encoding) : null; + } + catch (UnsupportedEncodingException e) { + throw new IllegalArgumentException(e); + } + } + + /** + * Utility method for null-safe conversion from byte[] to String + * + * @param bytes the byte[] to be converted + * @param encoding the encoding + * @return the String corresponding to the given byte[] and encoding, null if provided byte[] argument was null + * @throws IllegalArgumentException if the encoding is not supported + */ + public static String bytesToString(byte[] bytes, String encoding) { + try { + return bytes == null ? null : new String(bytes, encoding); + } + catch (UnsupportedEncodingException e) { + throw new IllegalArgumentException(e); + } + } } 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 new file mode 100644 index 0000000000..43c4c7dfbc --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EqualsResultMatcher.java @@ -0,0 +1,57 @@ +/* + * 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.test.matcher; + +import org.hamcrest.Description; +import org.hamcrest.DiagnosingMatcher; + +import org.springframework.util.ObjectUtils; + +/** + * A matcher that evaluates against the result of invoking a function, + * wrapped by the {@link EqualsResultMatcher.Evaluator} + * + * The goal is to defer the computation until the matcher needs to be actually evaluated. + * Mainly useful in conjunction with retrying matcherss such as {@link EventuallyMatcher} + * + * @author Marius Bogoevici + */ +public class EqualsResultMatcher extends DiagnosingMatcher { + + private Evaluator evaluator; + + public EqualsResultMatcher(Evaluator evaluator) { + this.evaluator = evaluator; + } + + @Override + protected boolean matches(Object item, Description mismatchDescription) { + return ObjectUtils.nullSafeEquals(item, evaluator.evaluate()); + } + + @Override + public void describeTo(Description description) { + } + + public interface Evaluator { + U evaluate(); + } + + public static EqualsResultMatcher equalsResult(Evaluator evaluator) { + return new EqualsResultMatcher(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 new file mode 100644 index 0000000000..80eca5c0d5 --- /dev/null +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/matcher/EventuallyMatcher.java @@ -0,0 +1,85 @@ +/* + * Copyright 2013-2104 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.test.matcher; + +import org.hamcrest.Description; +import org.hamcrest.DiagnosingMatcher; +import org.hamcrest.Matcher; + + +/** + * A matcher that will evaluate another matcher repeatedly until it matches, or fail after some number of attempts. + * + * @param the type the wrapped matcher operates on + * + * (Copied from {@code org.springframework.xd.test.fixtures.EventuallyMatcher}) + * + * @author Eric Bottard + */ +public class EventuallyMatcher extends DiagnosingMatcher { + + private final Matcher delegate; + + private int nbAttempts; + + private int pause; + + public EventuallyMatcher(Matcher delegate) { + this(delegate, 20, 100); + } + + public EventuallyMatcher(Matcher delegate, int nbAttempts, int pause) { + this.delegate = delegate; + this.nbAttempts = nbAttempts; + this.pause = pause; + } + + public static Matcher eventually(int nbAttempts, int pause, Matcher delegate) { + return new EventuallyMatcher(delegate,nbAttempts,pause); + } + + public static Matcher eventually(Matcher delegate) { + return new EventuallyMatcher(delegate); + } + + @Override + public void describeTo(Description description) { + description.appendDescriptionOf(delegate).appendText(String.format(", trying at most %d times", nbAttempts)); + } + + @Override + protected boolean matches(Object item, Description mismatchDescription) { + mismatchDescription.appendText(String.format("failed after %d*%d=%dms:%n", nbAttempts, pause, nbAttempts + * pause)); + for (int i = 0; i < nbAttempts; i++) { + boolean result = delegate.matches(item); + if (result) { + return true; + } + delegate.describeMismatch(item, mismatchDescription); + mismatchDescription.appendText(", "); + try { + Thread.sleep(pause); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + return false; + } +} 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 new file mode 100644 index 0000000000..cd7c585b26 --- /dev/null +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStore.java @@ -0,0 +1,390 @@ +/* + * 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.metadata; + +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.cache.ChildData; +import org.apache.curator.framework.recipes.cache.PathChildrenCache; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent; +import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener; +import org.apache.curator.utils.CloseableUtils; +import org.apache.curator.utils.EnsurePath; +import org.apache.zookeeper.KeeperException; +import org.apache.zookeeper.data.Stat; + +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.metadata.ListenableMetadataStore; +import org.springframework.integration.metadata.MetadataStoreListener; +import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.util.Assert; + +/** + * Zookeeper-based {@link ListenableMetadataStore} based on a Zookeeper node. Values are stored in the children node, + * the names of which are stored as keys. + * + * @author Marius Bogoevici + */ +public class ZookeeperMetadataStore implements ListenableMetadataStore, SmartLifecycle { + + private final Object lifecycleMonitor = new Object(); + + private final CuratorFramework client; + + private final List listeners = new CopyOnWriteArrayList(); + + /** + * An internal map storing local updates, ensuring that they have precedence if the cache contains stale data. + * As changes are propagated back from Zookeeper to the cache, entries are removed. + */ + private final ConcurrentMap updateMap = new ConcurrentHashMap(); + + private volatile String root = "/SpringIntegration-MetadataStore"; + + private volatile String encoding = "UTF-8"; + + private volatile PathChildrenCache cache; + + private volatile boolean running = false; + + private volatile boolean autoStartup = true; + + private volatile int phase = Integer.MAX_VALUE; + + public ZookeeperMetadataStore(CuratorFramework client) throws Exception { + Assert.notNull(client, "Client cannot be null"); + this.client = client; + } + + /** + * Encoding to use when storing data in ZooKeeper + * + * @param encoding encoding as text + */ + public void setEncoding(String encoding) { + Assert.hasText(encoding, "'encoding' cannot be null or empty."); + this.encoding = encoding; + } + + /** + * Root node - store entries are children of this node. + * + * @param root encoding as text + */ + public void setRoot(String root) { + Assert.notNull(root, "'root' must not be null."); + Assert.isTrue(root.startsWith("/"), "'root' must start with '/'"); + // remove trailing slash, if not root + this.root = "/".equals(root) || !root.endsWith("/") ? root : root.substring(0, root.length() - 1); + } + + public String getRoot() { + return this.root; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + @Override + public String putIfAbsent(String key, String value) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(value, "'value' must not be null."); + synchronized (this.updateMap) { + try { + createNode(key, value); + return null; + } + catch (KeeperException.NodeExistsException e) { + // so the data actually exists, we can read it + try { + byte[] bytes = this.client.getData().forPath(getPath(key)); + return IntegrationUtils.bytesToString(bytes, encoding); + } + catch (Exception exceptionDuringGet) { + throw new ZookeeperMetadataStoreException("Exception while reading node with key '" + key + "':", e); + } + } + catch (Exception e) { + throw new ZookeeperMetadataStoreException("Error while trying to set '" + key + "':", e); + } + } + } + + @Override + public boolean replace(String key, String oldValue, String newValue) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(oldValue, "'oldValue' must not be null."); + Assert.notNull(newValue, "'newValue' must not be null."); + synchronized (this.updateMap) { + Stat currentStat = new Stat(); + try { + byte[] bytes = this.client.getData().storingStatIn(currentStat).forPath(getPath(key)); + if (oldValue.equals(IntegrationUtils.bytesToString(bytes, encoding))) { + updateNode(key, newValue, currentStat.getVersion()); + } + return true; + } + catch (KeeperException.NoNodeException e) { + // ignore, the node doesn't exist there's nothing to replace + return false; + } + catch (KeeperException.BadVersionException e) { + // ignore + return false; + } + catch (Exception e) { + throw new ZookeeperMetadataStoreException("Cannot replace value"); + } + } + } + + @Override + public void addListener(MetadataStoreListener callback) { + this.listeners.add(callback); + } + + @Override + public void removeListener(MetadataStoreListener callback) { + this.listeners.remove(callback); + } + + @Override + public void put(String key, String value) { + Assert.notNull(key, "'key' must not be null."); + Assert.notNull(value, "'value' must not be null."); + synchronized (this.updateMap) { + try { + Stat currentNode = this.client.checkExists().forPath(getPath(key)); + if (currentNode == null) { + try { + createNode(key, value); + } + catch (KeeperException.NodeExistsException e) { + updateNode(key, value, -1); + } + } + else { + updateNode(key, value, -1); + } + } + catch (Exception e) { + throw new ZookeeperMetadataStoreException("Error while setting value for key '" + key + "':", e); + } + } + } + + @Override + public String get(String key) { + Assert.notNull(key, "'key' must not be null."); + synchronized (this.updateMap) { + ChildData currentData = this.cache.getCurrentData(getPath(key)); + if (currentData == null) { + if (this.updateMap.containsKey(key)) { + // we have saved the value, but the cache hasn't updated yet + // if the value had changed via replication, we would have been notified by the listener + return this.updateMap.get(key).getValue(); + } + else { + // the value just doesn't exist + return null; + } + } + else { + if (this.updateMap.containsKey(key)) { + // our version is more recent than the cache + if (this.updateMap.get(key).getVersion() >= currentData.getStat().getVersion()) { + return this.updateMap.get(key).getValue(); + } + } + return IntegrationUtils.bytesToString(currentData.getData(), encoding); + } + } + } + + @Override + public String remove(String key) { + Assert.notNull(key, "'key' must not be null."); + synchronized (this.updateMap) { + try { + byte[] bytes = this.client.getData().forPath(getPath(key)); + this.client.delete().forPath(getPath(key)); + // we guarantee that the deletion will supersede the existing data + this.updateMap.put(key, new LocalChildData(null, Integer.MAX_VALUE)); + return IntegrationUtils.bytesToString(bytes, encoding); + } + catch (KeeperException.NoNodeException e) { + // ignore - the node doesn't exist + return null; + } + catch (Exception e) { + throw new ZookeeperMetadataStoreException("Exception while deleting key '" + key + "'", e); + } + } + } + + private void updateNode(String key, String value, int version) throws Exception { + Stat stat = this.client.setData().withVersion(version).forPath(getPath(key), + IntegrationUtils.stringToBytes(value, this.encoding)); + this.updateMap.put(key, new LocalChildData(value, stat.getVersion())); + } + + private void createNode(String key, String value) throws Exception { + this.client.create().forPath(getPath(key), IntegrationUtils.stringToBytes(value, this.encoding)); + this.updateMap.put(key, new LocalChildData(value, 0)); + } + + public String getPath(String key) { + return "".equals(key) ? this.root : this.root + "/" + key; + } + + + @Override + public boolean isAutoStartup() { + return autoStartup; + } + + @Override + public void start() { + if (!this.running) { + synchronized (this.lifecycleMonitor) { + if (!this.running) { + try { + EnsurePath ensurePath = new EnsurePath(root); + ensurePath.ensure(client.getZookeeperClient()); + this.cache = new PathChildrenCache(this.client, this.root, true); + this.cache.getListenable().addListener(new MetadataStoreListenerInvokingPathChildrenCacheListener()); + this.cache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE); + running = true; + } + catch (Exception e) { + throw new ZookeeperMetadataStoreException("Exception while starting bean", e); + } + } + } + } + } + + @Override + public void stop() { + if (this.running) { + synchronized (this.lifecycleMonitor) { + if (this.running) { + if (this.cache != null) { + CloseableUtils.closeQuietly(this.cache); + } + this.cache = null; + this.running = false; + } + } + } + } + + @Override + public void stop(Runnable callback) { + stop(); + callback.run(); + } + + @Override + public boolean isRunning() { + return running; + } + + @Override + public int getPhase() { + return phase; + } + + private static class LocalChildData { + + private String value; + + private int version; + + public LocalChildData(String value, int version) { + this.value = value; + this.version = version; + } + + public String getValue() { + return this.value; + } + + public int getVersion() { + return this.version; + } + } + + private class MetadataStoreListenerInvokingPathChildrenCacheListener implements + PathChildrenCacheListener { + @Override + public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) + throws Exception { + synchronized (ZookeeperMetadataStore.this.updateMap) { + String eventPath = event.getData().getPath(); + String eventKey = getKey(eventPath); + byte[] eventData = event.getData().getData(); + switch (event.getType()) { + case CHILD_ADDED: + if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) { + if (event.getData().getStat().getVersion() >= + ZookeeperMetadataStore.this.updateMap.get(eventKey).getVersion()) { + ZookeeperMetadataStore.this.updateMap.remove(eventPath); + } + } + for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) { + listener.onAdd(eventKey, IntegrationUtils.bytesToString(eventData, encoding)); + } + break; + case CHILD_UPDATED: + if (ZookeeperMetadataStore.this.updateMap.containsKey(eventKey)) { + if (event.getData().getStat().getVersion() >= + ZookeeperMetadataStore.this.updateMap.get(eventKey).getVersion()) { + ZookeeperMetadataStore.this.updateMap.remove(eventPath); + } + } + for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) { + listener.onUpdate(eventKey, IntegrationUtils.bytesToString(eventData, encoding)); + } + break; + case CHILD_REMOVED: + ZookeeperMetadataStore.this.updateMap.remove(eventKey); + for (MetadataStoreListener listener : ZookeeperMetadataStore.this.listeners) { + listener.onRemove(eventKey, IntegrationUtils.bytesToString(eventData, encoding)); + } + break; + default: + // ignore all other events + break; + } + } + } + } + + private String getKey(String path) { + return path.replace(root + "/", ""); + } +} 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 new file mode 100644 index 0000000000..9d2be5d3f3 --- /dev/null +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreException.java @@ -0,0 +1,33 @@ +/* + * 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.metadata; + +/** + * @author Marius Bogoevici + */ +@SuppressWarnings("serial") +public class ZookeeperMetadataStoreException extends RuntimeException { + + public ZookeeperMetadataStoreException(String message) { + super(message); + } + + public ZookeeperMetadataStoreException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/package-info.java b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/package-info.java new file mode 100644 index 0000000000..d6f79ba203 --- /dev/null +++ b/spring-integration-zookeeper/src/main/java/org/springframework/integration/zookeeper/metadata/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides classes supporting the Zookeeper-based + * {@link org.springframework.integration.metadata.ListenableMetadataStore} + */ +package org.springframework.integration.zookeeper.metadata; 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 new file mode 100644 index 0000000000..cca427ac6b --- /dev/null +++ b/spring-integration-zookeeper/src/test/java/org/springframework/integration/zookeeper/metadata/ZookeeperMetadataStoreTests.java @@ -0,0 +1,422 @@ +/* + * 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.metadata; + +import static org.hamcrest.collection.IsCollectionWithSize.hasSize; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +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; +import java.util.List; +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.Before; +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.support.utils.IntegrationUtils; +import org.springframework.integration.test.matcher.EqualsResultMatcher.Evaluator; + +/** + * @author Marius Bogoevici + */ +public class ZookeeperMetadataStoreTests { + + private static final Log log = LogFactory.getLog(ZookeeperMetadataStore.class); + + private TestingServer testingServer; + + private CuratorFramework client; + + private ZookeeperMetadataStore metadataStore; + + @Before + public void setUp() throws Exception{ + testingServer = new TestingServer(true); + client = createNewClient(); + metadataStore = new ZookeeperMetadataStore(client); + metadataStore.start(); + } + + @After + public void tearDown() { + this.metadataStore.stop(); + closeClient(this.client); + try { + testingServer.stop(); + } + catch (IOException e) { + log.warn("Exception thrown while shutting down ZooKeeper: ", e); + } + testingServer.getTempDirectory().delete(); + } + + + @Test + public void testGetNonExistingKeyValue() { + String retrievedValue = metadataStore.get("does-not-exist"); + assertNull(retrievedValue); + } + + @Test + public void testPersistKeyValue() throws Exception { + String testKey = "ZookeeperMetadataStoreTests-Persist"; + metadataStore.put(testKey, "Integration"); + assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey))); + assertEquals("Integration", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + } + + + @Test + public void testGetValueFromMetadataStore() throws Exception { + String testKey = "ZookeeperMetadataStoreTests-GetValue"; + metadataStore.put(testKey, "Hello Zookeeper"); + String retrievedValue = metadataStore.get(testKey); + assertEquals("Hello Zookeeper", retrievedValue); + } + + + @Test + public void testPutIfAbsent() throws Exception { + final String testKey = "ZookeeperMetadataStoreTests-Persist"; + final String testKey2 = "ZookeeperMetadataStoreTests-Persist-2"; + metadataStore.put(testKey, "Integration"); + assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey))); + assertEquals("Integration", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + CuratorFramework otherClient = createNewClient(); + final ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient); + otherMetadataStore.start(); + otherMetadataStore.putIfAbsent(testKey, "OtherValue"); + assertEquals("Integration", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + assertEquals("Integration", metadataStore.get(testKey)); + assertThat("Integration", eventually(equalsResult(new Evaluator() { + @Override + public String evaluate() { + return otherMetadataStore.get(testKey); + } + }))); + otherMetadataStore.putIfAbsent(testKey2, "Integration-2"); + assertEquals("Integration-2", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey2)), "UTF-8")); + assertEquals("Integration-2", otherMetadataStore.get(testKey2)); + assertThat("Integration-2", eventually(equalsResult(new Evaluator() { + @Override + public String evaluate() { + return metadataStore.get(testKey2); + } + }))); + closeClient(otherClient); + } + + @Test + public void testReplace() throws Exception { + final String testKey = "ZookeeperMetadataStoreTests-Replace"; + metadataStore.put(testKey, "Integration"); + assertNotNull(client.checkExists().forPath(metadataStore.getPath(testKey))); + assertEquals("Integration", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + CuratorFramework otherClient = createNewClient(); + final ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient); + otherMetadataStore.start(); + otherMetadataStore.replace(testKey, "OtherValue", "Integration-2"); + assertEquals("Integration", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + assertEquals("Integration", metadataStore.get(testKey)); + assertThat("Integration", eventually(equalsResult(new Evaluator() { + @Override + public String evaluate() { + return otherMetadataStore.get(testKey); + } + }))); + otherMetadataStore.replace(testKey, "Integration", "Integration-2"); + assertEquals("Integration-2", + IntegrationUtils.bytesToString(client.getData().forPath(metadataStore.getPath(testKey)), "UTF-8")); + assertThat("Integration-2", eventually(equalsResult(new Evaluator() { + @Override + public String evaluate() { + return metadataStore.get(testKey); + } + }))); + assertEquals("Integration-2", otherMetadataStore.get(testKey)); + closeClient(otherClient); + } + + @Test + public void testPersistEmptyStringToMetadataStore() { + String testKey = "ZookeeperMetadataStoreTests-PersistEmpty"; + metadataStore.put(testKey, ""); + assertEquals("", metadataStore.get(testKey)); + } + + @Test + public void testPersistNullStringToMetadataStore() { + try { + metadataStore.put("ZookeeperMetadataStoreTests-PersistEmpty", null); + } + catch (IllegalArgumentException e) { + assertEquals("'value' must not be null.", e.getMessage()); + return; + } + fail("Expected an IllegalArgumentException to be thrown."); + } + + @Test + public void testPersistWithEmptyKeyToMetadataStore() { + metadataStore.put("", "PersistWithEmptyKey"); + String retrievedValue = metadataStore.get(""); + assertEquals("PersistWithEmptyKey", retrievedValue); + } + + @Test + public void testPersistWithNullKeyToMetadataStore() { + try { + metadataStore.put(null, "something"); + } + catch (IllegalArgumentException e) { + assertEquals("'key' must not be null.", e.getMessage()); + return; + } + fail("Expected an IllegalArgumentException to be thrown."); + } + + @Test + public void testGetValueWithNullKeyFromMetadataStore() { + try { + metadataStore.get(null); + } + catch (IllegalArgumentException e) { + assertEquals("'key' must not be null.", e.getMessage()); + return; + } + fail("Expected an IllegalArgumentException to be thrown."); + } + + @Test + public void testRemoveFromMetadataStore() throws Exception { + String testKey = "ZookeeperMetadataStoreTests-Remove"; + String testValue = "Integration"; + metadataStore.put(testKey, testValue); + assertEquals(testValue, metadataStore.remove(testKey)); + Thread.sleep(1000); + assertNull(metadataStore.remove(testKey)); + } + + @Test + public void testListenerInvokedOnLocalChanges() throws Exception { + String testKey = "ZookeeperMetadataStoreTests"; + + // register listeners + final List> notifiedChanges = new ArrayList>(); + final Map barriers = new HashMap(); + barriers.put("add", new CyclicBarrier(2)); + barriers.put("remove", new CyclicBarrier(2)); + barriers.put("update", new CyclicBarrier(2)); + metadataStore.addListener(new MetadataStoreListenerAdapter() { + @Override + public void onAdd(String key, String value) { + notifiedChanges.add(Arrays.asList("add", key, value)); + waitAtBarrier("add", barriers); + } + + @Override + public void onRemove(String key, String oldValue) { + notifiedChanges.add(Arrays.asList("remove", key, oldValue)); + waitAtBarrier("remove", barriers); + } + + @Override + public void onUpdate(String key, String newValue) { + notifiedChanges.add(Arrays.asList("update", key, newValue)); + waitAtBarrier("update", barriers); + } + }); + + // the tests themselves + barriers.get("add").reset(); + metadataStore.put(testKey, "Integration"); + waitAtBarrier("add", barriers); + assertThat(notifiedChanges, hasSize(1)); + assertThat(notifiedChanges.get(0), IsIterableContainingInOrder.contains("add", testKey, "Integration")); + + metadataStore.putIfAbsent(testKey, "Integration++"); + // there is no update and therefore we expect no changes + assertThat(notifiedChanges, hasSize(1)); + + barriers.get("update").reset(); + metadataStore.put(testKey, "Integration-2"); + waitAtBarrier("update", barriers); + assertThat(notifiedChanges, hasSize(2)); + assertThat(notifiedChanges.get(1), IsIterableContainingInOrder.contains("update", testKey, "Integration-2")); + + barriers.get("update").reset(); + metadataStore.replace(testKey, "Integration-2", "Integration-3"); + waitAtBarrier("update", barriers); + assertThat(notifiedChanges, hasSize(3)); + assertThat(notifiedChanges.get(2), IsIterableContainingInOrder.contains("update", testKey, "Integration-3")); + + metadataStore.replace(testKey, "Integration-2", "Integration-none"); + assertThat(notifiedChanges, hasSize(3)); + + barriers.get("remove").reset(); + metadataStore.remove(testKey); + waitAtBarrier("remove", barriers); + assertThat(notifiedChanges, hasSize(4)); + assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3")); + + // sleep and try to see if there were any other updates + Thread.sleep(1000); + assertThat(notifiedChanges, hasSize(4)); + } + + @Test + public void testListenerInvokedOnRemoteChanges() throws Exception { + String testKey = "ZookeeperMetadataStoreTests"; + + CuratorFramework otherClient = createNewClient(); + ZookeeperMetadataStore otherMetadataStore = new ZookeeperMetadataStore(otherClient); + + // register listeners + final List> notifiedChanges = new ArrayList>(); + final Map barriers = new HashMap(); + barriers.put("add", new CyclicBarrier(2)); + barriers.put("remove", new CyclicBarrier(2)); + barriers.put("update", new CyclicBarrier(2)); + metadataStore.addListener(new MetadataStoreListenerAdapter() { + @Override + public void onAdd(String key, String value) { + notifiedChanges.add(Arrays.asList("add", key, value)); + waitAtBarrier("add", barriers); + } + + @Override + public void onRemove(String key, String oldValue) { + notifiedChanges.add(Arrays.asList("remove", key, oldValue)); + waitAtBarrier("remove", barriers); + } + + @Override + public void onUpdate(String key, String newValue) { + notifiedChanges.add(Arrays.asList("update", key, newValue)); + waitAtBarrier("update", barriers); + } + }); + + // the tests themselves + barriers.get("add").reset(); + otherMetadataStore.put(testKey, "Integration"); + waitAtBarrier("add", barriers); + assertThat(notifiedChanges, hasSize(1)); + assertThat(notifiedChanges.get(0), IsIterableContainingInOrder.contains("add", testKey, "Integration")); + + otherMetadataStore.putIfAbsent(testKey, "Integration++"); + // there is no update and therefore we expect no changes + assertThat(notifiedChanges, hasSize(1)); + + barriers.get("update").reset(); + otherMetadataStore.put(testKey, "Integration-2"); + waitAtBarrier("update", barriers); + assertThat(notifiedChanges, hasSize(2)); + assertThat(notifiedChanges.get(1), IsIterableContainingInOrder.contains("update", testKey, "Integration-2")); + + barriers.get("update").reset(); + otherMetadataStore.replace(testKey, "Integration-2", "Integration-3"); + waitAtBarrier("update", barriers); + assertThat(notifiedChanges, hasSize(3)); + assertThat(notifiedChanges.get(2), IsIterableContainingInOrder.contains("update", testKey, "Integration-3")); + + otherMetadataStore.replace(testKey, "Integration-2", "Integration-none"); + assertThat(notifiedChanges, hasSize(3)); + + barriers.get("remove").reset(); + otherMetadataStore.remove(testKey); + waitAtBarrier("remove", barriers); + assertThat(notifiedChanges, hasSize(4)); + assertThat(notifiedChanges.get(3), IsIterableContainingInOrder.contains("remove", testKey, "Integration-3")); + + // sleep and try to see if there were any other updates - if there any pending updates, we should catch them by now + Thread.sleep(1000); + assertThat(notifiedChanges, hasSize(4)); + } + + @Test + public void testAddRemoveListener() throws Exception { + MetadataStoreListener mockListener = Mockito.mock(MetadataStoreListener.class); + DirectFieldAccessor accessor = new DirectFieldAccessor(metadataStore); + + @SuppressWarnings("unchecked") + List listeners = (List) accessor.getPropertyValue("listeners"); + + assertThat(listeners, hasSize(0)); + metadataStore.addListener(mockListener); + assertThat(listeners, hasSize(1)); + assertThat(listeners, IsIterableContainingInOrder.contains(mockListener)); + metadataStore.removeListener(mockListener); + assertThat(listeners, hasSize(0)); + } + + private void waitAtBarrier(String barrierName, Map barriers) { + try { + barriers.get(barrierName).await(10, TimeUnit.SECONDS); + } + catch (Exception e) { + throw new AssertionError("Test didn't complete: ", e); + } + } + + 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; + } + + private 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/resources/log4j.properties b/spring-integration-zookeeper/src/test/resources/log4j.properties new file mode 100644 index 0000000000..81c446c329 --- /dev/null +++ b/spring-integration-zookeeper/src/test/resources/log4j.properties @@ -0,0 +1,8 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n + +log4j.category.org.springframework.integration=WARN +log4j.category.org.springframework.integration.zookeeper=INFO