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

Fix compatibility with latest Spring Data Geode

Browse Source 
* Some configs and code style polishing in the Gemfire module tests
and in the `DefaultHeaderChannelRegistry`
This commit is contained in:
Artem Bilan
2018-05-18 15:14:08 -04:00
parent 1f05d3ec4c
commit 096bec5627
7 changed files with 51 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2016 the original author or authors.
* Copyright 2013-2018 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.
@@ -29,6 +29,7 @@ import org.springframework.context.Lifecycle;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.channel.HeaderChannelRegistry;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -42,6 +43,7 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0
*
*/
@@ -50,15 +52,15 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
private static final int DEFAULT_REAPER_DELAY = 60000;
protected final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<String, DefaultHeaderChannelRegistry.MessageChannelWrapper>();
protected static final AtomicLong id = new AtomicLong();
protected final Map<String, MessageChannelWrapper> channels = new ConcurrentHashMap<>();
protected final String uuid = UUID.randomUUID().toString() + ":";
private volatile boolean removeOnGet;
private boolean removeOnGet;
private volatile long reaperDelay;
private long reaperDelay;
private volatile ScheduledFuture<?> reaperScheduledFuture;
@@ -118,15 +120,17 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
@Override
protected void onInit() throws Exception {
super.onInit();
Assert.notNull(this.getTaskScheduler(), "a task scheduler is required");
Assert.notNull(getTaskScheduler(), "a task scheduler is required");
}
@Override
public synchronized void start() {
if (!this.running) {
Assert.notNull(this.getTaskScheduler(), "a task scheduler is required");
this.reaperScheduledFuture = this.getTaskScheduler().schedule(this,
new Date(System.currentTimeMillis() + this.reaperDelay));
Assert.notNull(getTaskScheduler(), "a task scheduler is required");
this.reaperScheduledFuture =
getTaskScheduler()
.schedule(this, new Date(System.currentTimeMillis() + this.reaperDelay));
this.running = true;
}
}
@@ -142,7 +146,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
}
public void stop(Runnable callback) {
this.stop();
stop();
callback.run();
}
@@ -152,17 +156,17 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
}
@Override
public Object channelToChannelName(Object channel) {
public Object channelToChannelName(@Nullable Object channel) {
return channelToChannelName(channel, this.reaperDelay);
}
@Override
public Object channelToChannelName(Object channel, long timeToLive) {
public Object channelToChannelName(@Nullable Object channel, long timeToLive) {
if (!this.running && !this.explicitlyStopped && this.getTaskScheduler() != null) {
start();
}
if (channel != null && channel instanceof MessageChannel) {
String name = this.uuid + DefaultHeaderChannelRegistry.id.incrementAndGet();
String name = this.uuid + id.incrementAndGet();
this.channels.put(name, new MessageChannelWrapper((MessageChannel) channel,
System.currentTimeMillis() + timeToLive));
if (logger.isDebugEnabled()) {
@@ -176,7 +180,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
}
@Override
public MessageChannel channelNameToChannel(String name) {
public MessageChannel channelNameToChannel(@Nullable String name) {
if (name != null) {
MessageChannelWrapper messageChannelWrapper;
if (this.removeOnGet) {
@@ -188,6 +192,7 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
if (logger.isDebugEnabled() && messageChannelWrapper != null) {
logger.debug("Retrieved " + messageChannelWrapper.getChannel() + " with " + name);
}
return messageChannelWrapper == null ? null : messageChannelWrapper.getChannel();
}
return null;
@@ -202,7 +207,8 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
this.reaperScheduledFuture.cancel(true);
this.reaperScheduledFuture = null;
}
this.run();
run();
}
@Override
@@ -221,8 +227,10 @@ public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport
iterator.remove();
}
}
this.reaperScheduledFuture = this.getTaskScheduler().schedule(this,
new Date(System.currentTimeMillis() + this.reaperDelay));
this.reaperScheduledFuture =
getTaskScheduler()
.schedule(this, new Date(System.currentTimeMillis() + this.reaperDelay));
if (logger.isTraceEnabled()) {
logger.trace("Reaper completed; channels size=" + this.channels.size());
}

3
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParserTests-context.xml
View File

@@ -1,15 +1,12 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:gfe="http://www.springframework.org/schema/gemfire"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire
http://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire.xsd
http://www.springframework.org/schema/gemfire
http://www.springframework.org/schema/gemfire/spring-gemfire.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans

23
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -28,8 +28,8 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.data.gemfire.CacheFactoryBean;
import org.springframework.data.gemfire.GenericRegionFactoryBean;
import org.springframework.data.gemfire.RegionAttributesFactoryBean;
import org.springframework.data.gemfire.RegionFactoryBean;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
@@ -38,6 +38,7 @@ import org.springframework.messaging.Message;
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class CacheListeningMessageProducerTests {
@@ -46,7 +47,7 @@ public class CacheListeningMessageProducerTests {
private static CacheFactoryBean cacheFactoryBean;
private static RegionFactoryBean<String, String> regionFactoryBean;
private static GenericRegionFactoryBean<String, String> regionFactoryBean;
private static Region<String, String> region;
@@ -54,9 +55,7 @@ public class CacheListeningMessageProducerTests {
public static void setup() throws Exception {
cacheFactoryBean = new CacheFactoryBean();
regionFactoryBean = new RegionFactoryBean<String, String>() {
};
regionFactoryBean = new GenericRegionFactoryBean<>();
regionFactoryBean.setName("test.receiveNewValuePayloadForCreateEvent");
regionFactoryBean.setCache(cacheFactoryBean.getObject());
setRegionAttributes(regionFactoryBean);
@@ -72,7 +71,7 @@ public class CacheListeningMessageProducerTests {
}
@Test
public void receiveNewValuePayloadForCreateEvent() throws Exception {
public void receiveNewValuePayloadForCreateEvent() {
QueueChannel channel = new QueueChannel();
CacheListeningMessageProducer producer = new CacheListeningMessageProducer(region);
producer.setPayloadExpression(PARSER.parseExpression("key + '=' + newValue"));
@@ -91,7 +90,7 @@ public class CacheListeningMessageProducerTests {
}
@Test
public void receiveNewValuePayloadForUpdateEvent() throws Exception {
public void receiveNewValuePayloadForUpdateEvent() {
QueueChannel channel = new QueueChannel();
CacheListeningMessageProducer producer = new CacheListeningMessageProducer(region);
producer.setPayloadExpression(PARSER.parseExpression("newValue"));
@@ -114,7 +113,7 @@ public class CacheListeningMessageProducerTests {
}
@Test
public void receiveOldValuePayloadForDestroyEvent() throws Exception {
public void receiveOldValuePayloadForDestroyEvent() {
QueueChannel channel = new QueueChannel();
CacheListeningMessageProducer producer = new CacheListeningMessageProducer(region);
producer.setSupportedEventTypes(EventType.DESTROYED);
@@ -136,7 +135,7 @@ public class CacheListeningMessageProducerTests {
}
@Test
public void receiveOldValuePayloadForInvalidateEvent() throws Exception {
public void receiveOldValuePayloadForInvalidateEvent() {
QueueChannel channel = new QueueChannel();
CacheListeningMessageProducer producer = new CacheListeningMessageProducer(region);
producer.setSupportedEventTypes(EventType.INVALIDATED);
@@ -158,7 +157,9 @@ public class CacheListeningMessageProducerTests {
}
@SuppressWarnings("unchecked")
private static void setRegionAttributes(RegionFactoryBean<String, String> regionFactoryBean) throws Exception {
private static void setRegionAttributes(GenericRegionFactoryBean<String, String> regionFactoryBean)
throws Exception {
RegionAttributesFactoryBean attributesFactoryBean = new RegionAttributesFactoryBean();
attributesFactoryBean.afterPropertiesSet();
regionFactoryBean.setAttributes(attributesFactoryBean.getObject());

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml
View File

@@ -10,7 +10,7 @@
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">
<gfe:cache id="gemfire-cache-2" use-bean-factory-locator="false"/>
<gfe:cache id="gemfire-cache-2"/>
<gfe:local-region id="region1" cache-ref="gemfire-cache-2"/>

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml
View File

@@ -9,7 +9,7 @@
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<gfe:cache use-bean-factory-locator="false"/>
<gfe:cache/>
<gfe:replicated-region id="region1"/>

23
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -33,7 +33,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.data.gemfire.CacheFactoryBean;
import org.springframework.data.gemfire.RegionFactoryBean;
import org.springframework.data.gemfire.GenericRegionFactoryBean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.store.MessageGroup;
@@ -48,6 +48,7 @@ import org.springframework.messaging.support.GenericMessage;
* @author David Turanski
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.1
*/
public class GemfireMessageStoreTests {
@@ -57,7 +58,7 @@ public class GemfireMessageStoreTests {
private static Region<Object, Object> region;
@Test
public void addAndGetMessage() throws Exception {
public void addAndGetMessage() {
GemfireMessageStore store = new GemfireMessageStore(region);
Message<?> message = MessageBuilder.withPayload("test").build();
store.addMessage(message);
@@ -67,9 +68,7 @@ public class GemfireMessageStoreTests {
@Test
public void testRegionConstructor() throws Exception {
RegionFactoryBean<Object, Object> region = new RegionFactoryBean<Object, Object>() {
};
GenericRegionFactoryBean<Object, Object> region = new GenericRegionFactoryBean<>();
region.setName("someRegion");
region.setCache(cacheFactoryBean.getObject());
region.afterPropertiesSet();
@@ -81,10 +80,10 @@ public class GemfireMessageStoreTests {
}
@Test
public void testWithMessageHistory() throws Exception {
public void testWithMessageHistory() {
GemfireMessageStore store = new GemfireMessageStore(region);
Message<?> message = new GenericMessage<String>("Hello");
Message<?> message = new GenericMessage<>("Hello");
DirectChannel fooChannel = new DirectChannel();
fooChannel.setBeanName("fooChannel");
DirectChannel barChannel = new DirectChannel();
@@ -103,11 +102,11 @@ public class GemfireMessageStoreTests {
}
@Test
public void testAddAndRemoveMessagesFromMessageGroup() throws Exception {
public void testAddAndRemoveMessagesFromMessageGroup() {
GemfireMessageStore messageStore = new GemfireMessageStore(region);
String groupId = "X";
List<Message<?>> messages = new ArrayList<Message<?>>();
List<Message<?>> messages = new ArrayList<>();
for (int i = 0; i < 25; i++) {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessagesToGroup(groupId, message);
@@ -121,11 +120,11 @@ public class GemfireMessageStoreTests {
}
@Test
public void testAddAndRemoveMessagesFromMessageGroupWithPrefix() throws Exception {
public void testAddAndRemoveMessagesFromMessageGroupWithPrefix() {
GemfireMessageStore messageStore = new GemfireMessageStore(region, "foo_");
String groupId = "X";
List<Message<?>> messages = new ArrayList<Message<?>>();
List<Message<?>> messages = new ArrayList<>();
for (int i = 0; i < 25; i++) {
Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build();
messageStore.addMessagesToGroup(groupId, message);

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml
View File

@@ -7,7 +7,7 @@
http://www.springframework.org/schema/geode http://www.springframework.org/schema/gemfire/spring-geode.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<gfe:cache use-bean-factory-locator="false"/>
<gfe:cache/>
<bean id="lockRegistry" class="org.springframework.integration.gemfire.util.GemfireLockRegistry">
<constructor-arg ref="gemfireCache"/>