SpringCloud/spring-cloud-zookeeper
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Listen for Zookeeper change events to trigger refresh.

Browse Source 
fixes gh-49
This commit is contained in:
Spencer Gibb
2016-02-05 14:54:51 -07:00
parent e1c214cf58
commit 810fdad749
11 changed files with 375 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
pom.xml
View File

@@ -78,6 +78,34 @@
</includes>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<version>${checkstyle.version}</version>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-build-tools</artifactId>
<version>${spring-cloud-build.version}</version>
</dependency>
</dependencies>
<executions>
<execution>
<id>validate</id>
<phase>validate</phase>
<configuration>
<configLocation>checkstyle.xml</configLocation>
<headerLocation>LICENSE.txt</headerLocation>
<consoleOutput>true</consoleOutput>
<failsOnError>true</failsOnError>
</configuration>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</pluginManagement>
</build>
@@ -132,6 +160,8 @@
<properties>
<spock.version>1.0-groovy-2.4</spock.version>
<checkstyle.version>2.17</checkstyle.version>
<spring-cloud-build.version>1.1.0.BUILD-SNAPSHOT</spring-cloud-build.version>
</properties>
<profiles>

5
spring-cloud-zookeeper-config/pom.xml
View File

@@ -40,6 +40,11 @@
<artifactId>spring-boot-starter-web</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>

103
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ConfigWatcher.java Normal file
View File

@@ -0,0 +1,103 @@
/*
* Copyright 2013-2016 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.cloud.zookeeper.config;
import java.io.Closeable;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.annotation.PostConstruct;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.zookeeper.KeeperException;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import lombok.extern.apachecommons.CommonsLog;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_ADDED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_REMOVED;
import static org.apache.curator.framework.recipes.cache.TreeCacheEvent.Type.NODE_UPDATED;
/**
* @author Spencer Gibb
*/
@CommonsLog
public class ConfigWatcher implements Closeable, TreeCacheListener, ApplicationEventPublisherAware{
private AtomicBoolean running = new AtomicBoolean(false);
private List<String> contexts;
private CuratorFramework source;
private ApplicationEventPublisher publisher;
private HashMap<String, TreeCache> caches;
public ConfigWatcher(List<String> contexts, CuratorFramework source) {
this.contexts = contexts;
this.source = source;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
this.publisher = publisher;
}
@PostConstruct
public void start() {
if (this.running.compareAndSet(false, true)) {
this.caches = new HashMap<>();
for (String context : contexts) {
if (!context.startsWith("/")) {
context = "/" + context;
}
try {
TreeCache cache = TreeCache.newBuilder(this.source, context).build();
cache.getListenable().addListener(this);
cache.start();
this.caches.put(context, cache);
// no race condition since ZookeeperAutoConfiguration.curatorFramework
// calls curator.blockUntilConnected
} catch (KeeperException.NoNodeException e) {
// no node, ignore
} catch (Exception e) {
log.error("Error initializing listener for context " + context, e);
}
}
}
}
@Override
public void close() {
if (this.running.compareAndSet(true, false)) {
for (TreeCache cache : this.caches.values()) {
cache.close();
}
this.caches = null;
}
}
@Override
public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
TreeCacheEvent.Type eventType = event.getType();
if (eventType == NODE_ADDED || eventType == NODE_REMOVED || eventType == NODE_UPDATED) {
this.publisher.publishEvent(new ZookeeperConfigRefreshEvent(this, event));
}
}
}

50
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperConfigAutoConfiguration.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2013-2016 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.cloud.zookeeper.config;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.endpoint.RefreshEndpoint;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Spencer Gibb
*/
@Configuration
public class ZookeeperConfigAutoConfiguration {
@Configuration
@ConditionalOnClass(RefreshEndpoint.class)
protected static class ZkRefreshConfiguration {
@Bean
@ConditionalOnBean(RefreshEndpoint.class)
public ZookeeperConfigRefreshListener zookeeperConfigRefreshListener(
RefreshEndpoint refreshEndpoint) {
return new ZookeeperConfigRefreshListener(refreshEndpoint);
}
@Bean
@ConditionalOnProperty(name = "spring.cloud.zookeeper.config.watcher.enabled", matchIfMissing = true)
public ConfigWatcher configWatcher(ZookeeperPropertySourceLocator locator,
CuratorFramework curator) {
return new ConfigWatcher(locator.getContexts(), curator);
}
}
}

2
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperConfigProperties.java
View File

@@ -36,6 +36,4 @@ public class ZookeeperConfigProperties {
@NotEmpty
private String profileSeparator = ",";
private boolean cacheEnabled = true;
}

55
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperConfigRefreshEvent.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2013-2016 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.cloud.zookeeper.config;
import java.nio.charset.Charset;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.springframework.context.ApplicationEvent;
/**
* @author Spencer Gibb
*/
public class ZookeeperConfigRefreshEvent extends ApplicationEvent {
private TreeCacheEvent event;
/**
* Create a new ApplicationEvent.
*
* @param source the object on which the event initially occurred (never {@code null})
*/
public ZookeeperConfigRefreshEvent(Object source, TreeCacheEvent event) {
super(source);
this.event = event;
}
public TreeCacheEvent getEvent() {
return event;
}
public String getEventDesc() {
StringBuffer out = new StringBuffer();
out.append("type="+event.getType());
out.append(", path="+event.getData().getPath());
byte[] data = event.getData().getData();
if (data != null && data.length > 0) {
out.append(", data="+new String(data, Charset.forName("UTF-8")));
}
return out.toString();
}
}

55
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperConfigRefreshListener.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2013-2016 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.cloud.zookeeper.config;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicBoolean;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.cloud.endpoint.RefreshEndpoint;
import org.springframework.context.event.EventListener;
import lombok.extern.apachecommons.CommonsLog;
/**
* @author Spencer Gibb
*/
@CommonsLog
public class ZookeeperConfigRefreshListener {
private RefreshEndpoint refresh;
private AtomicBoolean ready = new AtomicBoolean(false);
public ZookeeperConfigRefreshListener(RefreshEndpoint refresh) {
this.refresh = refresh;
}
@EventListener
public void handle(ApplicationReadyEvent event) {
this.ready.compareAndSet(false, true);
}
@EventListener
public void handle(ZookeeperConfigRefreshEvent event) {
if (this.ready.get()) { // don't handle events before app is ready
log.debug("Event received " + event.getEventDesc());
if (this.refresh != null) {
String[] keys = this.refresh.refresh();
log.info("Refresh keys changed: " + Arrays.asList(keys));
}
}
}
}

18
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperPropertySourceLocator.java
View File

@@ -20,7 +20,6 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.PreDestroy;
@@ -40,13 +39,17 @@ public class ZookeeperPropertySourceLocator implements PropertySourceLocator {
private CuratorFramework curator;
private ConcurrentHashMap<String, ZookeeperTreeCachePropertySource> lifecycleSources = new ConcurrentHashMap<>();
private List<String> contexts;
public ZookeeperPropertySourceLocator(CuratorFramework curator, ZookeeperConfigProperties properties) {
this.curator = curator;
this.properties = properties;
}
public List<String> getContexts() {
return contexts;
}
@Override
public PropertySource<?> locate(Environment environment) {
if (environment instanceof ConfigurableEnvironment) {
@@ -55,7 +58,7 @@ public class ZookeeperPropertySourceLocator implements PropertySourceLocator {
List<String> profiles = Arrays.asList(env.getActiveProfiles());
String root = properties.getRoot();
List<String> contexts = new ArrayList<>();
contexts = new ArrayList<>();
String defaultContext = root + "/" + properties.getDefaultContext();
contexts.add(defaultContext);
@@ -82,18 +85,9 @@ public class ZookeeperPropertySourceLocator implements PropertySourceLocator {
@PreDestroy
public void destroy() {
for (ZookeeperTreeCachePropertySource source : this.lifecycleSources.values()) {
source.stop();
}
}
private PropertySource<CuratorFramework> create(String context) {
if (this.properties.isCacheEnabled()) {
ZookeeperTreeCachePropertySource propertySource = new ZookeeperTreeCachePropertySource(context, curator);
propertySource.start();
lifecycleSources.put(propertySource.getName(), propertySource);
return propertySource;
}
return new ZookeeperPropertySource(context, curator);
}

113
spring-cloud-zookeeper-config/src/main/java/org/springframework/cloud/zookeeper/config/ZookeeperTreeCachePropertySource.java
View File

@@ -1,113 +0,0 @@
/*
* Copyright 2013-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.cloud.zookeeper.config;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import lombok.extern.apachecommons.CommonsLog;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.springframework.context.Lifecycle;
/**
* @author Spencer Gibb
*/
@CommonsLog
public class ZookeeperTreeCachePropertySource extends AbstractZookeeperPropertySource
implements Lifecycle {
private TreeCache cache;
private boolean running;
public ZookeeperTreeCachePropertySource(String context, CuratorFramework source) {
super(context, source);
}
@Override
public void start() {
try {
cache = TreeCache.newBuilder(source, this.getContext()).build();
cache.start();
running = true;
// no race condition since ZookeeperAutoConfiguration.curatorFramework
// calls curator.blockUntilConnected
}
catch (NoNodeException e) {
// no node, ignore
}
catch (Exception e) {
log.error("Error initializing ZookeperPropertySource", e);
}
}
@Override
public Object getProperty(String name) {
String fullPath = this.getContext() + "/" + name.replace(".", "/");
byte[] bytes = null;
ChildData data = cache.getCurrentData(fullPath);
if (data != null) {
bytes = data.getData();
}
if (bytes == null)
return null;
return new String(bytes, Charset.forName("UTF-8"));
}
@Override
public String[] getPropertyNames() {
List<String> keys = new ArrayList<>();
findKeys(keys, this.getContext());
return keys.toArray(new String[0]);
}
protected void findKeys(List<String> keys, String path) {
log.trace("enter findKeysCached for path: " + path);
Map<String, ChildData> children = cache.getCurrentChildren(path);
if (children == null)
return;
for (Map.Entry<String, ChildData> entry : children.entrySet()) {
ChildData child = entry.getValue();
if (child.getData() == null || child.getData().length == 0) {
findKeys(keys, child.getPath());
}
else {
keys.add(sanitizeKey(child.getPath()));
}
}
log.trace("leaving findKeysCached for path: " + path);
}
@Override
public void stop() {
cache.close();
running = false;
}
@Override
public boolean isRunning() {
return running;
}
}

4
spring-cloud-zookeeper-config/src/main/resources/META-INF/spring.factories
View File

@@ -1,3 +1,7 @@
# Auto Configuration
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.zookeeper.config.ZookeeperConfigAutoConfiguration
# Bootstrap Configuration
org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.zookeeper.config.ZookeeperConfigBootstrapConfiguration

97
spring-cloud-zookeeper-config/src/test/java/org/springframework/cloud/zookeeper/config/ZookeeperPropertySourceLocatorTests.java
View File

@@ -16,25 +16,34 @@
package org.springframework.cloud.zookeeper.config;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import java.util.List;
import java.util.UUID;
import lombok.SneakyThrows;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.RetryOneTime;
import org.apache.zookeeper.KeeperException;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.context.scope.refresh.RefreshScope;
import org.springframework.cloud.endpoint.RefreshEndpoint;
import org.springframework.cloud.zookeeper.ZookeeperProperties;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.ConfigurableEnvironment;
import lombok.SneakyThrows;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
/**
* @author Spencer Gibb
*/
@@ -44,50 +53,78 @@ public class ZookeeperPropertySourceLocatorTests {
public static final String PREFIX = "test__config__";
public static final String ROOT = "/"+PREFIX + UUID.randomUUID();
private ConfigurableApplicationContext context;
public static final String KEY = ROOT + "/application/testProp";
@Configuration
@EnableAutoConfiguration
static class Config {
@Bean
public CountDownLatch countDownLatch() {
return new CountDownLatch(1);
}
@Bean
public RefreshEndpoint refreshEndpoint(ConfigurableApplicationContext context,
RefreshScope scope) {
RefreshEndpoint endpoint = new TestRefreshEndpoint(context, scope, countDownLatch());
return endpoint;
}
}
static class TestRefreshEndpoint extends RefreshEndpoint {
private CountDownLatch latch;
public TestRefreshEndpoint( ConfigurableApplicationContext context, RefreshScope scope, CountDownLatch latch) {
super(context, scope);
this.latch = latch;
}
@Override
public synchronized String[] refresh() {
String[] keys = super.refresh();
this.latch.countDown();
return keys;
}
}
CuratorFramework curator;
ZookeeperConfigProperties properties;
@Before
@SneakyThrows
public void setup(boolean cacheEnabled) {
curator = CuratorFrameworkFactory.builder()
public void setup() {
this.curator = CuratorFrameworkFactory.builder()
.retryPolicy(new RetryOneTime(500))
.connectString(new ZookeeperProperties().getConnectString())
.build();
curator.start();
this.curator.start();
List<String> children = curator.getChildren().forPath("/");
List<String> children = this.curator.getChildren().forPath("/");
for (String child : children) {
if (child.startsWith(PREFIX) && child.length() > PREFIX.length()) {
delete("/" + child);
}
}
String key = ROOT + "/application/testProp";
String create = curator.create().creatingParentsIfNeeded().forPath(key, "testPropVal".getBytes());
curator.close();
String create = this.curator.create().creatingParentsIfNeeded().forPath(KEY, "testPropVal".getBytes());
this.curator.close();
System.out.println(create);
context = new SpringApplicationBuilder(Config.class)
this.context = new SpringApplicationBuilder(Config.class)
.web(false)
.run("--spring.spring.application.name=testZkPropertySource",
"--spring.cloud.zookeeper.config.cacheEnabled="+ cacheEnabled,
"--spring.cloud.zookeeper.config.root="+ROOT);
curator = context.getBean(CuratorFramework.class);
properties = context.getBean(ZookeeperConfigProperties.class);
environment = context.getEnvironment();
this.curator = context.getBean(CuratorFramework.class);
this.properties = context.getBean(ZookeeperConfigProperties.class);
this.environment = context.getEnvironment();
}
@SneakyThrows
public void delete(String path) {
try {
curator.delete().deletingChildrenIfNeeded().forPath(path);
this.curator.delete().deletingChildrenIfNeeded().forPath(path);
} catch (KeeperException e) {
if (e.code() != KeeperException.Code.NONODE) {
throw e;
@@ -95,6 +132,7 @@ public class ZookeeperPropertySourceLocatorTests {
}
}
@After
@SneakyThrows
public void after() {
try {
@@ -105,18 +143,17 @@ public class ZookeeperPropertySourceLocatorTests {
}
@Test
public void propertyLoadedCached() {
setup(true);
public void propertyLoadedAndUpdated() throws Exception {
String testProp = this.environment.getProperty("testProp");
assertThat("testProp was wrong", testProp, is(equalTo("testPropVal")));
after();
}
@Test
public void propertyLoadedNoCache() {
setup(false);
String testProp = this.environment.getProperty("testProp");
assertThat("testProp was wrong", testProp, is(equalTo("testPropVal")));
after();
this.curator.setData().forPath(KEY, "testPropValUpdate".getBytes());
CountDownLatch latch = this.context.getBean(CountDownLatch.class);
boolean receivedEvent = latch.await(5, TimeUnit.SECONDS);
assertThat("listener didn't receive event", receivedEvent, is(true));
testProp = this.environment.getProperty("testProp");
assertThat("testProp was wrong after update", testProp, is(equalTo("testPropValUpdate")));
}
}