INT-3107 Fix TCP Syslog ApplicationEventPublisher

No publisher was injected if the default embedded connection
factory was used.

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<SyslogReceivingChannelAdapterSupport>
-		 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) {

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);
 	}

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

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<Object>() {
+
+			@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));
 	}
 
 }