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

INT-3963: Add XMPP Extensions Support

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3963

* Update to Smack-4.1.6
* Introduce `stanza-filter` option for the `<int-xmpp:inbound-channel-adapter>`
* Introduce `payloadExpression` for the complex and specific `stanza` parsing, e.g. GCM packets
* Deprecate `extract-payload` in favor of `payload-expression`
* Add `ChatMessageListeningEndpointTests` test for GCM protocol
* Add `ChatMessageInboundChannelAdapterParser` test for new attributes
* Document changes

Polishing according PR comments

Extract `#extension` SpEL variable

Document the `#extension` SpEL variable
This commit is contained in:
Artem Bilan
2016-03-14 19:13:45 -04:00
committed by Gary Russell
parent 956cf275e1
commit f7c59b3b18
9 changed files with 443 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/config/ChatMessageInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2016 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.xmpp.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
@@ -27,6 +28,7 @@ import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
*
* @author Josh Long
* @author Oleg Zhurakousky
* @author Artem Bilan
* @since 2.0
*/
public class ChatMessageInboundChannelAdapterParser extends AbstractXmppInboundChannelAdapterParser {
@@ -38,7 +40,17 @@ public class ChatMessageInboundChannelAdapterParser extends AbstractXmppInboundC
@Override
protected void postProcess(Element element, ParserContext parserContext, BeanDefinitionBuilder builder){
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
if (element.hasAttribute("extract-payload")) {
parserContext.getReaderContext()
.warning("The 'extract-payload' is deprecated. Use 'payload-expression' instead.", element);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
}
BeanDefinition expression =
IntegrationNamespaceUtils.createExpressionDefIfAttributeDefined("payload-expression", element);
if (expression != null) {
builder.addPropertyValue("payloadExpression", expression);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "stanza-filter");
}
}

104
spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -16,18 +16,22 @@
package org.springframework.integration.xmpp.inbound;
import java.util.List;
import java.util.Map;
import org.jivesoftware.smack.StanzaListener;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.filter.StanzaFilter;
import org.jivesoftware.smack.packet.ExtensionElement;
import org.jivesoftware.smack.packet.Stanza;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.xmpp.core.AbstractXmppConnectionAwareEndpoint;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.integration.xmpp.support.XmppHeaderMapper;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* This component logs in as a user and forwards any messages <em>to</em> that
@@ -41,11 +45,15 @@ import org.springframework.util.StringUtils;
*/
public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEndpoint {
private volatile boolean extractPayload = true;
private final StanzaListener stanzaListener = new ChatMessagePublishingStanzaListener();
private volatile XmppHeaderMapper headerMapper = new DefaultXmppHeaderMapper();
private XmppHeaderMapper headerMapper = new DefaultXmppHeaderMapper();
private Expression payloadExpression;
private StanzaFilter stanzaFilter;
private EvaluationContext evaluationContext;
public ChatMessageListeningEndpoint() {
super();
@@ -64,11 +72,36 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
* Specify whether the text message body should be extracted when mapping to a
* Spring Integration Message payload. Otherwise, the full XMPP Message will be
* passed within the payload. This value is <em>true</em> by default.
*
* @param extractPayload true if the payload should be extracted.
* @deprecated since version 4.3 in favor of {@link #setPayloadExpression(Expression)}
*/
@Deprecated
public void setExtractPayload(boolean extractPayload) {
this.extractPayload = extractPayload;
if (this.payloadExpression == null) {
setPayloadExpression(extractPayload ? null : EXPRESSION_PARSER.parseExpression("#this"));
}
}
/**
* Specify a {@link StanzaFilter} to use for the incoming packets.
* @param stanzaFilter the {@link StanzaFilter} to use
* @since 4.3
* @see XMPPConnection#addAsyncStanzaListener(StanzaListener, StanzaFilter)
*/
public void setStanzaFilter(StanzaFilter stanzaFilter) {
this.stanzaFilter = stanzaFilter;
}
/**
* Specify a SpEL expression to evaluate a {@code payload} against an incoming
* {@link org.jivesoftware.smack.packet.Message}.
* @param payloadExpression the {@link Expression} for payload evaluation.
* @since 4.3
* @see StanzaListener
* @see org.jivesoftware.smack.packet.Message
*/
public void setPayloadExpression(Expression payloadExpression) {
this.payloadExpression = payloadExpression;
}
@Override
@@ -76,10 +109,16 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
return "xmpp:inbound-channel-adapter";
}
@Override protected void onInit() {
super.onInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
protected void doStart() {
Assert.isTrue(this.initialized, this.getComponentName() + " [" + this.getComponentType() + "] must be initialized");
this.xmppConnection.addAsyncStanzaListener(this.stanzaListener, null);
Assert.isTrue(this.initialized, this.getComponentName() + " [" + this.getComponentType()
+ "] must be initialized");
this.xmppConnection.addAsyncStanzaListener(this.stanzaListener, this.stanzaFilter);
}
@Override
@@ -98,23 +137,36 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) packet;
Map<String, ?> mappedHeaders = headerMapper.toHeadersFromRequest(xmppMessage);
String messageBody = xmppMessage.getBody();
/*
* Since there are several types of chat messages with different ChatState (e.g., composing, paused etc)
* we need to perform further validation since for now we only support messages that have
* content (e.g., Use A says 'Hello' to User B). We don't yet support messages with no
* content (e.g., User A is typing a message for User B etc.).
* See https://jira.springsource.org/browse/INT-1728
* Also see: packet.getExtensions()
*/
if (StringUtils.hasText(messageBody)){
Object payload = (extractPayload ? messageBody : xmppMessage);
Object messageBody = xmppMessage.getBody();
AbstractIntegrationMessageBuilder<?> messageBuilder =
ChatMessageListeningEndpoint.this.getMessageBuilderFactory()
.withPayload(payload)
.copyHeaders(mappedHeaders);
sendMessage(messageBuilder.build());
if (ChatMessageListeningEndpoint.this.payloadExpression != null) {
EvaluationContext evaluationContext = ChatMessageListeningEndpoint.this.evaluationContext;
List<ExtensionElement> extensions = xmppMessage.getExtensions();
if (extensions.size() == 1) {
ExtensionElement extension = extensions.get(0);
evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
evaluationContext.setVariable("extension", extension);
}
messageBody = ChatMessageListeningEndpoint.this.payloadExpression
.getValue(evaluationContext, xmppMessage);
}
if (messageBody != null) {
sendMessage(getMessageBuilderFactory()
.withPayload(messageBody)
.copyHeaders(mappedHeaders).build());
}
else if (logger.isInfoEnabled()) {
if (ChatMessageListeningEndpoint.this.payloadExpression != null) {
logger.info("The 'payloadExpression' ["
+ ChatMessageListeningEndpoint.this.payloadExpression.getExpressionString()
+ "] has been evaluated to 'null'. The XMPP Message [" + xmppMessage + "] is ignored.");
}
else {
logger.info("The XMPP Message [" + xmppMessage + "] with empty body is ignored.");
}
}
}
}

35
spring-integration-xmpp/src/main/resources/org/springframework/integration/xmpp/config/spring-integration-xmpp-4.3.xsd
View File

@@ -106,11 +106,40 @@
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="xmppInboundAdapterType">
<xsd:attribute name="extract-payload" type="xsd:string" default="true">
<xsd:attribute name="extract-payload" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Specifies if generated Message payload should consist of only
the text of the XMPP message or the entire XMPP (Smack API specific) message. Default is true.
[DEPRECATED]
Specifies if generated Message payload should consist of only
the text of the XMPP message or the entire XMPP (Smack API specific) message.
Default is true.
Deprecated since 4.3 in favor of 'payload-expression'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="payload-expression">
<xsd:annotation>
<xsd:documentation>
A SpEL expression to evaluate a 'payload' with the incoming
'org.jivesoftware.smack.packet.Message' as root object.
It useful in case of custom (XEP) XMPP interactions, e.g. GCM.
By default a Message 'body' is used as 'payload'.
The '#extension' SpEL variable is registered in the evaluation context
if one and only one extension is present in the Message.
Replaces 'extract-payload' attribute since 4.3.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="stanza-filter">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.jivesoftware.smack.filter.StanzaFilter"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Reference to an XMPP 'org.jivesoftware.smack.filter.StanzaFilter' bean.
See 'XMPPConnection.addAsyncStanzaListener(StanzaListener, StanzaFilter)' JavaDocs.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

29
spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/ChatMessageInboundChannelAdapterParserTests-context.xml
View File

@@ -9,24 +9,35 @@
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<beans:bean id="testConnection" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.jivesoftware.smack.XMPPConnection"/>
<beans:bean id="testConnection" class="org.mockito.Mockito" factory-method="spy">
<beans:constructor-arg>
<beans:bean class="org.jivesoftware.smack.tcp.XMPPTCPConnection">
<beans:constructor-arg value="guest"/>
<beans:constructor-arg value="guest"/>
</beans:bean>
</beans:constructor-arg>
</beans:bean>
<channel id="xmppInbound">
<queue/>
</channel>
<beans:bean id="stanzaFilter" class="org.mockito.Mockito" factory-method="mock">
<beans:constructor-arg value="org.jivesoftware.smack.filter.StanzaFilter"/>
</beans:bean>
<xmpp:inbound-channel-adapter id="xmppInboundAdapter" channel="xmppInbound"
xmpp-connection="testConnection" extract-payload="false"
auto-startup="false" error-channel="errorChannel"
mapped-request-headers="foo*, xmpp*"/>
xmpp-connection="testConnection" payload-expression="#root"
auto-startup="false" error-channel="errorChannel"
mapped-request-headers="foo*, xmpp*"
stanza-filter="stanzaFilter"/>
<xmpp:inbound-channel-adapter id="autoChannel"
xmpp-connection="testConnection" extract-payload="false"
auto-startup="false" error-channel="errorChannel"
mapped-request-headers="foo*, xmpp*"/>
xmpp-connection="testConnection" extract-payload="false"
auto-startup="false" error-channel="errorChannel"
mapped-request-headers="foo*, xmpp*"/>
<bridge input-channel="autoChannel" output-channel="nullChannel" />
<bridge input-channel="autoChannel" output-channel="nullChannel"/>
</beans:beans>

18
spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/config/ChatMessageInboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2016 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.
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import java.lang.reflect.Field;
import java.util.Map;
import org.jivesoftware.smack.SmackException.NotConnectedException;
import org.jivesoftware.smack.StanzaListener;
@@ -34,10 +35,10 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.xmpp.inbound.ChatMessageListeningEndpoint;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.annotation.DirtiesContext.ClassMode;
import org.springframework.test.context.ContextConfiguration;
@@ -49,6 +50,7 @@ import org.springframework.util.ReflectionUtils;
* @author Mark Fisher
* @author Gunnar Hillert
* @author Florian Schmaus
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -69,6 +71,7 @@ public class ChatMessageInboundChannelAdapterParserTests {
private ChatMessageListeningEndpoint autoChannelAdapter;
@Test
@SuppressWarnings("rawtypes")
public void testInboundAdapter() {
ChatMessageListeningEndpoint adapter = context.getBean("xmppInboundAdapter", ChatMessageListeningEndpoint.class);
MessageChannel errorChannel = (MessageChannel) TestUtils.getPropertyValue(adapter, "errorChannel");
@@ -77,7 +80,16 @@ public class ChatMessageInboundChannelAdapterParserTests {
QueueChannel channel = (QueueChannel) TestUtils.getPropertyValue(adapter, "outputChannel");
assertEquals("xmppInbound", channel.getComponentName());
XMPPConnection connection = (XMPPConnection) TestUtils.getPropertyValue(adapter, "xmppConnection");
assertEquals(connection, context.getBean("testConnection"));
assertSame(connection, context.getBean("testConnection"));
Object stanzaFilter = context.getBean("stanzaFilter");
assertSame(stanzaFilter, TestUtils.getPropertyValue(adapter, "stanzaFilter"));
assertEquals("#root", TestUtils.getPropertyValue(adapter, "payloadExpression.expression"));
adapter.start();
Map asyncRecvListeners = TestUtils.getPropertyValue(connection, "asyncRecvListeners", Map.class);
assertEquals(1, asyncRecvListeners.size());
assertSame(stanzaFilter,
TestUtils.getPropertyValue(asyncRecvListeners.values().iterator().next(), "packetFilter"));
adapter.stop();
}
@Test

134
spring-integration-xmpp/src/test/java/org/springframework/integration/xmpp/inbound/ChatMessageListeningEndpointTests.java
View File

@@ -16,26 +16,40 @@
package org.springframework.integration.xmpp.inbound;
import static org.hamcrest.core.IsInstanceOf.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.mockito.Mockito.doAnswer;
import static org.junit.Assert.assertThat;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.io.StringReader;
import java.util.HashSet;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.jivesoftware.smack.SmackException.NotConnectedException;
import org.jivesoftware.smack.StanzaListener;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.filter.StanzaFilter;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
import org.jivesoftware.smack.util.PacketParserUtils;
import org.jivesoftware.smackx.gcm.packet.GcmPacketExtension;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.xmlpull.v1.XmlPullParser;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
@@ -63,7 +77,7 @@ public class ChatMessageListeningEndpointTests {
XMPPConnection connection = mock(XMPPConnection.class);
ChatMessageListeningEndpoint endpoint = new ChatMessageListeningEndpoint(connection);
doAnswer(new Answer<Object>() {
willAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
@@ -71,17 +85,19 @@ public class ChatMessageListeningEndpointTests {
return null;
}
}).when(connection).addAsyncStanzaListener(Mockito.any(StanzaListener.class), Mockito.any(StanzaFilter.class));
}).given(connection)
.addAsyncStanzaListener(Mockito.any(StanzaListener.class), Mockito.any(StanzaFilter.class));
doAnswer(new Answer<Object>() {
willAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
packetListSet.remove((StanzaListener) invocation.getArguments()[0]);
packetListSet.remove(invocation.getArguments()[0]);
return null;
}
}).when(connection).removeAsyncStanzaListener(Mockito.any(StanzaListener.class));
}).given(connection)
.removeAsyncStanzaListener(Mockito.any(StanzaListener.class));
assertEquals(0, packetListSet.size());
endpoint.setOutputChannel(new QueueChannel());
@@ -151,4 +167,110 @@ public class ChatMessageListeningEndpointTests {
assertEquals("hello", ((MessagingException) msg.getPayload()).getFailedMessage().getPayload());
}
@Test
@SuppressWarnings("deprecation")
public void testExpression() throws Exception {
TestXMPPConnection testXMPPConnection = new TestXMPPConnection();
QueueChannel inputChannel = new QueueChannel();
ChatMessageListeningEndpoint endpoint = new ChatMessageListeningEndpoint(testXMPPConnection);
endpoint.setExtractPayload(false);
endpoint.setOutputChannel(inputChannel);
endpoint.setBeanFactory(mock(BeanFactory.class));
endpoint.afterPropertiesSet();
endpoint.start();
Message smackMessage = new Message();
smackMessage.setBody("foo");
XmlPullParser xmlPullParser = PacketParserUtils.newXmppParser(new StringReader(smackMessage.toString()));
xmlPullParser.next();
testXMPPConnection.parseAndProcessStanza(xmlPullParser);
org.springframework.messaging.Message<?> receive = inputChannel.receive(10000);
assertNotNull(receive);
Object payload = receive.getPayload();
assertThat(payload, instanceOf(Message.class));
assertEquals(smackMessage.getStanzaId(), ((Message) payload).getStanzaId());
assertEquals(smackMessage.getBody(), ((Message) payload).getBody());
Log logger = Mockito.spy(TestUtils.getPropertyValue(endpoint, "logger", Log.class));
given(logger.isInfoEnabled()).willReturn(true);
new DirectFieldAccessor(endpoint).setPropertyValue("logger", logger);
endpoint.setPayloadExpression(null);
smackMessage = new Message();
xmlPullParser = PacketParserUtils.newXmppParser(new StringReader(smackMessage.toString()));
xmlPullParser.next();
testXMPPConnection.parseAndProcessStanza(xmlPullParser);
ArgumentCaptor<String> argumentCaptor = new ArgumentCaptor<String>();
verify(logger).info(argumentCaptor.capture());
assertEquals("The XMPP Message [" + smackMessage + "] with empty body is ignored.",
argumentCaptor.getValue());
endpoint.stop();
}
@Test
public void testGcmExtension() throws Exception {
String data = "{\n" +
" \"to\":\"me\",\n" +
" \"notification\": {\n" +
" \"title\": \"Something interesting\",\n" +
" \"text\": \"Here we go\"\n" +
" },\n" +
" \"time_to_live\":\"600\"\n" +
" }\n" +
"}";
GcmPacketExtension packetExtension = new GcmPacketExtension(data);
Message smackMessage = new Message();
smackMessage.addExtension(packetExtension);
TestXMPPConnection testXMPPConnection = new TestXMPPConnection();
QueueChannel inputChannel = new QueueChannel();
ChatMessageListeningEndpoint endpoint = new ChatMessageListeningEndpoint(testXMPPConnection);
Expression payloadExpression = new SpelExpressionParser().parseExpression("#extension.json");
endpoint.setPayloadExpression(payloadExpression);
endpoint.setOutputChannel(inputChannel);
endpoint.setBeanFactory(mock(BeanFactory.class));
endpoint.afterPropertiesSet();
endpoint.start();
XmlPullParser xmlPullParser = PacketParserUtils.newXmppParser(new StringReader(smackMessage.toString()));
xmlPullParser.next();
testXMPPConnection.parseAndProcessStanza(xmlPullParser);
org.springframework.messaging.Message<?> receive = inputChannel.receive(10000);
assertNotNull(receive);
assertEquals(data, receive.getPayload());
endpoint.stop();
}
private static class TestXMPPConnection extends XMPPTCPConnection {
private TestXMPPConnection() {
super(null);
}
@Override
protected void parseAndProcessStanza(XmlPullParser parser) throws Exception {
super.parseAndProcessStanza(parser);
}
}
}