diff --git a/include/pulsar/Client.h b/include/pulsar/Client.h index c189a200..a8349eb5 100644 --- a/include/pulsar/Client.h +++ b/include/pulsar/Client.h @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -37,6 +38,7 @@ namespace pulsar { typedef std::function CreateProducerCallback; typedef std::function SubscribeCallback; typedef std::function ReaderCallback; +typedef std::function TableViewCallback; typedef std::function&)> GetPartitionsCallback; typedef std::function 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. * diff --git a/include/pulsar/TableView.h b/include/pulsar/TableView.h new file mode 100644 index 00000000..de5a7bcf --- /dev/null +++ b/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 +#include +#include + +#include +#include + +namespace pulsar { + +class TableViewImpl; + +typedef std::function ResultCallback; +typedef std::function 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 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 TableViewImplPtr; + TableViewImplPtr impl_; + explicit TableView(TableViewImplPtr); + + friend class PulsarFriend; + friend class ClientImpl; +}; +} // namespace pulsar + +#endif /* TABEL_VIEW_HPP_ */ diff --git a/include/pulsar/TableViewConfiguration.h b/include/pulsar/TableViewConfiguration.h new file mode 100644 index 00000000..eef58818 --- /dev/null +++ b/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 +#include + +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_ */