INT-3893: UnicastReceivingChannelAdapter: fix NPE

JIRA: https://jira.spring.io/browse/INT-3893

Polishing

Send on the calling thread if the executor is shut down.

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java

@@ -23,6 +23,8 @@ import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -143,29 +145,47 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 	}
 
 	protected boolean asyncSendMessage(final DatagramPacket packet) {
-		this.getTaskExecutor().execute(new Runnable(){
+		Executor taskExecutor = getTaskExecutor();
-			@Override
+		if (taskExecutor != null) {
-			public void run() {
+			try {
-				Message<byte[]> message = null;
+				taskExecutor.execute(new Runnable() {
-				try {
+
-					message = mapper.toMessage(packet);
+					@Override
-					if (logger.isDebugEnabled()) {
+					public void run() {
-						logger.debug("Received:" + message);
+						doSend(packet);
+
 					}
+				});
+			}
+			catch (RejectedExecutionException e) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Adapter stopped, sending on main thread");
 				}
-				catch (Exception e) {
+				doSend(packet);
-					logger.error("Failed to map packet to message ", e);
+			}
-				}
+		}
-				if (message != null) {
-					if (message.getHeaders().containsKey(IpHeaders.ACK_ADDRESS)) {
-						sendAck(message);
-					}
-					sendMessage(message);
-				}
-			}});
 		return true;
 	}
 
+	protected void doSend(final DatagramPacket packet) {
+		Message<byte[]> message = null;
+		try {
+			message = mapper.toMessage(packet);
+			if (logger.isDebugEnabled()) {
+				logger.debug("Received:" + message);
+			}
+		}
+		catch (Exception e) {
+			logger.error("Failed to map packet to message ", e);
+		}
+		if (message != null) {
+			if (message.getHeaders().containsKey(IpHeaders.ACK_ADDRESS)) {
+				sendAck(message);
+			}
+			sendMessage(message);
+		}
+	}
+
 	protected DatagramPacket receive() throws Exception {
 		DatagramSocket socket = this.getSocket();
 		final byte[] buffer = new byte[this.getReceiveBufferSize()];
@@ -247,4 +267,5 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 	public String getComponentType(){
 		return "ip:udp-inbound-channel-adapter";
 	}
+
 }

spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -17,7 +17,9 @@ package org.springframework.integration.ip.udp;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.mock;
 
 import java.io.IOException;
@@ -26,8 +28,11 @@ import java.net.DatagramSocket;
 import java.net.Inet4Address;
 import java.net.InetSocketAddress;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.LogFactory;
@@ -48,17 +53,82 @@ import org.springframework.messaging.SubscribableChannel;
 /**
  *
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.0
  *
  */
 public class UdpChannelAdapterTests {
 
-	@SuppressWarnings("unchecked")
 	@Test
 	public void testUnicastReceiver() throws Exception {
+		testUnicastReceiver(false);
+	}
+
+	@Test
+	public void testUnicastReceiverDeadExecutor() throws Exception {
+		testUnicastReceiver(true);
+	}
+
+	private void testUnicastReceiver(final boolean killExecutor) throws Exception {
 		QueueChannel channel = new QueueChannel(2);
 		int port = SocketUtils.findAvailableUdpSocket();
-		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port);
+		final CountDownLatch stopLatch = new CountDownLatch(1);
+		final CountDownLatch exitLatch = new CountDownLatch(1);
+		final AtomicBoolean stopping = new AtomicBoolean();
+		final AtomicReference<Exception> exceptionHolder = new AtomicReference<Exception>();
+		UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port) {
+
+			@Override
+			public boolean isActive() {
+				if (stopping.get()) {
+					try {
+						stopLatch.await(10, TimeUnit.SECONDS);
+					}
+					catch (InterruptedException e) {
+						fail();
+					}
+					return true;
+				}
+				else {
+					return super.isActive();
+				}
+			}
+
+			@Override
+			protected DatagramPacket receive() throws Exception {
+				if (stopping.get()) {
+					return new DatagramPacket(new byte[0], 0);
+				}
+				else {
+					return super.receive();
+				}
+			}
+
+			@Override
+			protected boolean asyncSendMessage(DatagramPacket packet) {
+				boolean result = false;
+				try {
+					result = super.asyncSendMessage(packet);
+				}
+				catch (Exception e) {
+					exceptionHolder.set(e);
+				}
+				if (stopping.get()) {
+					exitLatch.countDown();
+				}
+				return result;
+			}
+
+			@Override
+			public Executor getTaskExecutor() {
+				Executor taskExecutor = super.getTaskExecutor();
+				if (killExecutor && taskExecutor != null) {
+					((ExecutorService) taskExecutor).shutdown();
+				}
+				return taskExecutor;
+			}
+
+		};
 		adapter.setOutputChannel(channel);
 //		SocketUtils.setLocalNicIfPossible(adapter);
 		adapter.start();
@@ -71,9 +141,16 @@ public class UdpChannelAdapterTests {
 		DatagramSocket datagramSocket = new DatagramSocket(SocketUtils.findAvailableUdpSocket());
 		datagramSocket.send(packet);
 		datagramSocket.close();
-		Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(2000);
+		@SuppressWarnings("unchecked")
+		Message<byte[]> receivedMessage = (Message<byte[]>) channel.receive(10000);
+		assertNotNull(receivedMessage);
 		assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload()));
+		stopping.set(true);
 		adapter.stop();
+		stopLatch.countDown();
+		exitLatch.await(10, TimeUnit.SECONDS);
+		// Previously it failed with NPE
+		assertNull(exceptionHolder.get());
 	}
 
 	@SuppressWarnings("unchecked")