From be4fd62cea1ba2c51e6376563a719657860ac433 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 19 Nov 2015 21:36:14 -0500 Subject: [PATCH] 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. Conflicts: spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java Resolved. --- .../udp/UnicastReceivingChannelAdapter.java | 57 +++++++++---- .../ip/udp/UdpChannelAdapterTests.java | 85 ++++++++++++++++++- 2 files changed, 120 insertions(+), 22 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 7cb919749c..ac3f993e9d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/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; @@ -133,29 +135,47 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece } protected boolean asyncSendMessage(final DatagramPacket packet) { - this.getTaskExecutor().execute(new Runnable(){ - @Override - public void run() { - Message message = null; - try { - message = mapper.toMessage(packet); - if (logger.isDebugEnabled()) { - logger.debug("Received:" + message); + Executor taskExecutor = getTaskExecutor(); + if (taskExecutor != null) { + try { + taskExecutor.execute(new Runnable() { + + @Override + public void run() { + doSend(packet); + } + }); + } + catch (RejectedExecutionException e) { + if (logger.isDebugEnabled()) { + logger.debug("Adapter stopped, sending on main thread"); } - 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); - } - }}); + doSend(packet); + } + } return true; } + protected void doSend(final DatagramPacket packet) { + Message 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()]; @@ -237,4 +257,5 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece public String getComponentType(){ return "ip:udp-inbound-channel-adapter"; } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index c1e7e11b22..91eb13c656 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 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 java.io.IOException; import java.net.DatagramPacket; @@ -25,8 +27,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; @@ -46,17 +51,82 @@ import org.springframework.integration.test.util.SocketUtils; /** * * @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 exceptionHolder = new AtomicReference(); + 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(); @@ -69,9 +139,16 @@ public class UdpChannelAdapterTests { DatagramSocket datagramSocket = new DatagramSocket(SocketUtils.findAvailableUdpSocket()); datagramSocket.send(packet); datagramSocket.close(); - Message receivedMessage = (Message) channel.receive(2000); + @SuppressWarnings("unchecked") + Message receivedMessage = (Message) 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")