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

INTEXT-161 Add <poller> to the <int-hz:o-c-a>

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-161
This commit is contained in:
Artem Bilan
2015-04-04 17:32:38 +03:00
parent bfec7640ca
commit 4ed2153a45
3 changed files with 103 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
spring-integration-hazelcast/src/main/resources/org/springframework/integration/hazelcast/config/xml/spring-integration-hazelcast-1.0.xsd
View File

@@ -7,7 +7,7 @@
targetNamespace="http://www.springframework.org/schema/integration/hazelcast"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"
schemaLocation="http://www.springframework.org/schema/beans/spring-beans.xsd"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
@@ -15,16 +15,16 @@
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures Hazelcast Event-Driven Inbound Channel Adapter
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="cache" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -51,7 +51,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
@@ -63,26 +63,27 @@
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures Hazelcast Outbound Channel Adapter
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:all>
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:choice>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="cache" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -95,7 +96,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -104,22 +105,22 @@
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="cq-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures Hazelcast Continuous Query Inbound Channel Adapter
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="cache" use="required" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
@@ -146,7 +147,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="predicate" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
@@ -154,7 +155,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="include-value" type="xsd:boolean" default="true">
<xsd:annotation>
<xsd:documentation>
@@ -162,7 +163,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="cache-listening-policy" default="SINGLE" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[ Specifies cache listening policy. ]]></xsd:documentation>
@@ -174,24 +175,24 @@
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="ds-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Configures Hazelcast Distributed SQL Inbound Channel Adapter
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="cache" use="required" type="xsd:string">
@@ -206,7 +207,7 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="iteration-type" default="VALUE" use="optional">
<xsd:annotation>
<xsd:documentation><![CDATA[ Specifies Distributed-SQL Iteration Types. ]]></xsd:documentation>
@@ -220,7 +221,7 @@
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="distributed-sql" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
@@ -228,8 +229,8 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
</xsd:schema>

24
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests-context.xml
View File

@@ -11,14 +11,34 @@
http://www.springframework.org/schema/integration/hazelcast/spring-integration-hazelcast.xsd">
<int:channel id="mapChannel"/>
<int:channel id="listChannel"/>
<int:channel id="setChannel"/>
<int:channel id="queueChannel"/>
<int:channel id="queueChannel">
<int:queue/>
</int:channel>
<int:channel id="errorChannel">
<int:queue/>
</int:channel>
<int-hazelcast:outbound-channel-adapter channel="mapChannel" cache="distributedMap"/>
<int-hazelcast:outbound-channel-adapter channel="listChannel" cache="distributedList"/>
<int-hazelcast:outbound-channel-adapter channel="setChannel" cache="distributedSet"/>
<int-hazelcast:outbound-channel-adapter channel="queueChannel" cache="distributedQueue"/>
<bean id="testRequestHandlerAdvice"
class="org.springframework.integration.hazelcast.outbound.HazelcastOutboundChannelAdapterTests$TestRequestHandlerAdvice"/>
<int-hazelcast:outbound-channel-adapter channel="queueChannel" cache="distributedQueue">
<int:poller fixed-delay="100"/>
<int-hazelcast:request-handler-advice-chain>
<ref bean="testRequestHandlerAdvice"/>
</int-hazelcast:request-handler-advice-chain>
</int-hazelcast:outbound-channel-adapter>
<bean id="distributedMap" factory-bean="instance" factory-method="getMap">
<constructor-arg value="distributedMap"/>

60
spring-integration-hazelcast/src/test/java/org/springframework/integration/hazelcast/outbound/HazelcastOutboundChannelAdapterTests.java
View File

@@ -16,6 +16,11 @@
package org.springframework.integration.hazelcast.outbound;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -25,7 +30,9 @@ import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.annotation.Resource;
@@ -35,8 +42,12 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
@@ -46,6 +57,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* Hazelcast Outbound Channel Adapter Test Class
*
* @author Eren Avsarogullari
* @author Artem Bilan
* @since 1.0.0
*/
@RunWith(SpringJUnit4ClassRunner.class)
@@ -67,6 +79,9 @@ public class HazelcastOutboundChannelAdapterTests {
@Autowired
private MessageChannel queueChannel;
@Autowired
private PollableChannel errorChannel;
@Resource
private Map<?, ?> distributedMap;
@@ -79,6 +94,9 @@ public class HazelcastOutboundChannelAdapterTests {
@Resource
private Queue<?> distributedQueue;
@Autowired
private TestRequestHandlerAdvice testRequestHandlerAdvice;
@Before
public void setUp() {
distributedMap.clear();
@@ -109,11 +127,17 @@ public class HazelcastOutboundChannelAdapterTests {
}
@Test
public void testWriteDistributedQueue() {
Queue<Integer> queue = (Queue<Integer>) fillCollectionByEntryCount(
new LinkedBlockingQueue<Integer>(DATA_COUNT));
queueChannel.send(new GenericMessage<>(queue));
verifyDistributedQueue();
public void testWriteDistributedQueue() throws InterruptedException {
Collection<Integer> queue = fillCollectionByEntryCount(new LinkedBlockingQueue<Integer>(DATA_COUNT));
this.queueChannel.send(new GenericMessage<>(queue));
assertTrue(this.testRequestHandlerAdvice.executeLatch.await(10, TimeUnit.SECONDS));
Assert.assertEquals(true, this.distributedQueue.size() == DATA_COUNT);
int index = 0;
for (Object o : this.distributedQueue) {
Assert.assertEquals(index++, o);
}
}
@Test(expected = MessageHandlingException.class)
@@ -137,11 +161,15 @@ public class HazelcastOutboundChannelAdapterTests {
setChannel.send(new GenericMessage<>(list));
}
@Test(expected = MessageHandlingException.class)
public void testQueueChannelWithIncorrectDataType() {
Set<Integer> set = new HashSet<>();
set.add(1);
queueChannel.send(new GenericMessage<>(set));
this.queueChannel.send(new GenericMessage<>(set));
Message<?> receive = this.errorChannel.receive(10000);
assertNotNull(receive);
assertThat(receive, instanceOf(ErrorMessage.class));
assertThat(receive.getPayload(), instanceOf(MessageHandlingException.class));
}
private Map<Integer, String> createMapByEntryCount() {
@@ -192,12 +220,20 @@ public class HazelcastOutboundChannelAdapterTests {
}
}
private void verifyDistributedQueue() {
Assert.assertEquals(true, distributedQueue.size() == DATA_COUNT);
int index = 0;
for (Object o : distributedQueue) {
Assert.assertEquals(index++, o);
public static class TestRequestHandlerAdvice extends AbstractRequestHandlerAdvice {
public final CountDownLatch executeLatch = new CountDownLatch(1);
@Override
protected Object doInvoke(ExecutionCallback callback, Object target, Message<?> message) throws Exception {
try {
return callback.execute();
}
finally {
this.executeLatch.countDown();
}
}
}
}