i can't find anything about callback after acked. why there's no before and after method in MessageListener?

[ltbyun]

Affects Version(s): <Spring AMQP version>
all

Enhancement

sometimes we used some state in memory map. needs to get it in listener#onMessage, and remove it after acked.
but now i can only remove it in onMessage, if the method exits and ack failed, the state lost.
next time, the same message comes to onMessage, but the state is gone.
so, i was wondering why there's no interceptor like before and after method in MessageListener, then we can do some inilialize before invoke onMessage and release some resources after acked ok or not.

MessagePostProcessor seems to be act as before interceptor, what about create a new MessageAckProcessor?

[artembilan]

Let's see if adviceChain option on the Listener Container may help you!

https://docs.spring.io/spring-amqp/docs/current/reference/html/#containerAttributes

An array of AOP Advice to apply to the listener execution. This can be used to apply additional cross-cutting concerns, such as automatic retry in the event of broker death. Note that simple re-connection after an AMQP error is handled by the CachingConnectionFactory, as long as the broker is still alive.

So, you write your own MethodInterceptor to deal with these args before and after calling proceed(): Channel channel, Object data.

[ltbyun]

it works, thanks

[ltbyun]

the aop advice works on target method void onMessage(Message message), not after acked.
in some condition, i need to cancel consume queue after consume message acked.
the aop advice can not do this.

[garyrussell]

You would need to add a reference to the listener container to the advice and stop the container (use the async stop(() -> {}) to avoid a deadlock pause).

[ltbyun]

You would need to add a reference to the listener container to the advice and stop the container (use the async stop(() -> {}) to avoid a deadlock pause).

why stop container, i only need to cancel some not all.
i used one container to consume from thousands queues. but not all queues have the same behavior, some needs to cancel when run in some condition, the rest doesn't.
if every cancel by a stop, then nothing to do but stop, start, stop, start.

[ltbyun]

right now i used manual ack, then remove queueNames from container, may be another thread to remove, i'm not sure.
i wonder why not support ack callback. in a product environment, many things need to do after ack.
the advice chain works around on the method void onMessage(Message message), but the method is defined by developer himself, and only one MessageListener per container, we can just write it in the method, the advice chain seems to worth nothing.

            if (queueNameMeta.mode == TaskModeEnum.FULL && packet.isFullEnd()) {
                channel.basicAck(deliveryTag, false);
                container.removeQueueNames(props.getConsumerQueue());
                return;
            }

[garyrussell]

Ok; now I see what your requirements are.

You are correct, there is currently no callback after the ack.

Contributions are welcome.

[zysaaa]

This requirement seems simple to achieve. Define a method in AbstractMessageListenerContainer and call it where SMLC and DMLC call channel#ack, for example:

        interface AckListener {
		void onComplete(boolean success, long deliveryTag, @Nullable Throwable cause) throws Exception;
	}

	AckListener ackListener;

	protected void notifyAckListener(boolean success, long deliverTag, @Nullable Throwable cause) {
		if (ackListener == null) {
			return;
		}
		try {
			ackListener.onComplete(success, deliverTag, cause);
		} catch (Exception e) {
			logger.error("xxx.", e);
		}
	}

Is it necessary to implement this feature? If necessary, does this approach make sense? @garyrussell

[garyrussell]

Looks reasonable to me.