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

INT-2082 added support for evaluating payload expression in ContinuousQueryMessageProducer

Browse Source
This commit is contained in:
David Turanski
2011-08-25 13:06:00 -04:00
parent a97ddfa71e
commit 2cade4220e
6 changed files with 108 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java
View File

@@ -22,10 +22,6 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
@@ -47,7 +43,7 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
* @since 2.1
*/
@SuppressWarnings({"rawtypes", "unchecked"})
public class CacheListeningMessageProducer extends MessageProducerSupport {
public class CacheListeningMessageProducer extends SpelMessageProducerSupport {
private final Log logger = LogFactory.getLog(this.getClass());
@@ -58,9 +54,6 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
private volatile Set<EventType> supportedEventTypes =
new HashSet<EventType>(Arrays.asList(EventType.CREATED, EventType.UPDATED));
private volatile Expression payloadExpression;
private final SpelExpressionParser parser = new SpelExpressionParser();
public CacheListeningMessageProducer(Region<?, ?> region) {
Assert.notNull(region, "region must not be null");
@@ -74,14 +67,6 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
this.supportedEventTypes = new HashSet<EventType>(Arrays.asList(eventTypes));
}
public void setPayloadExpression(String payloadExpression) {
if (payloadExpression == null) {
this.payloadExpression = null;
}
else {
this.payloadExpression = this.parser.parseExpression(payloadExpression);
}
}
@Override
protected void doStart() {
@@ -105,8 +90,7 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
}
}
private class MessageProducingCacheListener extends CacheListenerAdapter {
@Override
@@ -137,14 +121,9 @@ public class CacheListeningMessageProducer extends MessageProducerSupport {
}
}
private void processEvent(EntryEvent event) {
if (payloadExpression != null) {
Object evaluationResult = payloadExpression.getValue(event);
this.publish(evaluationResult);
}
else {
this.publish(event);
}
private void processEvent(EntryEvent event) {
this.publish(evaluationResult(event));
}
private void publish(Object payload) {

10
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
View File

@@ -16,15 +16,15 @@
package org.springframework.integration.gemfire.inbound;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.gemfire.listener.CqQueryDefinition;
import org.springframework.data.gemfire.listener.QueryListener;
import org.springframework.data.gemfire.listener.QueryListenerContainer;
import org.springframework.integration.Message;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import com.gemstone.gemfire.cache.query.CqEvent;
/**
@@ -38,7 +38,7 @@ import com.gemstone.gemfire.cache.query.CqEvent;
* @since 2.1
*
*/
public class ContinuousQueryMessageProducer extends MessageProducerSupport implements QueryListener {
public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport implements QueryListener {
private static Log logger = LogFactory.getLog(ContinuousQueryMessageProducer.class);
private final String query;
@@ -95,7 +95,7 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport imple
if (logger.isDebugEnabled()){
logger.debug(String.format("processing cq event key [%s] event [%s]",event.getBaseOperation().toString(),event.getKey()));
}
Message<CqEvent> cqEventMessage = MessageBuilder.withPayload(event).build();
Message<?> cqEventMessage = MessageBuilder.withPayload(evaluationResult(event)).build();
sendMessage(cqEventMessage);
}

54
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2002-2011 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.inbound;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.endpoint.MessageProducerSupport;
/**
* @author David Turanski
* @since 2.1
*
*/
public class SpelMessageProducerSupport extends MessageProducerSupport {
private volatile Expression payloadExpression;
private final SpelExpressionParser parser = new SpelExpressionParser();
@Override
protected void onInit(){
super.onInit();
}
public void setPayloadExpression(String payloadExpression) {
if (payloadExpression == null) {
this.payloadExpression = null;
}
else {
this.payloadExpression = this.parser.parseExpression(payloadExpression);
}
}
protected Object evaluationResult(Object payload){
Object evaluationResult = payload;
if (payloadExpression != null) {
evaluationResult = payloadExpression.getValue(payload);
}
return evaluationResult;
}
}

1
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java
View File

@@ -23,7 +23,6 @@ import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

18
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml
View File

@@ -23,14 +23,26 @@
<property name="cache" ref="client-cache"/>
</bean>
<bean class="org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer">
<bean id="cqMessageProducer" class="org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer">
<constructor-arg ref="queryListenerContainer"/>
<constructor-arg value="select * from /test"/>
<property name="outputChannel" ref="outputChannel"/>
<property name="outputChannel" ref="outputChannel1"/>
<property name="durable" value="true"/>
</bean>
<int:channel id="outputChannel">
<int:channel id="outputChannel1">
<int:queue/>
</int:channel>
<bean id="spelCqMessageProducer" class="org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer">
<constructor-arg ref="queryListenerContainer"/>
<constructor-arg value="select * from /test"/>
<property name="outputChannel" ref="outputChannel2"/>
<property name="payloadExpression" value="newValue"/>
</bean>
<int:channel id="outputChannel2">
<int:queue/>
</int:channel>
</beans>

45
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java
View File

@@ -12,6 +12,7 @@
*/
package org.springframework.integration.gemfire.inbound.cq;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -19,6 +20,7 @@ import java.io.IOException;
import java.io.OutputStream;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -41,6 +43,8 @@ import com.gemstone.gemfire.internal.cache.LocalRegion;
@ContextConfiguration
public class ContinuousQueryMessageProducerTests {
static ConfigurableApplicationContext staticCtx;
@Autowired
LocalRegion region;
@@ -48,7 +52,10 @@ public class ContinuousQueryMessageProducerTests {
ConfigurableApplicationContext applicationContext;
@Autowired
PollableChannel outputChannel;
PollableChannel outputChannel1;
@Autowired
PollableChannel outputChannel2;
static OutputStream os;
@BeforeClass
@@ -56,30 +63,36 @@ public class ContinuousQueryMessageProducerTests {
os = ForkUtil.cacheServer();
}
@Before
public void setUp() {
staticCtx = applicationContext;
}
@Test
public void test() throws InterruptedException {
public void testCqEvent() throws InterruptedException {
region.put("one",1);
Message<?> msg = outputChannel.receive(1000);
Message<?> msg = outputChannel1.receive(1000);
assertNotNull(msg);
assertTrue(msg.getPayload() instanceof CqEvent);
/*
* Avoid shutdown errors
*/
applicationContext.close();
}
@Test
public void testPayloadExpression() throws InterruptedException {
region.put("one",1);
Message<?> msg = outputChannel2.receive(1000);
assertNotNull(msg);
assertEquals(1,msg.getPayload());
}
@AfterClass
public static void cleanUp() {
try {
Thread.sleep(3000);
}
catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
/*
* Avoid shutdown errors
*/
staticCtx.close();
sendSignal();
}