From 096bec56276ebe519c2f20eedf5b9ffe6bec0595 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 18 May 2018 15:14:08 -0400 Subject: [PATCH] Fix compatibility with latest Spring Data Geode * Some configs and code style polishing in the Gemfire module tests and in the `DefaultHeaderChannelRegistry` --- .../channel/DefaultHeaderChannelRegistry.java | 42 +++++++++++-------- ...boundChannelAdapterParserTests-context.xml | 3 -- .../CacheListeningMessageProducerTests.java | 23 +++++----- ...fireInboundChannelAdapterTests-context.xml | 2 +- ...ireOutboundChannelAdapterTests-context.xml | 2 +- .../store/GemfireMessageStoreTests.java | 23 +++++----- ...ggregatorWithGemfireLocksTests-context.xml | 2 +- 7 files changed, 51 insertions(+), 46 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java index f5bc908543..b655118251 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DefaultHeaderChannelRegistry.java @@ -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 channels = new ConcurrentHashMap(); - protected static final AtomicLong id = new AtomicLong(); + protected final Map 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()); } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParserTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParserTests-context.xml index f26b5c5fe9..9f8aeb41ff 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParserTests-context.xml @@ -1,15 +1,12 @@ regionFactoryBean; + private static GenericRegionFactoryBean regionFactoryBean; private static Region region; @@ -54,9 +55,7 @@ public class CacheListeningMessageProducerTests { public static void setup() throws Exception { cacheFactoryBean = new CacheFactoryBean(); - regionFactoryBean = new RegionFactoryBean() { - - }; + 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 regionFactoryBean) throws Exception { + private static void setRegionAttributes(GenericRegionFactoryBean regionFactoryBean) + throws Exception { + RegionAttributesFactoryBean attributesFactoryBean = new RegionAttributesFactoryBean(); attributesFactoryBean.afterPropertiesSet(); regionFactoryBean.setAttributes(attributesFactoryBean.getObject()); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml index c0b9b40c22..f0560c6970 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml @@ -10,7 +10,7 @@ http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml index 5e458e22a5..c91780b5da 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml @@ -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"> - + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java index 28308c27c6..5102af075d 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireMessageStoreTests.java @@ -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 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 region = new RegionFactoryBean() { - - }; + GenericRegionFactoryBean 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("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> messages = new ArrayList>(); + List> messages = new ArrayList<>(); for (int i = 0; i < 25; i++) { Message 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> messages = new ArrayList>(); + List> messages = new ArrayList<>(); for (int i = 0; i < 25; i++) { Message message = MessageBuilder.withPayload("foo").setCorrelationId(groupId).build(); messageStore.addMessagesToGroup(groupId, message); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml index 08d3bd214c..8f64d78850 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests-context.xml @@ -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"> - +