diff --git a/.github/workflows/pr-build-workflow.yml b/.github/workflows/pr-build-workflow.yml index 49d2d00e..bda4013e 100644 --- a/.github/workflows/pr-build-workflow.yml +++ b/.github/workflows/pr-build-workflow.yml @@ -30,8 +30,9 @@ jobs: arguments: check - name: Capture Test Results + if: failure() uses: actions/upload-artifact@v2 with: name: test-results - path: spring-rabbit/build/reports/tests + path: '*/build/reports/tests/**/*.*' retention-days: 3 diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java index b52092e3..ede632de 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/ConnectionFactoryUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -147,7 +147,8 @@ public final class ConnectionFactoryUtils { } resourceHolderToUse.addChannel(channel, connection); - if (!resourceHolderToUse.equals(resourceHolder)) { + if (!resourceHolderToUse.equals(resourceHolder) + && TransactionSynchronizationManager.isSynchronizationActive()) { bindResourceToTransaction(resourceHolderToUse, connectionFactory, resourceFactory.isSynchedLocalTransactionAllowed()); } @@ -171,6 +172,7 @@ public final class ConnectionFactoryUtils { public static RabbitResourceHolder bindResourceToTransaction(RabbitResourceHolder resourceHolder, ConnectionFactory connectionFactory, boolean synched) { + if (TransactionSynchronizationManager.hasResource(connectionFactory) || !TransactionSynchronizationManager.isActualTransactionActive() || !synched) { return (RabbitResourceHolder) TransactionSynchronizationManager.getResource(connectionFactory); // NOSONAR never null diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java index fafe709d..629bfa4b 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/ClientRecoveryCompatibilityTests.java @@ -17,7 +17,7 @@ package org.springframework.amqp.rabbit.connection; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.BDDMockito.given; @@ -43,6 +43,7 @@ public class ClientRecoveryCompatibilityTests { @Test public void testDefeatRecovery() throws Exception { + RabbitUtils.clearPhysicalCloseRequired(); // left over from some other test final Channel channel1 = mock(Channel.class); given(channel1.isOpen()).willReturn(true); final Channel channel2 = mock(Channel.class); @@ -57,37 +58,32 @@ public class ClientRecoveryCompatibilityTests { ccf.setExecutor(mock(ExecutorService.class)); Connection conn1 = ccf.createConnection(); Channel channel = conn1.createChannel(false); - verifyChannelIs(channel1, channel); + ChannelProxy proxy = (ChannelProxy) channel; + assertThat(proxy.getTargetChannel()).isSameAs(channel1); channel.close(); conn1.close(); Connection conn2 = ccf.createConnection(); assertThat(conn2).isSameAs(conn1); channel = conn1.createChannel(false); - verifyChannelIs(channel1, channel); + proxy = (ChannelProxy) channel; + assertThat(proxy.getTargetChannel()).isSameAs(channel1); channel.close(); conn2.close(); given(rabbitConn.isOpen()).willReturn(false).willReturn(true); given(channel1.isOpen()).willReturn(false); - conn2 = ccf.createConnection(); - try { - conn2.createChannel(false); - fail("Expected AutoRecoverConnectionNotCurrentlyOpenException"); - } - catch (AutoRecoverConnectionNotCurrentlyOpenException e) { - assertThat(e.getMessage()).isEqualTo("Auto recovery connection is not currently open"); - } + Connection conn3 = ccf.createConnection(); + assertThat(conn3).isSameAs(conn1); + assertThatExceptionOfType(AutoRecoverConnectionNotCurrentlyOpenException.class).isThrownBy(() -> + conn3.createChannel(false)) + .withMessage("Auto recovery connection is not currently open"); channel = conn2.createChannel(false); - verifyChannelIs(channel2, channel); + proxy = (ChannelProxy) channel; + assertThat(proxy.getTargetChannel()).isSameAs(channel2); channel.close(); verify(rabbitConn, never()).close(); verify(channel1).close(); // physically closed to defeat recovery } - private void verifyChannelIs(Channel mockChannel, Channel channel) { - ChannelProxy proxy = (ChannelProxy) channel; - assertThat(proxy.getTargetChannel()).isSameAs(mockChannel); - } - } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/TransactionalEventListenerTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/TransactionalEventListenerTests.java new file mode 100644 index 00000000..f7f6586a --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/TransactionalEventListenerTests.java @@ -0,0 +1,141 @@ +/* + * Copyright 2021 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import java.io.IOException; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.rabbit.connection.Connection; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.TransactionException; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.event.TransactionPhase; +import org.springframework.transaction.event.TransactionalEventListener; +import org.springframework.transaction.support.AbstractPlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionStatus; + +import com.rabbitmq.client.Channel; + +/** + * @author Gary Russell + * @since 2.2.16 + * + */ +@SpringJUnitConfig +@DirtiesContext +public class TransactionalEventListenerTests { + + @Test + void txCommits(@Autowired Config config, @Autowired AtomicBoolean committed, + @Autowired Channel channel) throws IOException { + + config.publish(); + assertThat(committed.get()).isTrue(); + verify(channel).txCommit(); + } + + @Configuration(proxyBeanMethods = false) + @EnableTransactionManagement + public static class Config { + + @Autowired + ApplicationEventPublisher publisher; + + @Autowired + RabbitTemplate template; + + @Bean + AtomicBoolean committed() { + return new AtomicBoolean(); + } + + @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT) + public void el(String in) { + this.template.convertAndSend("test"); + } + + @Bean + RabbitTemplate template(ConnectionFactory cf) { + RabbitTemplate template = new RabbitTemplate(cf); + template.setChannelTransacted(true); + return template; + } + + @Bean + Channel channel() { + return mock(Channel.class); + } + + @Bean + ConnectionFactory cf(Channel channel) { + ConnectionFactory cf = mock(ConnectionFactory.class); + Connection conn = mock(Connection.class); + given(conn.isOpen()).willReturn(true); + given(cf.createConnection()).willReturn(conn); + given(conn.createChannel(true)).willReturn(channel); + given(channel.isOpen()).willReturn(true); + return cf; + } + + @Transactional + public void publish() { + publisher.publishEvent("test"); + } + + @Bean + PlatformTransactionManager transactionManager(AtomicBoolean committed) { + return new AbstractPlatformTransactionManager() { + + @Override + protected void doRollback(DefaultTransactionStatus status) throws TransactionException { + } + + @Override + protected Object doGetTransaction() throws TransactionException { + return new Object(); + } + + @Override + protected void doCommit(DefaultTransactionStatus status) throws TransactionException { + committed.set(true); + } + + @Override + protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionException { + } + }; + } + + } + +}