From 353b61160ff10319961020b9442f01ecf330d1f5 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 31 Aug 2010 22:56:05 +0000 Subject: [PATCH] INT-1263, refactored MessageHandlers in several modules to subclass from AbstractMessageHandler to support MessageHistory --- .../gateway/AbstractMessagingGateway.java | 16 +++++- .../gateway/SimpleMessagingGateway.java | 17 +----- .../MessageHistoryIntegrationTests.java | 19 ------- .../messageHistoryWithHistoryWriter.xml | 10 +++- ...ssageHistoryWithHistoryWriterNamespace.xml | 10 +++- ...InternetProtocolSendingMessageHandler.java | 4 +- .../ip/tcp/TcpSendingMessageHandler.java | 7 +-- .../ip/udp/UnicastSendingMessageHandler.java | 9 +-- .../integration/jdbc/JdbcMessageHandler.java | 6 +- .../ChannelPublishingJmsMessageListener.java | 2 - .../jms/config/JmsMessageHistoryTests.java | 57 ------------------- .../config/MessageHistoryTests-context.xml | 6 +- .../mail/MailSendingMessageHandler.java | 5 +- .../ByteStreamWritingMessageHandler.java | 5 +- .../CharacterStreamWritingMessageHandler.java | 6 +- .../XmppRosterEventMessageSendingHandler.java | 30 +++++----- 16 files changed, 74 insertions(+), 135 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java index 82d70dce74..175403a7a6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java @@ -28,6 +28,8 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.BridgeHandler; +import org.springframework.integration.history.MessageHistory; +import org.springframework.integration.history.TrackableComponent; import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.support.MessageBuilder; @@ -42,10 +44,11 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public abstract class AbstractMessagingGateway extends AbstractEndpoint { +public abstract class AbstractMessagingGateway extends AbstractEndpoint implements TrackableComponent{ private static final long DEFAULT_TIMEOUT = 1000L; + private volatile boolean shouldTrack = false; private volatile InboundMessageMapper exceptionMapper; @@ -75,6 +78,10 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { public String getComponentType(){ return "gateway"; } + + public void setShouldTrack(boolean shouldTrack) { + this.shouldTrack = shouldTrack; + } /** * Set the request channel. @@ -152,6 +159,10 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { Assert.state(this.requestChannel != null, "send is not supported, because no request channel has been configured"); Message message = this.toMessage(object); + if (this.shouldTrack) { + message = MessageHistory.write(message, this); + } + Assert.notNull(message, "message must not be null"); this.messagingTemplate.send(this.requestChannel, message); } @@ -182,6 +193,9 @@ public abstract class AbstractMessagingGateway extends AbstractEndpoint { Object sendAndReceive(Object object, boolean shouldMapMessage) { Message request = this.toMessage(object); + if (this.shouldTrack) { + request = MessageHistory.write(request, this); + } Message reply = this.sendAndReceiveMessage(request); if (!shouldMapMessage) { return reply; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index 63f1f174aa..74d6770ef0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -18,8 +18,6 @@ package org.springframework.integration.gateway; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; -import org.springframework.integration.history.MessageHistory; -import org.springframework.integration.history.TrackableComponent; import org.springframework.integration.mapping.InboundMessageMapper; import org.springframework.integration.mapping.OutboundMessageMapper; import org.springframework.util.Assert; @@ -34,16 +32,13 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -@SuppressWarnings({"unchecked", "rawtypes"}) -public class SimpleMessagingGateway extends AbstractMessagingGateway implements TrackableComponent { +@SuppressWarnings({"unchecked"}) +public class SimpleMessagingGateway extends AbstractMessagingGateway { private final InboundMessageMapper inboundMapper; private final OutboundMessageMapper outboundMapper; - private volatile boolean shouldTrack = false; - - public SimpleMessagingGateway() { SimpleMessageMapper mapper = new SimpleMessageMapper(); this.inboundMapper = mapper; @@ -57,11 +52,6 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway implements this.outboundMapper = outboundMapper; } - - public void setShouldTrack(boolean shouldTrack) { - this.shouldTrack = shouldTrack; - } - public Message sendAndReceiveMessage(Object object) { return (Message) super.sendAndReceive(object, false); } @@ -89,9 +79,6 @@ public class SimpleMessagingGateway extends AbstractMessagingGateway implements Message message = null; try { message = this.inboundMapper.toMessage(object); - if (this.shouldTrack) { - message = MessageHistory.write(message, this); - } } catch (Exception e) { if (e instanceof RuntimeException) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java index 81882c765d..623e585939 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/history/MessageHistoryIntegrationTests.java @@ -20,11 +20,9 @@ import java.util.Iterator; import java.util.Map; import java.util.Properties; -import org.junit.Ignore; import org.junit.Test; import org.mockito.Mockito; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; @@ -45,17 +43,6 @@ import static org.junit.Assert.assertNull; */ public class MessageHistoryIntegrationTests { - @Test @Ignore - public void testHistoryAwareMessageHandler() { - ApplicationContext ac = new ClassPathXmlApplicationContext("messageHistoryWithHistoryWriter.xml", MessageHistoryIntegrationTests.class); - Map cefBeans = ac.getBeansOfType(ConsumerEndpointFactoryBean.class); - for (ConsumerEndpointFactoryBean cefBean : cefBeans.values()) { - DirectFieldAccessor bridgeAccessor = new DirectFieldAccessor(cefBean); - String handlerClassName = bridgeAccessor.getPropertyValue("handler").getClass().getName(); - assertEquals("org.springframework.integration.config.MessageHistoryWritingMessageHandler", handlerClassName); - } - } - @Test public void testNoHistoryAwareMessageHandler() { ApplicationContext ac = new ClassPathXmlApplicationContext("messageHistoryWithoutHistoryWriter.xml", MessageHistoryIntegrationTests.class); @@ -207,12 +194,6 @@ public class MessageHistoryIntegrationTests { new ClassPathXmlApplicationContext("messageHistoryWithHistoryWriterNamespace-fail.xml", MessageHistoryIntegrationTests.class); } - @Test(expected=BeanCreationException.class) @Ignore - public void testMessageHistoryMoreThanOneFail() { - new ClassPathXmlApplicationContext("messageHistoryWithHistoryWriter-fail.xml", MessageHistoryIntegrationTests.class); - } - - public static interface SampleGateway { public Message echo(String value); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriter.xml b/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriter.xml index ac44330c3b..a643a0f1cb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriter.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriter.xml @@ -9,7 +9,15 @@ service-interface="org.springframework.integration.history.MessageHistoryIntegrationTests.SampleGateway" default-request-channel="bridgeInChannel"/> - + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriterNamespace.xml b/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriterNamespace.xml index bf589cfe11..7f229c3c5c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriterNamespace.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/history/messageHistoryWithHistoryWriterNamespace.xml @@ -9,7 +9,15 @@ service-interface="org.springframework.integration.history.MessageHistoryIntegrationTests.SampleGateway" default-request-channel="bridgeInChannel"/> - + + + + + + + + + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index e23e60027f..16189c5e8e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -23,7 +23,7 @@ import java.net.SocketAddress; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.util.Assert; /** @@ -32,7 +32,7 @@ import org.springframework.util.Assert; * @author Gary Russell * @since 2.0 */ -public abstract class AbstractInternetProtocolSendingMessageHandler implements MessageHandler, CommonSocketOptions { +public abstract class AbstractInternetProtocolSendingMessageHandler extends AbstractMessageHandler implements CommonSocketOptions { protected final Log logger = LogFactory.getLog(getClass()); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java index 85a1e5ca92..dfeca5c243 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -25,7 +25,7 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageRejectedException; -import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.ip.IpHeaders; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; @@ -43,7 +43,7 @@ import org.springframework.integration.mapping.MessageMappingException; * @since 2.0 * */ -public class TcpSendingMessageHandler implements MessageHandler, TcpSender { +public class TcpSendingMessageHandler extends AbstractMessageHandler implements TcpSender { protected Log logger = LogFactory.getLog(this.getClass()); @@ -78,7 +78,7 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender { * message format. * @see org.springframework.integration.core.MessageHandler#handleMessage(org.springframework.integration.Message) */ - public void handleMessage(final Message message) throws MessageRejectedException, + public void handleMessageInternal(final Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { if (this.serverConnectionFactory != null) { // We don't own the connection @@ -161,5 +161,4 @@ public class TcpSendingMessageHandler implements MessageHandler, TcpSender { public void removeDeadConnection(TcpConnection connection) { connections.remove(connection.getConnectionId()); } - } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index 973a02b99d..63d88256a8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -32,7 +32,6 @@ import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; -import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; @@ -54,7 +53,7 @@ import org.springframework.util.Assert; * @since 2.0 */ public class UnicastSendingMessageHandler extends - AbstractInternetProtocolSendingMessageHandler implements Runnable, InitializingBean { + AbstractInternetProtocolSendingMessageHandler implements Runnable{ protected final DatagramPacketMessageMapper mapper = new DatagramPacketMessageMapper(); @@ -170,7 +169,7 @@ public class UnicastSendingMessageHandler extends } } - public void afterPropertiesSet() { + public void onInit() { if (this.acknowledge) { if (this.taskExecutor == null) { Executor executor = Executors @@ -188,7 +187,7 @@ public class UnicastSendingMessageHandler extends } } - public void handleMessage(Message message) + public void handleMessageInternal(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { if (this.acknowledge) { @@ -343,6 +342,4 @@ public class UnicastSendingMessageHandler extends public void setTaskExecutor(Executor taskExecutor) { this.taskExecutor = taskExecutor; } - - } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java index ada4e7b9e5..3661d354f8 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageHandler.java @@ -19,7 +19,7 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageRejectedException; -import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.core.namedparam.SqlParameterSource; import org.springframework.jdbc.core.simple.SimpleJdbcOperations; @@ -40,7 +40,7 @@ import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; * @author Dave Syer * @since 2.0 */ -public class JdbcMessageHandler implements MessageHandler { +public class JdbcMessageHandler extends AbstractMessageHandler { private final SimpleJdbcOperations jdbcOperations; @@ -83,7 +83,7 @@ public class JdbcMessageHandler implements MessageHandler { /** * Executes the update, passing the message into the {@link SqlParameterSourceFactory}. */ - public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, + protected void handleMessageInternal(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { executeUpdateQuery(message); } diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index 143af67038..04a5f4d236 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -28,7 +28,6 @@ import javax.jms.Session; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.Message; import org.springframework.integration.gateway.AbstractMessagingGateway; -import org.springframework.integration.history.MessageHistory; import org.springframework.integration.support.MessageBuilder; import org.springframework.jms.listener.SessionAwareMessageListener; import org.springframework.jms.support.converter.MessageConverter; @@ -223,7 +222,6 @@ public class ChannelPublishingJmsMessageListener extends AbstractMessagingGatewa Message requestMessage = (object instanceof Message) ? MessageBuilder.fromMessage((Message) object).copyHeaders(headers).build() : MessageBuilder.withPayload(object).copyHeaders(headers).build(); - requestMessage = MessageHistory.write(requestMessage, this); if (!this.expectReply) { this.send(requestMessage); } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageHistoryTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageHistoryTests.java index 4388f35674..8cf0ee786f 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageHistoryTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsMessageHistoryTests.java @@ -68,63 +68,6 @@ public class JmsMessageHistoryTests { assertEquals("jmsInputChannel", event.getProperty(MessageHistory.NAME_PROPERTY)); } - @Test @Ignore - public void testWithHeaderMapperPropagatingOutboundHistory() throws Exception{ - ActiveMqTestUtils.prepare(); - ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("MessageHistoryTests-withHeaderMapper.xml", JmsMessageHistoryTests.class); - DirectChannel input = applicationContext.getBean("outbound-channel", DirectChannel.class); - PollableChannel jmsInputChannel = applicationContext.getBean("jmsInputChannel", PollableChannel.class); - input.send(new GenericMessage("hello")); - Message message = jmsInputChannel.receive(50000); - Iterator historyIterator = message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); - Properties event = historyIterator.next(); - assertEquals("channel", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("outbound-channel", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("jms:outbound-channel-adapter", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("jmsOutbound", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("jms:inbound-channel-adapter", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("sampleJmsInboundAdapter", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("channel", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("jmsInputChannel", event.getProperty(MessageHistory.NAME_PROPERTY)); - } - - @Test @Ignore - public void testWithHeaderMapperPropagatingOutboundHistoryWithGateways() throws Exception{ - ActiveMqTestUtils.prepare(); - ConfigurableApplicationContext applicationContext = new ClassPathXmlApplicationContext("MessageHistoryTests-gateways.xml", JmsMessageHistoryTests.class); - SampleGateway gateway = applicationContext.getBean("sampleGateway", SampleGateway.class); - SubscribableChannel inboundJmsChannel = applicationContext.getBean("inbound-jms-channel", SubscribableChannel.class); - MessageHandler handler = new MessageHandler() { - public void handleMessage(Message message) { - Iterator historyIterator = message.getHeaders().get(MessageHistory.HEADER_NAME, MessageHistory.class).iterator(); - Properties event = historyIterator.next(); - assertEquals("gateway", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("sampleGateway", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("publish-subscribe-channel", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("channel-a", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("jms:outbound-gateway", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("jmsOutbound", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("jms:inbound-gateway", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("jmsInbound", event.getProperty(MessageHistory.NAME_PROPERTY)); - event = historyIterator.next(); - assertEquals("publish-subscribe-channel", event.getProperty(MessageHistory.TYPE_PROPERTY)); - assertEquals("inbound-jms-channel", event.getProperty(MessageHistory.NAME_PROPERTY)); - - MessageChannel channel = (MessageChannel) message.getHeaders().getReplyChannel(); - channel.send(new GenericMessage("OK")); - } - }; - handler = Mockito.spy(handler); - inboundJmsChannel.subscribe(handler); - gateway.echo("hello"); - Mockito.verify(handler, Mockito.times(1)).handleMessage(Mockito.any(Message.class)); - } public static interface SampleGateway { diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-context.xml index feb8a28483..7d675e0202 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-context.xml +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/MessageHistoryTests-context.xml @@ -19,8 +19,12 @@ + + - + + + diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailSendingMessageHandler.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailSendingMessageHandler.java index 3f5a5cc29e..1915378a4e 100644 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailSendingMessageHandler.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/MailSendingMessageHandler.java @@ -24,6 +24,7 @@ import org.springframework.integration.Message; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageHeaders; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.mapping.MessageMappingException; import org.springframework.mail.MailMessage; import org.springframework.mail.SimpleMailMessage; @@ -49,7 +50,7 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Oleg Zhurakousky */ -public class MailSendingMessageHandler implements MessageHandler { +public class MailSendingMessageHandler extends AbstractMessageHandler { private final JavaMailSender mailSender; @@ -66,7 +67,7 @@ public class MailSendingMessageHandler implements MessageHandler { } - public final void handleMessage(Message message) { + protected final void handleMessageInternal(Message message) { MailMessage mailMessage = this.convertMessageToMailMessage(message); if (mailMessage instanceof SimpleMailMessage) { this.mailSender.send((SimpleMailMessage) mailMessage); diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamWritingMessageHandler.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamWritingMessageHandler.java index 93f8254463..6a5845bd0c 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamWritingMessageHandler.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/ByteStreamWritingMessageHandler.java @@ -26,13 +26,14 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.Message; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; /** * A {@link MessageHandler} that writes a byte array to an {@link OutputStream}. * * @author Mark Fisher */ -public class ByteStreamWritingMessageHandler implements MessageHandler { +public class ByteStreamWritingMessageHandler extends AbstractMessageHandler { private final Log logger = LogFactory.getLog(this.getClass()); @@ -53,7 +54,7 @@ public class ByteStreamWritingMessageHandler implements MessageHandler { } - public void handleMessage(Message message) { + protected void handleMessageInternal(Message message) { Object payload = message.getPayload(); if (payload == null) { if (logger.isWarnEnabled()) { diff --git a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandler.java b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandler.java index f9046d22ca..70755caee1 100644 --- a/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandler.java +++ b/spring-integration-stream/src/main/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandler.java @@ -26,10 +26,10 @@ import java.io.Writer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.integration.Message; import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.util.Assert; /** @@ -41,7 +41,7 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class CharacterStreamWritingMessageHandler implements MessageHandler { +public class CharacterStreamWritingMessageHandler extends AbstractMessageHandler { private final Log logger = LogFactory.getLog(this.getClass()); @@ -117,7 +117,7 @@ public class CharacterStreamWritingMessageHandler implements MessageHandler { this.shouldAppendNewLine = shouldAppendNewLine; } - public void handleMessage(Message message) { + protected void handleMessageInternal(Message message) { Object payload = message.getPayload(); if (payload == null) { if (logger.isWarnEnabled()) { diff --git a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageSendingHandler.java b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageSendingHandler.java index b18428762d..2cb6f7aeec 100644 --- a/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageSendingHandler.java +++ b/spring-integration-xmpp/src/main/java/org/springframework/integration/xmpp/presence/XmppRosterEventMessageSendingHandler.java @@ -6,10 +6,7 @@ import org.jivesoftware.smack.XMPPConnection; import org.jivesoftware.smack.packet.Presence; import org.springframework.context.Lifecycle; import org.springframework.integration.Message; -import org.springframework.integration.MessageDeliveryException; -import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.MessageRejectedException; -import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.mapping.OutboundMessageMapper; /** @@ -17,13 +14,14 @@ import org.springframework.integration.mapping.OutboundMessageMapper; * {@link org.jivesoftware.smack.Roster#setSubscriptionMode(org.jivesoftware.smack.Roster.SubscriptionMode)} property. * * @author Josh Long + * @author Oleg Zhurakousky * @see org.jivesoftware.smack.packet.Presence.Mode the mode (i.e.: * {@link org.jivesoftware.smack.packet.Presence.Mode#away}) * @see org.jivesoftware.smack.packet.Presence.Type the type (i.e.: * {@link org.jivesoftware.smack.packet.Presence.Type#available} ) * @since 2.0 */ -public class XmppRosterEventMessageSendingHandler implements MessageHandler, Lifecycle { +public class XmppRosterEventMessageSendingHandler extends AbstractMessageHandler implements Lifecycle { private static final Log logger = LogFactory.getLog(XmppRosterEventMessageDrivenEndpoint.class); private volatile boolean running; @@ -36,17 +34,6 @@ public class XmppRosterEventMessageSendingHandler implements MessageHandler, Lif this.xmppConnection = xmppConnection; } - public void handleMessage(final Message message) throws MessageRejectedException, MessageHandlingException, - MessageDeliveryException { - try { - Presence presence = this.messageMapper.fromMessage(message); - this.xmppConnection.sendPacket(presence); - } - catch (Exception e) { - logger.error("Failed to map packet to message ", e); - } - } - public boolean isRunning() { return this.running; } @@ -80,4 +67,15 @@ public class XmppRosterEventMessageSendingHandler implements MessageHandler, Lif public void setMessageMapper(OutboundMessageMapper messageMapper) { this.messageMapper = messageMapper; } + + @Override + protected void handleMessageInternal(Message message) throws Exception { + try { + Presence presence = this.messageMapper.fromMessage(message); + this.xmppConnection.sendPacket(presence); + } + catch (Exception e) { + logger.error("Failed to map packet to message ", e); + } + } }