|
77 | 77 |
|
78 | 78 | import com.rabbitmq.client.Address;
|
79 | 79 | import com.rabbitmq.client.Channel;
|
| 80 | +import com.rabbitmq.client.ConfirmListener; |
80 | 81 | import com.rabbitmq.client.ConnectionFactory;
|
81 | 82 | import com.rabbitmq.client.GetResponse;
|
82 | 83 | import com.rabbitmq.client.ShutdownSignalException;
|
@@ -590,35 +591,77 @@ public void testCheckoutLimitWithPublisherConfirmsPhysical() throws IOException,
|
590 | 591 | testCheckoutLimitWithPublisherConfirms(true);
|
591 | 592 | }
|
592 | 593 |
|
593 |
| - public void testCheckoutLimitWithPublisherConfirms(boolean physicalClose) throws IOException, Exception { |
| 594 | + private void testCheckoutLimitWithPublisherConfirms(boolean physicalClose) throws IOException, Exception { |
594 | 595 | com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
|
595 | 596 | com.rabbitmq.client.Connection mockConnection = mock(com.rabbitmq.client.Connection.class);
|
596 |
| - Channel mockChannel1 = mock(Channel.class); |
| 597 | + Channel mockChannel = mock(Channel.class); |
597 | 598 |
|
598 | 599 | when(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).thenReturn(mockConnection);
|
599 |
| - when(mockConnection.createChannel()).thenReturn(mockChannel1); |
| 600 | + when(mockConnection.createChannel()).thenReturn(mockChannel); |
600 | 601 | when(mockConnection.isOpen()).thenReturn(true);
|
601 | 602 |
|
602 | 603 | // Called during physical close
|
603 |
| - when(mockChannel1.isOpen()).thenReturn(true); |
| 604 | + when(mockChannel.isOpen()).thenReturn(true); |
| 605 | + CountDownLatch confirmsLatch = new CountDownLatch(1); |
| 606 | + doAnswer(invoc -> { |
| 607 | + confirmsLatch.await(10, TimeUnit.SECONDS); |
| 608 | + return null; |
| 609 | + }).when(mockChannel).waitForConfirmsOrDie(anyLong()); |
| 610 | + AtomicReference<ConfirmListener> confirmListener = new AtomicReference<>(); |
| 611 | + doAnswer(invoc -> { |
| 612 | + confirmListener.set(invoc.getArgument(0)); |
| 613 | + return null; |
| 614 | + }).when(mockChannel).addConfirmListener(any()); |
| 615 | + when(mockChannel.getNextPublishSeqNo()).thenReturn(1L); |
604 | 616 |
|
605 | 617 | CachingConnectionFactory ccf = new CachingConnectionFactory(mockConnectionFactory);
|
606 |
| - ccf.setExecutor(mock(ExecutorService.class)); |
| 618 | + ExecutorService exec = Executors.newCachedThreadPool(); |
| 619 | + ccf.setExecutor(exec); |
607 | 620 | ccf.setChannelCacheSize(1);
|
608 | 621 | ccf.setChannelCheckoutTimeout(1);
|
609 | 622 | ccf.setPublisherConfirms(true);
|
610 | 623 |
|
611 | 624 | final Connection con = ccf.createConnection();
|
612 | 625 |
|
| 626 | + RabbitTemplate rabbitTemplate = new RabbitTemplate(ccf); |
613 | 627 | if (physicalClose) {
|
614 |
| - Channel channel = con.createChannel(false); |
615 |
| - RabbitUtils.setPhysicalCloseRequired(channel, physicalClose); |
616 |
| - channel.close(); |
| 628 | + Channel channel1 = con.createChannel(false); |
| 629 | + RabbitUtils.setPhysicalCloseRequired(channel1, physicalClose); |
| 630 | + channel1.close(); |
617 | 631 | }
|
618 | 632 | else {
|
619 |
| - new RabbitTemplate(ccf).convertAndSend("foo", "bar"); // pending confirm |
| 633 | + rabbitTemplate.convertAndSend("foo", "bar"); // pending confirm |
620 | 634 | }
|
621 | 635 | assertThatThrownBy(() -> con.createChannel(false)).isInstanceOf(AmqpTimeoutException.class);
|
| 636 | + int n = 0; |
| 637 | + if (physicalClose) { |
| 638 | + confirmsLatch.countDown(); |
| 639 | + Channel channel2 = null; |
| 640 | + while (channel2 == null && n++ < 100) { |
| 641 | + try { |
| 642 | + channel2 = con.createChannel(false); |
| 643 | + } |
| 644 | + catch (Exception e) { |
| 645 | + Thread.sleep(100); |
| 646 | + } |
| 647 | + } |
| 648 | + assertThat(channel2).isNotNull(); |
| 649 | + } |
| 650 | + else { |
| 651 | + confirmListener.get().handleAck(1L, false); |
| 652 | + boolean ok = false; |
| 653 | + while (!ok && n++ < 100) { |
| 654 | + try { |
| 655 | + rabbitTemplate.convertAndSend("foo", "bar"); |
| 656 | + ok = true; |
| 657 | + } |
| 658 | + catch (Exception e) { |
| 659 | + Thread.sleep(100); |
| 660 | + } |
| 661 | + } |
| 662 | + assertThat(ok).isTrue(); |
| 663 | + } |
| 664 | + exec.shutdownNow(); |
622 | 665 | }
|
623 | 666 |
|
624 | 667 | @Test
|
@@ -1485,7 +1528,7 @@ public void testConsumerChannelWithPubConfPhysicallyClosedWhenNotIsOpen() throws
|
1485 | 1528 | testConsumerChannelPhysicallyClosedWhenNotIsOpenGuts(true);
|
1486 | 1529 | }
|
1487 | 1530 |
|
1488 |
| - public void testConsumerChannelPhysicallyClosedWhenNotIsOpenGuts(boolean confirms) throws Exception { |
| 1531 | + private void testConsumerChannelPhysicallyClosedWhenNotIsOpenGuts(boolean confirms) throws Exception { |
1489 | 1532 | ExecutorService executor = Executors.newSingleThreadExecutor();
|
1490 | 1533 | try {
|
1491 | 1534 | com.rabbitmq.client.ConnectionFactory mockConnectionFactory = mock(com.rabbitmq.client.ConnectionFactory.class);
|
|
0 commit comments