From 3ca56e8b79540f6aa8d62e9cf95029311eabc707 Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Tue, 7 Feb 2023 17:50:21 +0800 Subject: [PATCH 1/2] [TableView-1] Add table view API --- include/pulsar/Client.h | 32 ++++++ include/pulsar/TableView.h | 129 ++++++++++++++++++++++++ include/pulsar/TableViewConfiguration.h | 73 ++++++++++++++ lib/TableViewConfiguration.cc | 50 +++++++++ lib/TableViewConfigurationImpl.h | 32 ++++++ 5 files changed, 316 insertions(+) create mode 100644 include/pulsar/TableView.h create mode 100644 include/pulsar/TableViewConfiguration.h create mode 100644 lib/TableViewConfiguration.cc create mode 100644 lib/TableViewConfigurationImpl.h 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..c74c6264 --- /dev/null +++ b/include/pulsar/TableViewConfiguration.h @@ -0,0 +1,73 @@ +/** + * 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 + +#include + +namespace pulsar { + +struct TableViewConfigurationImpl; + +/** + * Class specifying the configuration of a consumer. + */ +class PULSAR_PUBLIC TableViewConfiguration { + public: + TableViewConfiguration(); + ~TableViewConfiguration(); + TableViewConfiguration(const TableViewConfiguration&); + TableViewConfiguration& operator=(const TableViewConfiguration&); + + /** + * @return the schema information declared for this consumer + */ + const SchemaInfo& getSchemaInfo() const; + + /** + * 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. + * + * @param schemaInfo the schema definition object + */ + TableViewConfiguration& setSchemaInfo(const SchemaInfo& schemaInfo); + + /** + * @return subscriptionName + */ + const std::string& getSubscriptionName() const; + + /** + * Set the internal consumer subscription name of the {@link TableView}. + * + * @param subscriptionName the name of the subscription to the topic. + * Default value is reader-{random string}. + */ + TableViewConfiguration& setSubscriptionName(const std::string subscriptionName); + + private: + std::shared_ptr impl_; +}; +} // namespace pulsar +#endif /* PULSAR_TABLEVIEW_CONFIGURATION_H_ */ diff --git a/lib/TableViewConfiguration.cc b/lib/TableViewConfiguration.cc new file mode 100644 index 00000000..b3435741 --- /dev/null +++ b/lib/TableViewConfiguration.cc @@ -0,0 +1,50 @@ +/** + * 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 + +#include "TableViewConfigurationImpl.h" + +namespace pulsar { + +TableViewConfiguration::TableViewConfiguration() : impl_(std::make_shared()) {} + +TableViewConfiguration::~TableViewConfiguration() {} + +TableViewConfiguration::TableViewConfiguration(const TableViewConfiguration& x) : impl_(x.impl_) {} + +TableViewConfiguration& TableViewConfiguration::operator=(const TableViewConfiguration& x) { + impl_ = x.impl_; + return *this; +} + +const SchemaInfo& TableViewConfiguration::getSchemaInfo() const { return impl_->schemaInfo_; } + +TableViewConfiguration& TableViewConfiguration::setSchemaInfo(const SchemaInfo& schemaInfo) { + impl_->schemaInfo_ = schemaInfo; + return *this; +} + +const std::string& TableViewConfiguration::getSubscriptionName() const { return impl_->subscriptionName_; } + +TableViewConfiguration& TableViewConfiguration::setSubscriptionName(const std::string subscriptionName) { + impl_->subscriptionName_ = subscriptionName; + return *this; +} + +} // namespace pulsar diff --git a/lib/TableViewConfigurationImpl.h b/lib/TableViewConfigurationImpl.h new file mode 100644 index 00000000..7e27cb10 --- /dev/null +++ b/lib/TableViewConfigurationImpl.h @@ -0,0 +1,32 @@ +/** + * 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 LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ +#define LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ + +#include + +namespace pulsar { + +struct TableViewConfigurationImpl { + SchemaInfo schemaInfo_; + std::string subscriptionName_; +}; +} // namespace pulsar + +#endif /* LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ */ From 8b3aa57b672e61562c51d16d27b470a4098fee8a Mon Sep 17 00:00:00 2001 From: Baodi Shi Date: Tue, 14 Feb 2023 10:00:08 +0800 Subject: [PATCH 2/2] Use POD redesign tv config. --- include/pulsar/TableViewConfiguration.h | 51 ++++--------------------- lib/TableViewConfiguration.cc | 50 ------------------------ lib/TableViewConfigurationImpl.h | 32 ---------------- 3 files changed, 7 insertions(+), 126 deletions(-) delete mode 100644 lib/TableViewConfiguration.cc delete mode 100644 lib/TableViewConfigurationImpl.h diff --git a/include/pulsar/TableViewConfiguration.h b/include/pulsar/TableViewConfiguration.h index c74c6264..eef58818 100644 --- a/include/pulsar/TableViewConfiguration.h +++ b/include/pulsar/TableViewConfiguration.h @@ -22,52 +22,15 @@ #include #include -#include - namespace pulsar { -struct TableViewConfigurationImpl; - -/** - * Class specifying the configuration of a consumer. - */ -class PULSAR_PUBLIC TableViewConfiguration { - public: - TableViewConfiguration(); - ~TableViewConfiguration(); - TableViewConfiguration(const TableViewConfiguration&); - TableViewConfiguration& operator=(const TableViewConfiguration&); - - /** - * @return the schema information declared for this consumer - */ - const SchemaInfo& getSchemaInfo() const; - - /** - * 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. - * - * @param schemaInfo the schema definition object - */ - TableViewConfiguration& setSchemaInfo(const SchemaInfo& schemaInfo); - - /** - * @return subscriptionName - */ - const std::string& getSubscriptionName() const; - - /** - * Set the internal consumer subscription name of the {@link TableView}. - * - * @param subscriptionName the name of the subscription to the topic. - * Default value is reader-{random string}. - */ - TableViewConfiguration& setSubscriptionName(const std::string subscriptionName); - - private: - std::shared_ptr impl_; +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_ */ diff --git a/lib/TableViewConfiguration.cc b/lib/TableViewConfiguration.cc deleted file mode 100644 index b3435741..00000000 --- a/lib/TableViewConfiguration.cc +++ /dev/null @@ -1,50 +0,0 @@ -/** - * 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 - -#include "TableViewConfigurationImpl.h" - -namespace pulsar { - -TableViewConfiguration::TableViewConfiguration() : impl_(std::make_shared()) {} - -TableViewConfiguration::~TableViewConfiguration() {} - -TableViewConfiguration::TableViewConfiguration(const TableViewConfiguration& x) : impl_(x.impl_) {} - -TableViewConfiguration& TableViewConfiguration::operator=(const TableViewConfiguration& x) { - impl_ = x.impl_; - return *this; -} - -const SchemaInfo& TableViewConfiguration::getSchemaInfo() const { return impl_->schemaInfo_; } - -TableViewConfiguration& TableViewConfiguration::setSchemaInfo(const SchemaInfo& schemaInfo) { - impl_->schemaInfo_ = schemaInfo; - return *this; -} - -const std::string& TableViewConfiguration::getSubscriptionName() const { return impl_->subscriptionName_; } - -TableViewConfiguration& TableViewConfiguration::setSubscriptionName(const std::string subscriptionName) { - impl_->subscriptionName_ = subscriptionName; - return *this; -} - -} // namespace pulsar diff --git a/lib/TableViewConfigurationImpl.h b/lib/TableViewConfigurationImpl.h deleted file mode 100644 index 7e27cb10..00000000 --- a/lib/TableViewConfigurationImpl.h +++ /dev/null @@ -1,32 +0,0 @@ -/** - * 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 LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ -#define LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ - -#include - -namespace pulsar { - -struct TableViewConfigurationImpl { - SchemaInfo schemaInfo_; - std::string subscriptionName_; -}; -} // namespace pulsar - -#endif /* LIB_TABLEVIEW_CONFIGURATIONIMPL_H_ */