Anonymitaet commented on a change in pull request #11570:
URL: https://github.com/apache/pulsar/pull/11570#discussion_r686473267
##########
File path: site2/docs/client-libraries-cpp.md
##########
@@ -253,52 +253,227 @@ pulsar+ssl://pulsar.us-west.example.com:6651
## Create a consumer
-To use Pulsar as a consumer, you need to create a consumer on the C++ client.
The following is an example.
+To use Pulsar as a consumer, you need to create a consumer on the C++ client.
There are two main ways of using the consumer:
+- blocking style: synchronously calling `receive(msg)`
+- asynchronous (event based) style: using a message listener
+
+### Blocking example
+
+The benefit of this approach is that it is the simplest code. Simply keeps
calling `receive(msg)` which blocks until a message is received.
+
+This example starts a subscription at the earliest offset and consumes 100
messages.
```c++
-Client client("pulsar://localhost:6650");
+#include <pulsar/Client.h>
+
+using namespace pulsar;
+
+int main() {
+ Client client("pulsar://localhost:6650");
-Consumer consumer;
-Result result = client.subscribe("my-topic", "my-subscription-name", consumer);
-if (result != ResultOk) {
- LOG_ERROR("Failed to subscribe: " << result);
- return -1;
+ Consumer consumer;
+ ConsumerConfiguration config;
+ config.setSubscriptionInitialPosition(InitialPositionEarliest);
+ Result result = client.subscribe("persistent://public/default/my-topic",
"consumer-1", config, consumer);
+ if (result != ResultOk) {
+ std::cout << "Failed to subscribe: " << result << std::endl;
+ return -1;
+ }
+
+ Message msg;
+ int ctr = 0;
+ // consume 100 messages
+ while (ctr < 100) {
+ consumer.receive(msg);
+ std::cout << "Received: " << msg
+ << " with payload '" << msg.getDataAsString() << "'" << std::endl;
+
+ consumer.acknowledge(msg);
+ ctr++;
+ }
+
+ std::cout << "Finished consuming synchronously!" << std::endl;
+
+ client.close();
+ return 0;
}
+```
+
+### Consumer with a message listener
+
+We can avoid the need to run a loop with blocking calls with an event based
style by using a message listener which is invoked for each message that is
received.
+
+This example starts a subscription at the earliest offset and consumes 100
messages.
-Message msg;
+```c++
+#include <pulsar/Client.h>
+#include <atomic>
+#include <thread>
+
+using namespace pulsar;
-while (true) {
- consumer.receive(msg);
- LOG_INFO("Received: " << msg
- << " with payload '" << msg.getDataAsString() << "'");
+std::atomic<uint32_t> messagesReceived;
- consumer.acknowledge(msg);
+void handleAckComplete(Result res) {
+ std::cout << "Ack res: " << res << std::endl;
}
-client.close();
+void listener(Consumer consumer, const Message& msg) {
+ std::cout << "Got message " << msg << " with content '" <<
msg.getDataAsString() << "'" << std::endl;
+ messagesReceived++;
+ consumer.acknowledgeAsync(msg.getMessageId(), handleAckComplete);
+}
+
+int main() {
+ Client client("pulsar://localhost:6650");
+
+ Consumer consumer;
+ ConsumerConfiguration config;
+ config.setMessageListener(listener);
+ config.setSubscriptionInitialPosition(InitialPositionEarliest);
+ Result result = client.subscribe("persistent://public/default/my-topic",
"consumer-1", config, consumer);
+ if (result != ResultOk) {
+ std::cout << "Failed to subscribe: " << result << std::endl;
+ return -1;
+ }
+
+ // wait for 100 messages to be consumed
+ while (messagesReceived < 100) {
+ std::this_thread::sleep_for(std::chrono::milliseconds(100));
+ }
+
+ std::cout << "Finished consuming asynchronously!" << std::endl;
+
+ client.close();
+ return 0;
+}
```
## Create a producer
-To use Pulsar as a producer, you need to create a producer on the C++ client.
The following is an example.
+To use Pulsar as a producer, you need to create a producer on the C++ client.
There are two main ways of using a producer:
+- blocking style where each call to `send` waits for an ack from the broker.
+- non-blocking asynchronous style where `sendAsync` is called instead of
`send` and a callback is supplied for when the ack is received from the broker.
+
+### Simple blocking example
+
+This example sends 100 messages using the blocking style. While simple, it
does not produce high throughput as it waits for each ack to come back before
sending the next message.
+
+```c++
+#include <pulsar/Client.h>
+#include <thread>
+
+using namespace pulsar;
+
+int main() {
+ Client client("pulsar://localhost:6650");
+
+ Result result =
client.createProducer("persistent://public/default/my-topic", producer);
+ if (result != ResultOk) {
+ std::cout << "Error creating producer: " << result << std::endl;
+ return -1;
+ }
+
+ // Send 100 messages synchronously
+ int ctr = 0;
+ while (ctr < 100) {
+ std::string content = "msg" + std::to_string(ctr);
+ Message msg = MessageBuilder().setContent(content).setProperty("x",
"1").build();
+ Result result = producer.send(msg);
+ if (result != ResultOk) {
+ std::cout << "The message " << content << " could not be sent,
received code: " << result << std::endl;
+ } else {
+ std::cout << "The message " << content << " sent successfully" <<
std::endl;
+ }
+
+ std::this_thread::sleep_for(std::chrono::milliseconds(100));
+ ctr++;
+ }
+
+ std::cout << "Finished producing synchronously!" << std::endl;
+
+ client.close();
+ return 0;
+}
+```
+
+### Non-blockging example
Review comment:
blocking?
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
