add producer flush method in cpp client (apache#3020)

### Motivation We already have flush() method in java api: http://pulsar.apache.org/api/client/org/apache/pulsar/client/api/Producer.html#flush-- It is better to provide a same method for cpp client. ### Modifications Add related methods in cpp client. Add related tests. ### Result Cpp unit tests pass.
larslevie · Nov 20, 2018 · b982b5d · b982b5d
1 parent 5fde6f3
commit b982b5d
Show file tree

Hide file tree

Showing 10 changed files with 320 additions and 12 deletions.
diff --git a/pulsar-client-cpp/include/pulsar/Producer.h b/pulsar-client-cpp/include/pulsar/Producer.h
@@ -29,6 +29,9 @@ namespace pulsar {
 class ProducerImplBase;
 class PulsarWrapper;
 class PulsarFriend;
+
+typedef boost::function<void(Result)> FlushCallback;
+
 class Producer {
    public:
     /**
@@ -80,6 +83,18 @@ class Producer {
      */
     void sendAsync(const Message& msg, SendCallback callback);
 
+    /**
+     * Flush all the messages buffered in the client and wait until all messages have been successfully
+     * persisted.
+     */
+    Result flush();
+
+    /**
+     * Flush all the messages buffered in the client and wait until all messages have been successfully
+     * persisted.
+     */
+    void flushAsync(FlushCallback callback);
+
     /**
      * Get the last sequence id that was published by this producer.
      *

diff --git a/pulsar-client-cpp/lib/BatchMessageContainer.cc b/pulsar-client-cpp/lib/BatchMessageContainer.cc
@@ -47,7 +47,7 @@ void BatchMessageContainer::add(const Message& msg, SendCallback sendCallback, b
                     << "]");
     if (!(disableCheck || hasSpaceInBatch(msg))) {
         LOG_DEBUG(*this << " Batch is full");
-        sendMessage();
+        sendMessage(NULL);
         add(msg, sendCallback, true);
         return;
     }
@@ -71,7 +71,7 @@ void BatchMessageContainer::add(const Message& msg, SendCallback sendCallback, b
     LOG_DEBUG(*this << " Batch Payload Size In Bytes = " << batchSizeInBytes_);
     if (isFull()) {
         LOG_DEBUG(*this << " Batch is full.");
-        sendMessage();
+        sendMessage(NULL);
     }
 }
 
@@ -83,11 +83,14 @@ void BatchMessageContainer::startTimer() {
                                    boost::asio::placeholders::error));
 }
 
-void BatchMessageContainer::sendMessage() {
+void BatchMessageContainer::sendMessage(FlushCallback flushCallback) {
     // Call this function after acquiring the ProducerImpl lock
     LOG_DEBUG(*this << "Sending the batch message container");
     if (isEmpty()) {
         LOG_DEBUG(*this << " Batch is empty - returning.");
+        if (flushCallback) {
+            flushCallback(ResultOk);
+        }
         return;
     }
     impl_->metadata.set_num_messages_in_batch(messagesContainerListPtr_->size());
@@ -101,8 +104,8 @@ void BatchMessageContainer::sendMessage() {
     msg.impl_ = impl_;
 
     // bind keeps a copy of the parameters
-    SendCallback callback =
-        boost::bind(&BatchMessageContainer::batchMessageCallBack, _1, messagesContainerListPtr_);
+    SendCallback callback = boost::bind(&BatchMessageContainer::batchMessageCallBack, _1,
+                                        messagesContainerListPtr_, flushCallback);
 
     producer_.sendMessage(msg, callback);
     clear();
@@ -131,8 +134,12 @@ void BatchMessageContainer::clear() {
     batchSizeInBytes_ = 0;
 }
 
-void BatchMessageContainer::batchMessageCallBack(Result r, MessageContainerListPtr messagesContainerListPtr) {
+void BatchMessageContainer::batchMessageCallBack(Result r, MessageContainerListPtr messagesContainerListPtr,
+                                                 FlushCallback flushCallback) {
     if (!messagesContainerListPtr) {
+        if (flushCallback) {
+            flushCallback(ResultOk);
+        }
         return;
     }
     LOG_DEBUG("BatchMessageContainer::batchMessageCallBack called with [Result = "
@@ -142,6 +149,9 @@ void BatchMessageContainer::batchMessageCallBack(Result r, MessageContainerListP
         // callback(result, message)
         iter->sendCallback_(r, iter->message_);
     }
+    if (flushCallback) {
+        flushCallback(ResultOk);
+    }
 }
 
 BatchMessageContainer::~BatchMessageContainer() {

diff --git a/pulsar-client-cpp/lib/BatchMessageContainer.h b/pulsar-client-cpp/lib/BatchMessageContainer.h
@@ -64,7 +64,7 @@ class BatchMessageContainer {
 
     void clear();
 
-    static void batchMessageCallBack(Result r, MessageContainerListPtr messages);
+    static void batchMessageCallBack(Result r, MessageContainerListPtr messages, FlushCallback callback);
 
     friend inline std::ostream& operator<<(std::ostream& os,
                                            const BatchMessageContainer& batchMessageContainer);
@@ -108,7 +108,7 @@ class BatchMessageContainer {
 
     void startTimer();
 
-    void sendMessage();
+    void sendMessage(FlushCallback callback);
 };
 
 bool BatchMessageContainer::hasSpaceInBatch(const Message& msg) const {

diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc
@@ -241,4 +241,37 @@ void PartitionedProducerImpl::triggerFlush() {
     }
 }
 
+void PartitionedProducerImpl::flushAsync(FlushCallback callback) {
+    if (!flushPromise_ || flushPromise_->isComplete()) {
+        flushPromise_ = boost::make_shared<Promise<Result, bool_type>>();
+    } else {
+        // already in flushing, register a listener callback
+        boost::function<void(Result, bool)> listenerCallback = [this, callback](Result result, bool_type v) {
+            if (v) {
+                callback(ResultOk);
+            } else {
+                callback(ResultUnknownError);
+            }
+            return;
+        };
+
+        flushPromise_->getFuture().addListener(listenerCallback);
+        return;
+    }
+
+    FlushCallback subFlushCallback = [this, callback](Result result) {
+        int previous = flushedPartitions_.fetch_add(1);
+        if (previous == producers_.size() - 1) {
+            flushedPartitions_.store(0);
+            flushPromise_->setValue(true);
+            callback(result);
+        }
+        return;
+    };
+
+    for (ProducerList::const_iterator prod = producers_.begin(); prod != producers_.end(); prod++) {
+        (*prod)->flushAsync(subFlushCallback);
+    }
+}
+
 }  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.h b/pulsar-client-cpp/lib/PartitionedProducerImpl.h
@@ -71,6 +71,8 @@ class PartitionedProducerImpl : public ProducerImplBase,
 
     virtual void triggerFlush();
 
+    virtual void flushAsync(FlushCallback callback);
+
     void handleSinglePartitionProducerCreated(Result result, ProducerImplBaseWeakPtr producerBaseWeakPtr,
                                               const unsigned int partitionIndex);
 
@@ -115,6 +117,9 @@ class PartitionedProducerImpl : public ProducerImplBase,
     Promise<Result, ProducerImplBaseWeakPtr> partitionedProducerCreatedPromise_;
 
     MessageRoutingPolicyPtr getMessageRouter();
+
+    std::atomic<int> flushedPartitions_;
+    boost::shared_ptr<Promise<Result, bool_type>> flushPromise_;
 };
 
 }  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/Producer.cc b/pulsar-client-cpp/lib/Producer.cc
@@ -77,4 +77,22 @@ void Producer::closeAsync(CloseCallback callback) {
 
     impl_->closeAsync(callback);
 }
+
+Result Producer::flush() {
+    Promise<bool, Result> promise;
+    flushAsync(WaitForCallback(promise));
+
+    Result result;
+    promise.getFuture().get(result);
+    return result;
+}
+
+void Producer::flushAsync(FlushCallback callback) {
+    if (!impl_) {
+        callback(ResultProducerNotInitialized);
+        return;
+    }
+
+    impl_->flushAsync(callback);
+}
 }  // namespace pulsar
diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc
@@ -245,7 +245,7 @@ void ProducerImpl::failPendingMessages(Result result) {
     }
 
     // this function can handle null pointer
-    BatchMessageContainer::batchMessageCallBack(ResultTimeout, messageContainerListPtr);
+    BatchMessageContainer::batchMessageCallBack(ResultTimeout, messageContainerListPtr, NULL);
 }
 
 void ProducerImpl::resendMessages(ClientConnectionPtr cnx) {
@@ -285,10 +285,43 @@ void ProducerImpl::statsCallBackHandler(Result res, const Message& msg, SendCall
     }
 }
 
+void ProducerImpl::flushAsync(FlushCallback callback) {
+    if (batchMessageContainer) {
+        if (!flushPromise_ || flushPromise_->isComplete()) {
+            flushPromise_ = boost::make_shared<Promise<Result, bool_type>>();
+        } else {
+            // already in flushing, register a listener callback
+            boost::function<void(Result, bool)> listenerCallback = [this, callback](Result result,
+                                                                                    bool_type v) {
+                if (v) {
+                    callback(ResultOk);
+                } else {
+                    callback(ResultUnknownError);
+                }
+                return;
+            };
+
+            flushPromise_->getFuture().addListener(listenerCallback);
+            return;
+        }
+
+        FlushCallback innerCallback = [this, callback](Result result) {
+            flushPromise_->setValue(true);
+            callback(result);
+            return;
+        };
+
+        Lock lock(mutex_);
+        batchMessageContainer->sendMessage(innerCallback);
+    } else {
+        callback(ResultOk);
+    }
+}
+
 void ProducerImpl::triggerFlush() {
     if (batchMessageContainer) {
         Lock lock(mutex_);
-        batchMessageContainer->sendMessage();
+        batchMessageContainer->sendMessage(NULL);
     }
 }
 
@@ -364,7 +397,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) {
         // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout
         if (batchMessageContainer) {
             LOG_DEBUG(getName() << " - sending batch message immediately");
-            batchMessageContainer->sendMessage();
+            batchMessageContainer->sendMessage(NULL);
         }
         lock.unlock();
         cb(ResultProducerQueueIsFull, msg);
@@ -412,7 +445,7 @@ void ProducerImpl::batchMessageTimeoutHandler(const boost::system::error_code& e
     }
     LOG_DEBUG(getName() << " - Batch Message Timer expired");
     Lock lock(mutex_);
-    batchMessageContainer->sendMessage();
+    batchMessageContainer->sendMessage(NULL);
 }
 
 void ProducerImpl::printStats() {

diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h
@@ -34,6 +34,7 @@
 using namespace pulsar;
 
 namespace pulsar {
+typedef bool bool_type;
 
 class BatchMessageContainer;
 
@@ -90,6 +91,8 @@ class ProducerImpl : public HandlerBase,
 
     virtual void triggerFlush();
 
+    virtual void flushAsync(FlushCallback callback);
+
    protected:
     ProducerStatsBasePtr producerStatsBasePtr_;
 
@@ -156,6 +159,7 @@ class ProducerImpl : public HandlerBase,
     MessageCryptoPtr msgCrypto_;
     DeadlineTimerPtr dataKeyGenTImer_;
     uint32_t dataKeyGenIntervalSec_;
+    boost::shared_ptr<Promise<Result, bool_type>> flushPromise_;
 };
 
 struct ProducerImplCmp {

diff --git a/pulsar-client-cpp/lib/ProducerImplBase.h b/pulsar-client-cpp/lib/ProducerImplBase.h
@@ -43,6 +43,7 @@ class ProducerImplBase {
     virtual const std::string& getTopic() const = 0;
     virtual Future<Result, ProducerImplBaseWeakPtr> getProducerCreatedFuture() = 0;
     virtual void triggerFlush() = 0;
+    virtual void flushAsync(FlushCallback callback) = 0;
 };
 }  // namespace pulsar
 #endif  // PULSAR_PRODUCER_IMPL_BASE_HEADER