[TableView-2] Implement all interfaces.

include/pulsar/Reader.h

@@ -164,6 +164,7 @@ class PULSAR_PUBLIC Reader {
     friend class PulsarFriend;
     friend class PulsarWrapper;
     friend class ReaderImpl;
+    friend class TableViewImpl;
     friend class ReaderTest;
 };
 }  // namespace pulsar

include/pulsar/TableView.h

@@ -112,7 +112,7 @@ class PULSAR_PUBLIC TableView {
     void closeAsync(ResultCallback callback);
 
     /**
-     * Close the consumer and stop the broker to push more messages
+     * Close the table view and stop the broker to push more messages
      */
     Result close();
 

lib/Client.cc

@@ -150,6 +150,20 @@ void Client::createReaderAsync(const std::string& topic, const MessageId& startM
     impl_->createReaderAsync(topic, startMessageId, conf, callback);
 }
 
+Result Client::createTableView(const std::string& topic, const TableViewConfiguration& conf,
+                               TableView& tableView) {
+    Promise<Result, TableView> promise;
+    createTableViewAsync(topic, conf, WaitForCallbackValue<TableView>(promise));
+    Future<Result, TableView> future = promise.getFuture();
+
+    return future.get(tableView);
+}
+
+void Client::createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
+                                  TableViewCallback callBack) {
+    impl_->createTableViewAsync(topic, conf, callBack);
+}
+
 Result Client::getPartitionsForTopic(const std::string& topic, std::vector<std::string>& partitions) {
     Promise<Result, std::vector<std::string> > promise;
     getPartitionsForTopicAsync(topic, WaitForCallbackValue<std::vector<std::string> >(promise));

lib/ClientImpl.cc

@@ -35,6 +35,7 @@
 #include "ProducerImpl.h"
 #include "ReaderImpl.h"
 #include "RetryableLookupService.h"
+#include "TableViewImpl.h"
 #include "TimeUtils.h"
 #include "TopicName.h"
 
@@ -244,6 +245,33 @@ void ClientImpl::createReaderAsync(const std::string& topic, const MessageId& st
                   std::placeholders::_2, topicName, msgId, conf, callback));
 }
 
+void ClientImpl::createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
+                                      TableViewCallback callback) {
+    TopicNamePtr topicName;
+    {
+        Lock lock(mutex_);
+        if (state_ != Open) {
+            lock.unlock();
+            callback(ResultAlreadyClosed, TableView());
+            return;
+        } else if (!(topicName = TopicName::get(topic))) {
+            lock.unlock();
+            callback(ResultInvalidTopicName, TableView());
+            return;
+        }
+    }
+
+    TableViewImplPtr tableViewPtr =
+        std::make_shared<TableViewImpl>(shared_from_this(), topicName->toString(), conf);
+    tableViewPtr->start().addListener([callback](Result result, TableViewImplPtr tableViewImplPtr) {
+        if (result == ResultOk) {
+            callback(result, TableView{tableViewImplPtr});
+        } else {
+            callback(result, {});
+        }
+    });
+}
+
 void ClientImpl::handleReaderMetadataLookup(const Result result, const LookupDataResultPtr partitionMetadata,
                                             TopicNamePtr topicName, MessageId startMessageId,
                                             ReaderConfiguration conf, ReaderCallback callback) {

lib/ClientImpl.h

@@ -42,6 +42,9 @@ class ReaderImpl;
 typedef std::shared_ptr<ReaderImpl> ReaderImplPtr;
 typedef std::weak_ptr<ReaderImpl> ReaderImplWeakPtr;
 
+class TableViewImpl;
+typedef std::shared_ptr<TableViewImpl> TableViewImplPtr;
+
 class ConsumerImplBase;
 typedef std::weak_ptr<ConsumerImplBase> ConsumerImplBaseWeakPtr;
 
@@ -87,6 +90,9 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
     void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
                            const ReaderConfiguration& conf, ReaderCallback callback);
 
+    void createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
+                              TableViewCallback callback);
+
     void getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback);
 
     Future<Result, ClientConnectionWeakPtr> getConnection(const std::string& topic);

lib/MultiTopicsConsumerImpl.cc

@@ -995,7 +995,7 @@ void MultiTopicsConsumerImpl::subscribeSingleNewConsumer(
 
     auto consumer = std::make_shared<ConsumerImpl>(client, topicPartitionName, subscriptionName_, config,
                                                    topicName->isPersistent(), internalListenerExecutor, true,
-                                                   Partitioned);
+                                                   Partitioned, subscriptionMode_, startMessageId_);
     consumer->getConsumerCreatedFuture().addListener(
         [this, weakSelf, partitionsNeedCreate, topicSubResultPromise](
             Result result, const ConsumerImplBaseWeakPtr& consumerImplBaseWeakPtr) {

lib/SynchronizedHashMap.h

@@ -122,7 +122,6 @@ class SynchronizedHashMap {
         return pairs;
     }
 
-    // This method is only used for test
     size_t size() const noexcept {
         Lock lock(mutex_);
         return data_.size();

lib/TableView.cc

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include <pulsar/TableView.h>
+
+#include "TableViewImpl.h"
+#include "Utils.h"
+
+namespace pulsar {
+
+TableView::TableView() {}
+
+TableView::TableView(TableViewImplPtr impl) : impl_(impl) {}
+
+bool TableView::retrieveValue(const std::string& key, std::string& value) {
+    if (impl_) {
+        return impl_->retrieveValue(key, value);
+    }
+    return false;
+}
+
+bool TableView::getValue(const std::string& key, std::string& value) const {
+    if (impl_) {
+        return impl_->getValue(key, value);
+    }
+    return false;
+}
+
+bool TableView::containsKey(const std::string& key) const {
+    if (impl_) {
+        return impl_->containsKey(key);
+    }
+    return false;
+}
+
+std::unordered_map<std::string, std::string> TableView::snapshot() {
+    if (impl_) {
+        return impl_->snapshot();
+    }
+    return {};
+}
+
+std::size_t TableView::size() const {
+    if (impl_) {
+        return impl_->size();
+    }
+    return 0;
+}
+
+void TableView::forEach(TableViewAction action) {
+    if (impl_) {
+        impl_->forEach(action);
+    }
+}
+
+void TableView::forEachAndListen(TableViewAction action) {
+    if (impl_) {
+        impl_->forEachAndListen(action);
+    }
+}
+
+void TableView::closeAsync(ResultCallback callback) {
+    if (!impl_) {
+        callback(ResultConsumerNotInitialized);
+        return;
+    }
+
+    impl_->closeAsync(callback);
+}
+
+Result TableView::close() {
+    if (!impl_) {
+        return ResultConsumerNotInitialized;
+    }
+    Promise<bool, Result> promise;
+    impl_->closeAsync(WaitForCallback(promise));
+    Result result;
+    promise.getFuture().get(result);
+    return result;
+}
+
+}  // namespace pulsar

lib/TableViewImpl.cc

@@ -0,0 +1,169 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "TableViewImpl.h"
+
+#include "ClientImpl.h"
+#include "LogUtils.h"
+#include "ReaderImpl.h"
+#include "TimeUtils.h"
+
+namespace pulsar {
+
+DECLARE_LOG_OBJECT()
+
+TableViewImpl::TableViewImpl(ClientImplPtr client, const std::string& topic,
+                             const TableViewConfiguration& conf)
+    : client_(client), topic_(topic), conf_(conf) {}
+
+Future<Result, TableViewImplPtr> TableViewImpl::start() {
+    Promise<Result, TableViewImplPtr> promise;
+    ReaderConfiguration readerConfiguration;
+    readerConfiguration.setSchema(conf_.schemaInfo);
+    readerConfiguration.setReadCompacted(true);
+    readerConfiguration.setInternalSubscriptionName(conf_.subscriptionName);
+
+    TableViewImplPtr self = shared_from_this();
+    ReaderCallback readerCallback = [self, promise](Result res, Reader reader) {
+        if (res == ResultOk) {
+            self->reader_ = reader.impl_;
+            self->readAllExistingMessages(promise, TimeUtils::currentTimeMillis(), 0);
+        } else {
+            promise.setFailed(res);
+        }
+    };
+    client_->createReaderAsync(topic_, MessageId::earliest(), readerConfiguration, readerCallback);
+    return promise.getFuture();
+}
+
+bool TableViewImpl::retrieveValue(const std::string& key, std::string& value) {
+    auto optValue = data_.remove(key);
+    if (optValue) {
+        value = optValue.value();
+        return true;
+    }
+    return false;
+}
+
+bool TableViewImpl::getValue(const std::string& key, std::string& value) const {
+    auto optValue = data_.find(key);
+    if (optValue) {
+        value = optValue.value();
+        return true;
+    }
+    return false;
+}
+
+bool TableViewImpl::containsKey(const std::string& key) const { return data_.find(key) != boost::none; }
+
+std::unordered_map<std::string, std::string> TableViewImpl::snapshot() { return data_.move(); }
+
+std::size_t TableViewImpl::size() const { return data_.size(); }
+
+void TableViewImpl::forEach(TableViewAction action) { data_.forEach(action); }
+
+void TableViewImpl::forEachAndListen(TableViewAction action) {
+    data_.forEach(action);
+    Lock lock(listenersMutex_);
+    listeners_.emplace_back(action);
+}
+
+void TableViewImpl::closeAsync(ResultCallback callback) {
+    if (reader_) {
+        reader_->closeAsync([callback, this](Result result) {
+            reader_.reset();
+            callback(result);
+        });
+    } else {
+        callback(ResultConsumerNotInitialized);
+    }
+}
+
+void TableViewImpl::handleMessage(const Message& msg) {
+    if (msg.hasPartitionKey()) {
+        auto value = msg.getDataAsString();
+        LOG_DEBUG("Applying message from " << topic_ << " key=" << msg.getPartitionKey()
+                                           << " value=" << value)
+
+        if (msg.getLength() == 0) {
+            data_.remove(msg.getPartitionKey());
+        } else {
+            data_.emplace(msg.getPartitionKey(), value);
+        }
+
+        Lock lock(listenersMutex_);
+        for (const auto& listener : listeners_) {
+            try {
+                listener(msg.getPartitionKey(), value);
+            } catch (const std::exception& exc) {
+                LOG_ERROR("Table view listener raised an exception: " << exc.what());
+            }
+        }
+    }
+}
+
+void TableViewImpl::readAllExistingMessages(Promise<Result, TableViewImplPtr> promise, long startTime,
+                                            long messagesRead) {
+    std::weak_ptr<TableViewImpl> weakSelf{shared_from_this()};
+    reader_->hasMessageAvailableAsync(
+        [weakSelf, promise, startTime, messagesRead](Result result, bool hasMessage) {
+            auto self = weakSelf.lock();
+            if (!self || result != ResultOk) {
+                promise.setFailed(result);
+                return;
+            }
+            if (hasMessage) {
+                Message msg;
+                self->reader_->readNextAsync(
+                    [weakSelf, promise, startTime, messagesRead](Result res, const Message& msg) {
+                        auto self = weakSelf.lock();
+                        if (!self || res != ResultOk) {
+                            promise.setFailed(res);
+                            LOG_ERROR("Start table view failed, reader msg for "
+                                      << self->topic_ << " error: " << strResult(res));
+                        } else {
+                            self->handleMessage(msg);
+                            auto tmpMessagesRead = messagesRead + 1;
+                            self->readAllExistingMessages(promise, startTime, tmpMessagesRead);
+                        }
+                    });
+            } else {
+                long endTime = TimeUtils::currentTimeMillis();
+                long durationMillis = endTime - startTime;
+                LOG_INFO("Started table view for " << self->topic_ << "Replayed: " << messagesRead
+                                                   << " message in " << durationMillis << " millis");
+                promise.setValue(self);
+                self->readTailMessage();
+            }
+        });
+}
+
+void TableViewImpl::readTailMessage() {
+    auto self = shared_from_this();
+    reader_->readNextAsync([self](Result result, const Message& msg) {
+        if (result == ResultOk) {
+            self->handleMessage(msg);
+            self->readTailMessage();
+        } else {
+            LOG_WARN("Reader " << self->topic_ << " was interrupted: " << result);
+        }
+    });
+}
+
+}  // namespace pulsar