diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index e85f9142e3..c7ba6dd6fc 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -16,7 +16,9 @@ package org.springframework.integration.ip.tcp.connection; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; @@ -27,6 +29,7 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import javax.net.SocketFactory; @@ -40,9 +43,11 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializ import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; /** * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ public class TcpNioConnectionReadTests { @@ -210,14 +215,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -239,6 +246,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendLengthOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Message length 2147483647 exceeds max message length: 2048")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -252,14 +265,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -281,6 +296,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("ETX not found before max message length: 1024")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -294,14 +315,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -323,6 +346,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendCrLfOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("CRLF not found before max message length: 1024")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -331,7 +360,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -341,14 +369,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -368,6 +398,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -375,7 +411,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -385,14 +420,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -413,6 +450,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -420,7 +463,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ @Test @@ -431,10 +473,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupStxEtx() throws Exception { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -443,10 +483,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupLengthHeader() throws Exception { ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); @@ -455,16 +493,19 @@ public class TcpNioConnectionReadTests { private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage) throws Exception { - final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - responses.add(message); - return false; + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } + return false; }, new TcpSender() { @Override public void addNewConnection(TcpConnection connection) { @@ -482,6 +523,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop();