Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4313: GemfireMetadataSt: Add Listener support

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-4313

* Add `GemfireCacheListener` in `GemfireMetadataStore` to delegate cache
events to `MetadataStoreListener`s.
* Add GemfireMetadataStore cache listener tests.
* Update ascii doc.
* Addresses review comments.
* Some polishing to Java 8 style
* Some code style polishing
* Minor Doc improvement
This commit is contained in:
Venil Noronha
2017-09-18 01:24:34 -07:00
committed by Artem Bilan
parent 40783ff547
commit b1865dab3a
4 changed files with 236 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStore.java
View File

@@ -16,11 +16,19 @@
package org.springframework.integration.gemfire.metadata;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArrayList;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.metadata.ListenableMetadataStore;
import org.springframework.integration.metadata.MetadataStoreListener;
import org.springframework.util.Assert;
/**
@@ -30,22 +38,30 @@ import org.springframework.util.Assert;
* restarts.
*
* @author Artem Bilan
* @author Venil Noronha
*
* @since 4.0
*/
public class GemfireMetadataStore implements ConcurrentMetadataStore {
public class GemfireMetadataStore implements ListenableMetadataStore {
public static final String KEY = "MetaData";
private final GemfireCacheListener cacheListener = new GemfireCacheListener();
private final Region<String, String> region;
public GemfireMetadataStore(Cache cache) {
Assert.notNull(cache, "'cache' must not be null");
this.region = cache.<String, String>createRegionFactory().setScope(Scope.LOCAL).create(KEY);
this(Objects.requireNonNull(cache, "'cache' must not be null")
.<String, String>createRegionFactory()
.setScope(Scope.LOCAL)
.create(KEY));
}
public GemfireMetadataStore(Region<String, String> region) {
Assert.notNull(region, "'region' must not be null");
this.region = region;
this.region.getAttributesMutator()
.addCacheListener(this.cacheListener);
}
@Override
@@ -82,4 +98,40 @@ public class GemfireMetadataStore implements ConcurrentMetadataStore {
return this.region.remove(key);
}
@Override
public void addListener(MetadataStoreListener listener) {
Assert.notNull(listener, "'listener' must not be null");
this.cacheListener.listeners.add(listener);
}
@Override
public void removeListener(MetadataStoreListener listener) {
this.cacheListener.listeners.remove(listener);
}
private static class GemfireCacheListener extends CacheListenerAdapter<String, String> {
private final List<MetadataStoreListener> listeners = new CopyOnWriteArrayList<>();
GemfireCacheListener() {
super();
}
@Override
public void afterCreate(EntryEvent<String, String> event) {
this.listeners.forEach(listener -> listener.onAdd(event.getKey(), event.getNewValue()));
}
@Override
public void afterUpdate(EntryEvent<String, String> event) {
this.listeners.forEach(listener -> listener.onUpdate(event.getKey(), event.getNewValue()));
}
@Override
public void afterDestroy(EntryEvent<String, String> event) {
this.listeners.forEach(listener -> listener.onRemove(event.getKey(), event.getOldValue()));
}
}
}

158
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/metadata/GemfireMetadataStoreCacheListenerTests.java Normal file
View File

@@ -0,0 +1,158 @@
/*
* Copyright 2017 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.gemfire.metadata;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.Region;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.integration.metadata.MetadataStoreListenerAdapter;
import org.springframework.util.Assert;
/**
* @author Venil Noronha
*
* @since 5.0
*
*/
public class GemfireMetadataStoreCacheListenerTests {
private static Cache cache;
private static GemfireMetadataStore metadataStore;
private static Region<Object, Object> region;
@BeforeClass
public static void startUp() throws Exception {
cache = new CacheFactory().create();
metadataStore = new GemfireMetadataStore(cache);
region = cache.getRegion(GemfireMetadataStore.KEY);
}
@AfterClass
public static void cleanUp() {
if (region != null) {
region.close();
}
if (cache != null) {
cache.close();
Assert.isTrue(cache.isClosed(), "Cache did not close after close() call");
}
}
@Before
@After
public void setup() {
if (region != null) {
region.clear();
}
}
@Test
public void testAdd() throws InterruptedException {
String testKey = "key";
String testValue = "value";
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> actualKey = new AtomicReference<>();
AtomicReference<String> actualValue = new AtomicReference<>();
metadataStore.addListener(new MetadataStoreListenerAdapter() {
@Override
public void onAdd(String key, String value) {
actualKey.set(key);
actualValue.set(value);
latch.countDown();
}
});
metadataStore.put(testKey, testValue);
latch.await(10, TimeUnit.SECONDS);
assertEquals(testKey, actualKey.get());
assertEquals(testValue, actualValue.get());
}
@Test
public void testRemove() throws InterruptedException {
String testKey = "key";
String testValue = "value";
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> actualKey = new AtomicReference<>();
AtomicReference<String> actualValue = new AtomicReference<>();
metadataStore.addListener(new MetadataStoreListenerAdapter() {
@Override
public void onRemove(String key, String oldValue) {
actualKey.set(key);
actualValue.set(oldValue);
latch.countDown();
}
});
metadataStore.put(testKey, testValue);
metadataStore.remove(testKey);
latch.await(10, TimeUnit.SECONDS);
assertEquals(testKey, actualKey.get());
assertEquals(testValue, actualValue.get());
}
@Test
public void testUpdate() throws InterruptedException {
String testKey = "key";
String testValue = "value";
String testNewValue = "new-value";
CountDownLatch latch = new CountDownLatch(1);
AtomicReference<String> actualKey = new AtomicReference<>();
AtomicReference<String> actualValue = new AtomicReference<>();
metadataStore.addListener(new MetadataStoreListenerAdapter() {
@Override
public void onUpdate(String key, String newValue) {
actualKey.set(key);
actualValue.set(newValue);
latch.countDown();
}
});
metadataStore.put(testKey, testValue);
metadataStore.put(testKey, testNewValue);
latch.await(10, TimeUnit.SECONDS);
assertEquals(testKey, actualKey.get());
assertEquals(testNewValue, actualValue.get());
}
}