hrsakai commented on a change in pull request #1977: Enable listener to receive
messages even if receiver queue size is zero
URL: https://github.com/apache/incubator-pulsar/pull/1977#discussion_r196266417
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
##########
@@ -722,12 +726,12 @@ void messageReceived(MessageIdData messageId, ByteBuf
headersAndPayload, ClientC
// if the conf.getReceiverQueueSize() is 0 then discard
message if no one is waiting for it.
// if asyncReceive is waiting then notify callback without
adding to incomingMessages queue
unAckedMessageTracker.add((MessageIdImpl)
message.getMessageId());
- boolean asyncReceivedWaiting = !pendingReceives.isEmpty();
- if ((conf.getReceiverQueueSize() != 0 ||
waitingOnReceiveForZeroQueueSize) && !asyncReceivedWaiting) {
- incomingMessages.add(message);
- }
- if (asyncReceivedWaiting) {
+ if (!pendingReceives.isEmpty()) {
notifyPendingReceivedCallback(message, null);
+ } else if (conf.getReceiverQueueSize() != 0 ||
waitingOnReceiveForZeroQueueSize) {
+ incomingMessages.add(message);
+ } else if (conf.getReceiverQueueSize() == 0 && listener !=
null) {
+ triggerZeroQueueSizeListener(message);
Review comment:
why do you call `triggerZeroQueueSizeListener()` instead of adding message
to incommingMessages?
you can't use [existing
logic](https://github.com/massakam/pulsar/blob/c1ab9d0edfec283c46e346d6d48b6134542bd01f/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java#L761-L792
)?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
