From d6776b4889be17940b293f68d93a8fd458cf0fcf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 7 Aug 2013 13:52:52 -0400 Subject: [PATCH] INT-3107 Fix TCP Syslog ApplicationEventPublisher No publisher was injected if the default embedded connection factory was used. --- ...logReceivingChannelAdapterFactoryBean.java | 14 ++++++++++- .../TcpSyslogReceivingChannelAdapter.java | 14 ++++++++++- ...logReceivingChannelAdapterParserTests.java | 2 ++ .../SyslogReceivingChannelAdapterTests.java | 23 +++++++++++++++++++ 4 files changed, 51 insertions(+), 2 deletions(-) diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java index 96d9ee8d8e..7f9bf13be0 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterFactoryBean.java @@ -17,6 +17,8 @@ package org.springframework.integration.syslog.config; import org.springframework.beans.factory.BeanNameAware; import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.integration.MessageChannel; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; @@ -34,7 +36,7 @@ import org.springframework.util.Assert; * */ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBean - implements SmartLifecycle, BeanNameAware { + implements SmartLifecycle, BeanNameAware, ApplicationEventPublisherAware { public enum Protocol { udp, tcp }; @@ -62,6 +64,8 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea private volatile String beanName; + private volatile ApplicationEventPublisher applicationEventPublisher; + /** * Instantiates a factory bean that creates a {@link UdpSyslogReceivingChannelAdapter} * if the protocol is {@link Protocol#udp} or a {@link TcpSyslogReceivingChannelAdapter} if @@ -109,6 +113,11 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea this.converter = converter; } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + @Override public void start() { if (this.adapter != null) { @@ -168,6 +177,9 @@ public class SyslogReceivingChannelAdapterFactoryBean extends AbstractFactoryBea Assert.isNull(this.port, "Cannot specify both 'port' and 'connectionFactory'"); ((TcpSyslogReceivingChannelAdapter) adapter).setConnectionFactory(this.connectionFactory); } + else if (this.applicationEventPublisher != null) { + ((TcpSyslogReceivingChannelAdapter) adapter).setApplicationEventPublisher(this.applicationEventPublisher); + } Assert.isNull(this.udpAdapter, "Cannot specifiy 'udp-attributes' when the protocol is 'tcp'"); } else if(this.protocol == Protocol.udp) { diff --git a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java index 11f6857f72..5c283d34cf 100644 --- a/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java +++ b/spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java @@ -16,6 +16,8 @@ package org.springframework.integration.syslog.inbound; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpListener; @@ -30,10 +32,12 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer; * */ public class TcpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdapterSupport - implements TcpListener { + implements TcpListener, ApplicationEventPublisherAware { private volatile AbstractServerConnectionFactory connectionFactory; + private volatile ApplicationEventPublisher applicationEventPublisher; + /** * @param connectionFactory */ @@ -41,12 +45,20 @@ public class TcpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap this.connectionFactory = connectionFactory; } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + @Override protected void onInit() { super.onInit(); if (this.connectionFactory == null) { this.connectionFactory = new TcpNioServerConnectionFactory(this.getPort()); this.connectionFactory.setDeserializer(new ByteArrayLfSerializer()); + if (this.applicationEventPublisher != null) { + this.connectionFactory.setApplicationEventPublisher(this.applicationEventPublisher); + } } this.connectionFactory.registerListener(this); } diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java index 9c54bc535f..4ee0bf08c6 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/config/SyslogReceivingChannelAdapterParserTests.java @@ -31,6 +31,7 @@ import javax.net.SocketFactory; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -139,6 +140,7 @@ public class SyslogReceivingChannelAdapterParserTests { Message message = bar.receive(10000); assertNotNull(message); adapter2.stop(); + assertNotNull(TestUtils.getPropertyValue(adapter2, "connectionFactory.applicationEventPublisher")); } @Test diff --git a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java index e44a8d2a3e..a4551aff85 100644 --- a/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java +++ b/spring-integration-syslog/src/test/java/org/springframework/integration/syslog/inbound/SyslogReceivingChannelAdapterTests.java @@ -17,15 +17,26 @@ package org.springframework.integration.syslog.inbound; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; import java.net.DatagramPacket; import java.net.DatagramSocket; import java.net.InetSocketAddress; import java.net.Socket; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import javax.net.SocketFactory; import org.junit.Test; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; @@ -70,6 +81,17 @@ public class SyslogReceivingChannelAdapterTests { factory.setPort(port); PollableChannel outputChannel = new QueueChannel(); factory.setOutputChannel(outputChannel); + ApplicationEventPublisher publisher = mock(ApplicationEventPublisher.class); + final CountDownLatch latch = new CountDownLatch(2); + doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + latch.countDown(); + return null; + } + }).when(publisher).publishEvent(any(ApplicationEvent.class)); + factory.setApplicationEventPublisher(publisher); factory.afterPropertiesSet(); factory.start(); TcpSyslogReceivingChannelAdapter adapter = (TcpSyslogReceivingChannelAdapter) factory.getObject(); @@ -82,6 +104,7 @@ public class SyslogReceivingChannelAdapterTests { assertNotNull(message); assertEquals("WEBERN", message.getHeaders().get("syslog_HOST")); adapter.stop(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); } }