Skip to content

Commit 76348cb

Browse files
zysaaagaryrussell
zysaaa
authored and
garyrussell
committed
GH-1338: Add MessageAckListener
Resolves #1338 GH-1338: Test case for MessageAckListener
1 parent bd46d47 commit 76348cb

8 files changed

Lines changed: 341 additions & 12 deletions

File tree

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
3131
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
3232
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
33+
import org.springframework.amqp.rabbit.listener.MessageAckListener;
3334
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
3435
import org.springframework.amqp.support.ConsumerTagStrategy;
3536
import org.springframework.amqp.support.converter.MessageConverter;
@@ -115,6 +116,8 @@ public abstract class AbstractRabbitListenerContainerFactory<C extends AbstractM
115116

116117
private Boolean deBatchingEnabled;
117118

119+
private MessageAckListener messageAckListener;
120+
118121
/**
119122
* @param connectionFactory The connection factory.
120123
* @see AbstractMessageListenerContainer#setConnectionFactory(ConnectionFactory)
@@ -323,6 +326,15 @@ public void setGlobalQos(boolean globalQos) {
323326
this.globalQos = globalQos;
324327
}
325328

329+
/**
330+
* Set a {@link MessageAckListener} to use when ack a message(messages) in {@link AcknowledgeMode#AUTO} mode.
331+
* @param messageAckListener the messageAckListener.
332+
* @see AbstractMessageListenerContainer#setMessageAckListener(MessageAckListener)
333+
*/
334+
public void setMessageAckListener(MessageAckListener messageAckListener) {
335+
this.messageAckListener = messageAckListener;
336+
}
337+
326338
@Override
327339
public C createListenerContainer(RabbitListenerEndpoint endpoint) {
328340
C instance = createContainerInstance();
@@ -355,7 +367,8 @@ public C createListenerContainer(RabbitListenerEndpoint endpoint) {
355367
.acceptIfNotNull(this.autoStartup, instance::setAutoStartup)
356368
.acceptIfNotNull(this.phase, instance::setPhase)
357369
.acceptIfNotNull(this.afterReceivePostProcessors, instance::setAfterReceivePostProcessors)
358-
.acceptIfNotNull(this.deBatchingEnabled, instance::setDeBatchingEnabled);
370+
.acceptIfNotNull(this.deBatchingEnabled, instance::setDeBatchingEnabled)
371+
.acceptIfNotNull(this.messageAckListener, instance::setMessageAckListener);
359372
if (this.batchListener && this.deBatchingEnabled == null) {
360373
// turn off container debatching by default for batch listeners
361374
instance.setDeBatchingEnabled(false);

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -252,6 +252,8 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor
252252

253253
private boolean asyncReplies;
254254

255+
private MessageAckListener messageAckListener;
256+
255257
@Override
256258
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
257259
this.applicationEventPublisher = applicationEventPublisher;
@@ -1188,6 +1190,21 @@ public void setjavaLangErrorHandler(JavaLangErrorHandler javaLangErrorHandler) {
11881190
this.javaLangErrorHandler = javaLangErrorHandler;
11891191
}
11901192

1193+
/**
1194+
* Set a {@link MessageAckListener} to use when ack a message(messages) in {@link AcknowledgeMode#AUTO} mode.
1195+
* @param messageAckListener the messageAckListener.
1196+
* @see MessageAckListener
1197+
* @see AcknowledgeMode
1198+
*/
1199+
public void setMessageAckListener(MessageAckListener messageAckListener) {
1200+
Assert.notNull(messageAckListener, "'messageAckListener' cannot be null");
1201+
this.messageAckListener = messageAckListener;
1202+
}
1203+
1204+
protected MessageAckListener getMessageAckListener() {
1205+
return this.messageAckListener;
1206+
}
1207+
11911208
/**
11921209
* Delegates to {@link #validateConfiguration()} and {@link #initialize()}.
11931210
*/

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@
8787
* @author Alex Panchenko
8888
* @author Johno Crawford
8989
* @author Ian Roberts
90+
* @author Cao Weibo
9091
*/
9192
public class BlockingQueueConsumer {
9293

@@ -836,11 +837,11 @@ public void rollbackOnExceptionIfNecessary(Throwable ex, long tag) {
836837
/**
837838
* Perform a commit or message acknowledgement, as appropriate.
838839
* @param localTx Whether the channel is locally transacted.
840+
* @param messageAckListener MessageAckListener set on the message listener.
839841
* @return true if at least one delivery tag exists.
840842
* @throws IOException Any IOException.
841843
*/
842-
public boolean commitIfNecessary(boolean localTx) throws IOException {
843-
844+
public boolean commitIfNecessary(boolean localTx, MessageAckListener messageAckListener) throws IOException {
844845
if (this.deliveryTags.isEmpty()) {
845846
return false;
846847
}
@@ -857,7 +858,14 @@ public boolean commitIfNecessary(boolean localTx) throws IOException {
857858

858859
if (ackRequired && (!this.transactional || isLocallyTransacted)) {
859860
long deliveryTag = new ArrayList<Long>(this.deliveryTags).get(this.deliveryTags.size() - 1);
860-
this.channel.basicAck(deliveryTag, true);
861+
try {
862+
this.channel.basicAck(deliveryTag, true);
863+
notifyMessageAckListener(messageAckListener, true, deliveryTag, null);
864+
}
865+
catch (Exception e) {
866+
logger.error("Error acking.", e);
867+
notifyMessageAckListener(messageAckListener, false, deliveryTag, e);
868+
}
861869
}
862870

863871
if (isLocallyTransacted) {
@@ -874,6 +882,28 @@ public boolean commitIfNecessary(boolean localTx) throws IOException {
874882

875883
}
876884

885+
/**
886+
* Notify MessageAckListener set on the relevant message listener.
887+
* @param messageAckListener MessageAckListener set on the message listener.
888+
* @param success Whether ack succeeded.
889+
* @param deliveryTag The deliveryTag of ack.
890+
* @param cause If an exception occurs.
891+
*/
892+
private void notifyMessageAckListener(@Nullable MessageAckListener messageAckListener,
893+
boolean success,
894+
long deliveryTag,
895+
@Nullable Throwable cause) {
896+
if (messageAckListener == null) {
897+
return;
898+
}
899+
try {
900+
messageAckListener.onComplete(success, deliveryTag, cause);
901+
}
902+
catch (Exception e) {
903+
logger.error("An exception occured on MessageAckListener.", e);
904+
}
905+
}
906+
877907
@Override
878908
public String toString() {
879909
return "Consumer@" + ObjectUtils.getIdentityHexString(this) + ": "

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/DirectMessageListenerContainer.java‎

Lines changed: 47 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,7 @@
9292
* @author Gary Russell
9393
* @author Artem Bilan
9494
* @author Nicolas Ristock
95+
* @author Cao Weibo
9596
*
9697
* @since 2.0
9798
*
@@ -531,7 +532,7 @@ private void checkConsumers(long now) {
531532
try {
532533
consumer.ackIfNecessary(now);
533534
}
534-
catch (IOException e) {
535+
catch (Exception e) {
535536
this.logger.error("Exception while sending delayed ack", e);
536537
}
537538
}
@@ -882,7 +883,7 @@ private void cancelConsumer(SimpleConsumer consumer) {
882883
try {
883884
consumer.ackIfNecessary(0L);
884885
}
885-
catch (IOException e) {
886+
catch (Exception e) {
886887
this.logger.error("Exception while sending delayed ack", e);
887888
}
888889
}
@@ -1175,7 +1176,7 @@ private void handleAck(long deliveryTag, boolean channelLocallyTransacted) {
11751176
}
11761177
}
11771178
else if (!isChannelTransacted() || isLocallyTransacted) {
1178-
getChannel().basicAck(deliveryTag, false);
1179+
sendAckWithNotify(deliveryTag, false);
11791180
}
11801181
}
11811182
if (isLocallyTransacted) {
@@ -1194,7 +1195,7 @@ else if (!isChannelTransacted() || isLocallyTransacted) {
11941195
* @param now the current time.
11951196
* @throws IOException if one occurs.
11961197
*/
1197-
synchronized void ackIfNecessary(long now) throws IOException {
1198+
synchronized void ackIfNecessary(long now) throws Exception {
11981199
if (this.pendingAcks >= this.messagesPerAck || (
11991200
this.pendingAcks > 0 && (now - this.lastAck > this.ackTimeout || this.canceled))) {
12001201
sendAck(now);
@@ -1217,7 +1218,7 @@ private void rollback(long deliveryTag, Exception e) {
12171218
getChannel().basicNack(deliveryTag, !isAsyncReplies(),
12181219
ContainerUtils.shouldRequeue(isDefaultRequeueRejected(), e, this.logger));
12191220
}
1220-
catch (IOException e1) {
1221+
catch (Exception e1) {
12211222
this.logger.error("Failed to nack message", e1);
12221223
}
12231224
}
@@ -1226,12 +1227,51 @@ private void rollback(long deliveryTag, Exception e) {
12261227
}
12271228
}
12281229

1229-
protected synchronized void sendAck(long now) throws IOException {
1230-
getChannel().basicAck(this.latestDeferredDeliveryTag, true);
1230+
protected synchronized void sendAck(long now) throws Exception {
1231+
sendAckWithNotify(this.latestDeferredDeliveryTag, true);
12311232
this.lastAck = now;
12321233
this.pendingAcks = 0;
12331234
}
12341235

1236+
/**
1237+
* Send ack and notify MessageAckListener(if set).
1238+
* @param deliveryTag DeliveryTag of this ack.
1239+
* @param multiple Whether multiple ack.
1240+
* @throws Exception Occured when ack.
1241+
*/
1242+
private void sendAckWithNotify(long deliveryTag, boolean multiple) throws Exception {
1243+
try {
1244+
getChannel().basicAck(deliveryTag, multiple);
1245+
notifyMessageAckListener(getMessageAckListener(), true, deliveryTag, null);
1246+
}
1247+
catch (Exception e) {
1248+
notifyMessageAckListener(getMessageAckListener(), false, deliveryTag, e);
1249+
throw e;
1250+
}
1251+
}
1252+
1253+
/**
1254+
* Notify MessageAckListener set on message listener.
1255+
* @param messageAckListener MessageAckListener set on the message listener.
1256+
* @param success Whether ack succeeded.
1257+
* @param deliveryTag The deliveryTag of ack.
1258+
* @param cause If an exception occurs.
1259+
*/
1260+
private void notifyMessageAckListener(@Nullable MessageAckListener messageAckListener,
1261+
boolean success,
1262+
long deliveryTag,
1263+
@Nullable Throwable cause) {
1264+
if (messageAckListener == null) {
1265+
return;
1266+
}
1267+
try {
1268+
messageAckListener.onComplete(success, deliveryTag, cause);
1269+
}
1270+
catch (Exception e) {
1271+
this.logger.error("An exception occured on MessageAckListener.", e);
1272+
}
1273+
}
1274+
12351275
@Override
12361276
public void handleConsumeOk(String consumerTag) {
12371277
super.handleConsumeOk(consumerTag);

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageAckListener.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
/*
2+
* Copyright 2021 the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package org.springframework.amqp.rabbit.listener;
18+
19+
import org.springframework.amqp.core.AcknowledgeMode;
20+
import org.springframework.lang.Nullable;
21+
22+
/**
23+
* A listener for message ack when using {@link AcknowledgeMode#AUTO}.
24+
*
25+
* @author Cao Weibo
26+
* @see AbstractMessageListenerContainer#setMessageAckListener(MessageAckListener)
27+
*/
28+
@FunctionalInterface
29+
public interface MessageAckListener {
30+
31+
/**
32+
* Listener callback.
33+
* @param success Whether ack succeed.
34+
* @param deliveryTag The deliveryTag of ack.
35+
* @param cause The cause of failed ack.
36+
*
37+
* @throws Exception the exception during callback.
38+
*/
39+
void onComplete(boolean success, long deliveryTag, @Nullable Throwable cause) throws Exception;
40+
41+
}

‎spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1044,7 +1044,7 @@ private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Excep
10441044
executeWithList(channel, messages, deliveryTag, consumer);
10451045
}
10461046

1047-
return consumer.commitIfNecessary(isChannelLocallyTransacted());
1047+
return consumer.commitIfNecessary(isChannelLocallyTransacted(), getMessageAckListener());
10481048

10491049
}
10501050

0 commit comments

Comments
 (0)