From d38370ed059c339e993588fa4c157daa76755c31 Mon Sep 17 00:00:00 2001 From: Kazuki Shimizu Date: Fri, 5 May 2023 22:32:55 +0900 Subject: [PATCH] GH-8609: Fix TcpConnectorInterceptor chain propagation Fixes https://github.com/spring-projects/spring-integration/issues/8609 * Changed to passed the self instance to `addNewConnection()` instead of argument's connection in a `TcpConnectionInterceptorSupport` to compose a chain of interceptors from top to down. This way the target `Sender` get the last interceptor in a chain as its connection. **Cherry-pick to `6.0.x` & `5.5.x`** --- .../TcpConnectionInterceptorSupport.java | 5 +++-- .../ip/tcp/connection/TcpSenderTests.java | 16 ++++++++++++++-- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index f3d649625e..8bde317cc0 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2022 the original author or authors. + * Copyright 2002-2023 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. @@ -33,6 +33,7 @@ import org.springframework.messaging.support.ErrorMessage; * to the underlying {@link TcpConnection}. * * @author Gary Russell + * @author Kazuki Shimizu * * @since 2.0 */ @@ -232,7 +233,7 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo @Override public void addNewConnection(TcpConnection connection) { if (this.interceptedSenders != null) { - this.interceptedSenders.forEach(sender -> sender.addNewConnection(connection)); + this.interceptedSenders.forEach(sender -> sender.addNewConnection(this)); } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpSenderTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpSenderTests.java index 980a672952..b2d183008b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpSenderTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpSenderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-2023 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. @@ -18,7 +18,9 @@ package org.springframework.integration.ip.tcp.connection; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -30,6 +32,7 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Gary Russell + * @author Kazuki Shimizu * @since 5.3.10 * */ @@ -81,11 +84,14 @@ public class TcpSenderTests { List addOrder = Collections.synchronizedList(new ArrayList<>()); List remOrder = Collections.synchronizedList(new ArrayList<>()); AtomicReference thread = new AtomicReference<>(); + Map interceptorsPerInstance = new HashMap<>(); + List passedConnectionsToSenderViaAddNewConnection = new ArrayList<>(); class InterceptorFactory extends HelloWorldInterceptorFactory { @Override public TcpConnectionInterceptorSupport getInterceptor() { - return new TcpConnectionInterceptorSupport() { + + TcpConnectionInterceptorSupport interceptor = new TcpConnectionInterceptorSupport() { private final int instance = instances.incrementAndGet(); @@ -107,6 +113,8 @@ public class TcpSenderTests { } }; + interceptorsPerInstance.put(instances.get(), interceptor); + return interceptor; } } @@ -118,6 +126,7 @@ public class TcpSenderTests { @Override public void addNewConnection(TcpConnection connection) { addOrder.add(99); + passedConnectionsToSenderViaAddNewConnection.add(connection); adds.countDown(); } @@ -146,6 +155,9 @@ public class TcpSenderTests { assertThat(remOrder).containsExactly(1, 2, 99, 3, 4, 5, 99, 6); assertThat(interceptorAddCalled.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(interceptorRemCalled.await(10, TimeUnit.SECONDS)).isTrue(); + // should be passed the last interceptor to the real sender via addNewConnection method + assertThat(passedConnectionsToSenderViaAddNewConnection.get(0)).isSameAs(interceptorsPerInstance.get(3)); + assertThat(passedConnectionsToSenderViaAddNewConnection.get(1)).isSameAs(interceptorsPerInstance.get(6)); } }