1+ // Copyright (c) 2025 Broadcom. All Rights Reserved.
2+ // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
3+ //
4+ // This software, the RabbitMQ Java client library, is triple-licensed under the
5+ // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
6+ // ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
7+ // LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2. For the ASL,
8+ // please see LICENSE-APACHE2.
9+ //
10+ // This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
11+ // either express or implied. See the LICENSE file for specific language governing
12+ // rights and limitations of this software.
13+ //
14+ // If you have any questions regarding licensing, please contact us at
15+ 116package com .rabbitmq .client .test ;
217
18+ import static com .rabbitmq .client .test .TestUtils .LatchConditions .completed ;
19+ import static org .assertj .core .api .Assertions .assertThat ;
20+
321import com .rabbitmq .client .AMQP ;
422import com .rabbitmq .client .Channel ;
5- import com .rabbitmq .client .ConfirmCallback ;
6- import com .rabbitmq .client .ConfirmListener ;
723import com .rabbitmq .client .Connection ;
824import com .rabbitmq .client .ConnectionFactory ;
925import com .rabbitmq .client .DefaultConsumer ;
1026import com .rabbitmq .client .Envelope ;
11- import org .assertj .core .api .Assertions ;
12- import org .junit .jupiter .api .BeforeEach ;
13- import org .junit .jupiter .api .Test ;
14-
27+ import io .netty .channel .EventLoopGroup ;
28+ import io .netty .channel .IoHandlerFactory ;
29+ import io .netty .channel .MultiThreadIoEventLoopGroup ;
30+ import io .netty .channel .epoll .EpollIoHandler ;
31+ import io .netty .channel .epoll .EpollSocketChannel ;
32+ import io .netty .channel .kqueue .KQueueIoHandler ;
33+ import io .netty .channel .kqueue .KQueueSocketChannel ;
1534import java .io .IOException ;
1635import java .nio .charset .StandardCharsets ;
36+ import java .util .Set ;
37+ import java .util .concurrent .ConcurrentHashMap ;
1738import java .util .concurrent .CountDownLatch ;
18-
19- import static com .rabbitmq .client .test .TestUtils .LatchConditions .completed ;
20- import static org .assertj .core .api .Assertions .assertThat ;
39+ import java .util .concurrent .TimeUnit ;
40+ import java .util .concurrent .TimeoutException ;
41+ import org .junit .jupiter .api .BeforeEach ;
42+ import org .junit .jupiter .api .Test ;
43+ import org .junit .jupiter .api .condition .EnabledIfSystemProperty ;
44+ import org .junit .jupiter .api .condition .EnabledOnOs ;
45+ import org .junit .jupiter .api .condition .OS ;
2146
2247public class NettyTest {
2348
@@ -26,34 +51,79 @@ public class NettyTest {
2651 @ BeforeEach
2752 void init () {
2853 cf = TestUtils .connectionFactory ();
54+ cf .useNetty ();
55+ }
56+
57+ @ Test
58+ void publishConsumeDefaults () throws Exception {
59+ publishConsume (cf );
60+ }
61+
62+ @ Test
63+ @ EnabledOnOs (OS .MAC )
64+ @ EnabledIfSystemProperty (named = "os.arch" , matches = "aarch64" )
65+ void kqueue () throws Exception {
66+ nativeIoTest (KQueueIoHandler .newFactory (), KQueueSocketChannel .class );
2967 }
3068
3169 @ Test
32- void test () throws Exception {
70+ @ EnabledOnOs (OS .LINUX )
71+ @ EnabledIfSystemProperty (named = "os.arch" , matches = "amd64" )
72+ void epoll () throws Exception {
73+ nativeIoTest (EpollIoHandler .newFactory (), EpollSocketChannel .class );
74+ }
75+
76+ private static void nativeIoTest (
77+ IoHandlerFactory ioHandlerFactory , Class <? extends io .netty .channel .Channel > channelClass )
78+ throws IOException , TimeoutException {
79+ ConnectionFactory cf = TestUtils .connectionFactory ();
80+ EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup (ioHandlerFactory );
81+ Set <io .netty .channel .Channel > channels = ConcurrentHashMap .newKeySet ();
82+ cf .netty ()
83+ .eventLoopGroup (eventLoopGroup )
84+ .bootstrapCustomizer (b -> b .channel (channelClass ))
85+ .channelCustomizer (channels ::add );
86+ try {
87+ publishConsume (cf );
88+ assertThat (channels )
89+ .isNotEmpty ()
90+ .allMatch (ch -> ch .getClass ().isAssignableFrom (channelClass ));
91+ } finally {
92+ eventLoopGroup .shutdownGracefully (0 , 0 , TimeUnit .SECONDS );
93+ }
94+ }
95+
96+ private static void publishConsume (ConnectionFactory cf ) throws IOException , TimeoutException {
3397 try (Connection c = cf .newConnection ()) {
3498 Channel ch1 = c .createChannel ();
3599 String q = ch1 .queueDeclare ().getQueue ();
36100
37101 Channel ch2 = c .createChannel ();
38102 CountDownLatch consumeLatch = new CountDownLatch (1 );
39103 CountDownLatch cancelLatch = new CountDownLatch (1 );
40- String ctag = ch2 .basicConsume (q , new DefaultConsumer (ch2 ) {
41- @ Override
42- public void handleDelivery (String ctg , Envelope envelope , AMQP .BasicProperties properties , byte [] body ) throws IOException {
43- ch2 .basicAck (envelope .getDeliveryTag (), false );
44- consumeLatch .countDown ();
45- }
46-
47- @ Override
48- public void handleCancelOk (String consumerTag ) {
49- cancelLatch .countDown ();
50- }
51- });
104+ String ctag =
105+ ch2 .basicConsume (
106+ q ,
107+ new DefaultConsumer (ch2 ) {
108+ @ Override
109+ public void handleDelivery (
110+ String ctg , Envelope envelope , AMQP .BasicProperties properties , byte [] body )
111+ throws IOException {
112+ ch2 .basicAck (envelope .getDeliveryTag (), false );
113+ consumeLatch .countDown ();
114+ }
115+
116+ @ Override
117+ public void handleCancelOk (String consumerTag ) {
118+ cancelLatch .countDown ();
119+ }
120+ });
52121
53122 Channel ch3 = c .createChannel ();
54123 ch3 .confirmSelect ();
55124 CountDownLatch confirmLatch = new CountDownLatch (1 );
56- ch3 .addConfirmListener ((deliveryTag , multiple ) -> confirmLatch .countDown (), (dtag , multiple ) -> {});
125+ ch3 .addConfirmListener (
126+ (deliveryTag , multiple ) -> confirmLatch .countDown (), (dtag , multiple ) -> {});
57127 ch3 .basicPublish ("" , q , null , "hello" .getBytes (StandardCharsets .UTF_8 ));
58128 assertThat (confirmLatch ).is (completed ());
59129 assertThat (consumeLatch ).is (completed ());
@@ -70,5 +140,4 @@ public void handleCancelOk(String consumerTag) {
70140 ch1 .close ();
71141 }
72142 }
73-
74143}
0 commit comments