step-tv-1

include/pulsar/Client.h

@@ -29,6 +29,7 @@
 #include <pulsar/Reader.h>
 #include <pulsar/Result.h>
 #include <pulsar/Schema.h>
+#include <pulsar/TableView.h>
 #include <pulsar/defines.h>
 
 #include <string>
@@ -37,6 +38,7 @@ namespace pulsar {
 typedef std::function<void(Result, Producer)> CreateProducerCallback;
 typedef std::function<void(Result, Consumer)> SubscribeCallback;
 typedef std::function<void(Result, Reader)> ReaderCallback;
+typedef std::function<void(Result, TableView)> TableViewCallback;
 typedef std::function<void(Result, const std::vector<std::string>&)> GetPartitionsCallback;
 typedef std::function<void(Result)> CloseCallback;
 
@@ -301,6 +303,36 @@ class PULSAR_PUBLIC Client {
     void createReaderAsync(const std::string& topic, const MessageId& startMessageId,
                            const ReaderConfiguration& conf, ReaderCallback callback);
 
+    /**
+     * Create a table view with given {@code TableViewConfiguration} for specified topic.
+     *
+     * The TableView provides a key-value map view of a compacted topic. Messages without keys will
+     * be ignored.
+     *
+     * @param topic The name of the topic.
+     * @param conf The {@code TableViewConfiguration} object
+     * @param tableView The {@code TableView} object
+     * @return Returned when the TableView is successfully linked to the topic and the map is built from a
+     * message that already exists
+     */
+    Result createTableView(const std::string& topic, const TableViewConfiguration& conf,
+                           TableView& tableView);
+
+    /**
+     * Async create a table view with given {@code TableViewConfiguration} for specified topic.
+     *
+     * The TableView provides a key-value map view of a compacted topic. Messages without keys will
+     * be ignored.
+     *
+     * @param topic The name of the topic.
+     * @param conf The {@code TableViewConfiguration} object
+     * @param callBack
+     * The callback that is triggered when the TableView is successfully linked to the topic and the map is
+     * built from a message that already exists
+     */
+    void createTableViewAsync(const std::string& topic, const TableViewConfiguration& conf,
+                              TableViewCallback callBack);
+
     /**
      * Get the list of partitions for a given topic.
      *

include/pulsar/TableView.h

@@ -0,0 +1,129 @@
+/**
+ * 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.
+ */
+#ifndef TABEL_VIEW_HPP_
+#define TABEL_VIEW_HPP_
+
+#include <pulsar/Result.h>
+#include <pulsar/TableViewConfiguration.h>
+#include <pulsar/defines.h>
+
+#include <functional>
+#include <unordered_map>
+
+namespace pulsar {
+
+class TableViewImpl;
+
+typedef std::function<void(Result result)> ResultCallback;
+typedef std::function<void(const std::string& key, const std::string& value)> TableViewAction;
+/**
+ *
+ */
+class PULSAR_PUBLIC TableView {
+   public:
+    /**
+     * Construct an uninitialized tableView object
+     */
+    TableView();
+
+    /**
+     * Move the latest value associated with the key.
+     *
+     * Example:
+     *
+     * ```c++
+     * TableView view;
+     * std::string value;
+     * while (true) {
+     *     if (view.retrieveValue("key")) {
+     *         std::cout << "value is updated to: " << value;
+     *     } else {
+     *         // sleep for a while or print the message that value is not updated
+     *     }
+     * }
+     * ```
+     *
+     * @param key
+     * @param value the value associated with the key
+     * @return true if there is an associated value of the key, otherwise false
+     *
+     * NOTE: Once the value has been retrieved successfully, the associated value
+     * will be removed from the table view until next time the value is updated.
+     */
+    bool retrieveValue(const std::string& key, std::string& value);
+
+    /**
+     * It's similar with retrievedValue except the value is copied into `value`.
+     *
+     * @param key
+     * @param value the value associated with the key
+     * @return Whether the key exists in the table view.
+     */
+    bool getValue(const std::string& key, std::string& value) const;
+
+    /**
+     * Check if the key exists in the table view.
+     *
+     * @return true if the key exists in the table view
+     */
+    bool containsKey(const std::string& key) const;
+
+    /**
+     * Move the table view data into the unordered map.
+     */
+    std::unordered_map<std::string, std::string> snapshot();
+
+    /**
+     * Get the size of the elements.
+     */
+    std::size_t size() const;
+
+    /**
+     * Performs the given action for each entry in this map until all entries have been processed or the
+     * action throws an exception.
+     */
+    void forEach(TableViewAction action);
+
+    /**
+     * Performs the given action for each entry in this map until all entries have been processed and
+     * register the callback, which will be called each time a key-value pair is updated.
+     */
+    void forEachAndListen(TableViewAction action);
+
+    /**
+     * Asynchronously close the tableview and stop the broker to push more messages
+     */
+    void closeAsync(ResultCallback callback);
+
+    /**
+     * Close the consumer and stop the broker to push more messages
+     */
+    Result close();
+
+   private:
+    typedef std::shared_ptr<TableViewImpl> TableViewImplPtr;
+    TableViewImplPtr impl_;
+    explicit TableView(TableViewImplPtr);
+
+    friend class PulsarFriend;
+    friend class ClientImpl;
+};
+}  // namespace pulsar
+
+#endif /* TABEL_VIEW_HPP_ */

include/pulsar/TableViewConfiguration.h

@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+#ifndef PULSAR_TABLEVIEW_CONFIGURATION_H_
+#define PULSAR_TABLEVIEW_CONFIGURATION_H_
+
+#include <pulsar/Schema.h>
+#include <pulsar/defines.h>
+
+namespace pulsar {
+
+struct TableViewConfiguration {
+    // Declare the schema of the data that this table view will be accepting.
+    // The schema will be checked against the schema of the topic, and the
+    // table view creation will fail if it's not compatible.
+    SchemaInfo schemaInfo;
+    // The name of the subscription to the topic. Default value is reader-{random string}.
+    std::string subscriptionName;
+};
+}  // namespace pulsar
+#endif /* PULSAR_TABLEVIEW_CONFIGURATION_H_ */