You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/12/09 03:39:04 UTC
[GitHub] [pulsar] h-j-13 opened a new issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
h-j-13 opened a new issue #8869:
URL: https://github.com/apache/pulsar/issues/8869
I want use `consumer.receiveAsync()` to receive a single message from topic. but i don't know how to confirm message properly, because the `receiveAsync()` callback function don't provide a consumer as args.
```cpp
// >>> lib/Consumer.cc#receiveAsync
void Consumer::receiveAsync(ReceiveCallback callback) {
if (!impl_) {
Message msg;
callback(ResultConsumerNotInitialized, msg);
return;
}
impl_->receiveAsync(callback);
}
// callback function
// >>> include/pulsar/ConsumerConfiguration.h#ReceiveCallback
typedef std::function<void(Result, const Message& msg)> ReceiveCallback;
```
OR the message has already been confirmed before callback function called?
My Demo. i don't know how to properly confirm message. confused.
```cpp
#include <iostream>
#include <pulsar/Client.h>
#include <lib/LogUtils.h>
DECLARE_LOG_OBJECT()
using namespace pulsar;
using namespace std;
void callback(Result code, const Message& msg) {
cout << "Received code: " << code << "Msg Length" << msg.getLength() << endl;
// consumer.acknowledge(); ?
}
int main() {
cout << "Sync Consumer Test | pulsar://100.76.43.216:6650" << endl;
Client client("pulsar://localhost:6650");
Consumer consumer;
Consumer consumer2;
Consumer consumer3;
ConsumerConfiguration config;
config.setConsumerType(ConsumerShared);
Result result =
client.subscribe("persistent://HJ13/Test/performance_test", "consumerRecvAsync", config, consumer);
if (result != ResultOk) {
LOG_ERROR("Failed to subscribe: " << result);
return -1;
}
consumer.receiveAsync(callback);
// consumer.acknowledge(); ?
// Wait
int n;
std::cin >> n;
client.close();
}
```
I want confirm message like `pulsar-client-cpp/examples/SampleConsumerListener.cc`
```cpp
...
void listener(Consumer consumer, const Message& msg) {
LOG_INFO("Got message " << msg << " with content '" << msg.getDataAsString() << "'");
consumer.acknowledge(msg.getMessageId());
}
...
```
https://github.com/apache/pulsar/blob/master/pulsar-client-cpp/examples/SampleConsumerListener.cc#L32
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] zhoutong1996 commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
zhoutong1996 commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-741507143
that must be a problem
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] h-j-13 commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
h-j-13 commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-742364484
Thx for your explanation!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-742260653
It's a thing about API design. Though both `MessageListener` and `ReceiveCallback` are `std::function` callbacks, they're much different. Let's see what Java API does first.
```java
public interface MessageListener<T> extends Serializable {
/**
* This method is called whenever a new message is received.
*
* <p>Messages are guaranteed to be delivered in order and from the same thread for a single consumer
*
* <p>This method will only be called once for each message, unless either application or broker crashes.
*
* <p>Application is responsible for acking message by calling any of consumer acknowledgement methods.
*
* <p>Application is responsible of handling any exception that could be thrown while processing the message.
*
* @param consumer
* the consumer that received the message
* @param msg
* the message object
*/
void received(Consumer<T> consumer, Message<T> msg);
}
```
The `MessageListener` is a real callback and it's noted that the callback is responsible for asking messages.
```java
/**
* Receive a single message
*
* <p>Retrieves a message when it will be available and completes {@link CompletableFuture} with received message.
*
* <p>{@code receiveAsync()} should be called subsequently once returned {@code CompletableFuture} gets complete
* with received message. Else it creates <i> backlog of receive requests </i> in the application.
*
* <p>The returned future can be cancelled before completion by calling {@code .cancel(false)}
* ({@link CompletableFuture#cancel(boolean)}) to remove it from the the backlog of receive requests. Another
* choice for ensuring a proper clean up of the returned future is to use the CompletableFuture.orTimeout method
* which is available on JDK9+. That would remove it from the backlog of receive requests if receiving exceeds
* the timeout.
*
* @return {@link CompletableFuture}<{@link Message}> will be completed when message is available
*/
CompletableFuture<Message<T>> receiveAsync();
```
There's no `ReceiveCallback` in Java client because `receiveAsync` just returns a future of `Message`. C++ client has no future before C++11 while it seems that the Pulsar C++ client is written with C++03 initially. Even with C++11, the `std::future` is not so convenient as Java's `CompletableFuture`.
Therefore, C++ client API usually passes an extra callback argument `std::function<Result, const T&>` to replace Java API's `CompletableFuture<T>`.
The `receiveAsync` is only responsible for processing the future of `Message` but not how you acknowledge it. For example, you can also capture a list of `MessageId` and acknowledge it later.
```c++
std::vector<MessageId> messageIds;
std::mutex mtx;
std::thread ackPeriodlyTask([&consumer, &messageIds, &mtx] {
while (true) { // TODO: add a flag to stop the task
std::this_thread::sleep_for(std::chrono::seconds(1));
std::lock_guard<std::mutex> lock(mtx);
// TODO: you need to acknowledge all message ids if the subscription mode is Shared or Key_Shared
consumer.acknowledgeCumulative(messageIds.back());
messageIds.clear();
}
});
for (int i = 0; i < 100000; i++) {
consumer.receiveAsync([&messageIds, &mtx](Result result, const Message& msg) {
// TODO: handle `result` and `msg`
std::lock_guard<std::mutex> lock(mtx);
messageIds.emplace_back(msg.getMessageId());
});
}
```
Here we don't acknowledge each message because in failover subscription mode, it's unnecessary and inefficient. Though you can set `AckGrouingTimeMs` and `AckGroupingMaxSize` now, the above code is just an example. What I want to say is:
- `receiveAsync` only focus on the received message in the future.
- `MessageListener` requires you to handle the acknowledgement.
The difference makes that `MessageListener` has a `Consumer` parameter while `ReceiveCallback` doesn't.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] BewareMyPower commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-741830890
Use lambda capture
```c++
consumer.receiveAsync([&consumer](Result code, const Message& msg) {
// process `code` and `msg`...
consumer.acknowledge(msg);
});
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] h-j-13 commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
h-j-13 commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-741507339
Related to https://github.com/apache/pulsar/pull/577
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] h-j-13 closed issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
h-j-13 closed issue #8869:
URL: https://github.com/apache/pulsar/issues/8869
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] h-j-13 commented on issue #8869: [pulsar-client-cpp] When I use the `consumer.receiveAsync()`, how to properly confirm message?
Posted by GitBox <gi...@apache.org>.
h-j-13 commented on issue #8869:
URL: https://github.com/apache/pulsar/issues/8869#issuecomment-742217933
> Use lambda capture
>
> ```c++
> consumer.receiveAsync([&consumer](Result code, const Message& msg) {
> // process `code` and `msg`...
> consumer.acknowledge(msg);
> });
> ```
Thx, it worked. But I still think that the parameters of the callback function should be updated like
```c
typedef std::function<void(Consumer consumer, Result, const Message& msg)> ReceiveCallback;
```
to keep the code style uniform
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org