feat: Support table view for C client. (apache#294)

shibd · web-flow · commit 51d11db0c6ed · 2023-07-07T16:22:35.000+08:00
diff --git a/include/pulsar/TableView.h b/include/pulsar/TableView.h
@@ -51,7 +51,7 @@ class PULSAR_PUBLIC TableView {
      * TableView view;
      * std::string value;
      * while (true) {
-     *     if (view.retrieveValue("key")) {
+     *     if (view.retrieveValue("key", value)) {
      *         std::cout << "value is updated to: " << value;
      *     } else {
      *         // sleep for a while or print the message that value is not updated
diff --git a/include/pulsar/c/client.h b/include/pulsar/c/client.h
@@ -30,6 +30,8 @@
 #include <pulsar/c/reader_configuration.h>
 #include <pulsar/c/result.h>
 #include <pulsar/c/string_list.h>
+#include <pulsar/c/table_view.h>
+#include <pulsar/c/table_view_configuration.h>
 #include <pulsar/defines.h>
 
 #ifdef __cplusplus
@@ -47,6 +49,7 @@ typedef void (*pulsar_create_producer_callback)(pulsar_result result, pulsar_pro
 
 typedef void (*pulsar_subscribe_callback)(pulsar_result result, pulsar_consumer_t *consumer, void *ctx);
 typedef void (*pulsar_reader_callback)(pulsar_result result, pulsar_reader_t *reader, void *ctx);
+typedef void (*pulsar_table_view_callback)(pulsar_result result, pulsar_table_view_t *tableView, void *ctx);
 typedef void (*pulsar_get_partitions_callback)(pulsar_result result, pulsar_string_list_t *partitions,
                                                void *ctx);
 
@@ -172,6 +175,54 @@ PULSAR_PUBLIC void pulsar_client_create_reader_async(pulsar_client_t *client, co
                                                      const pulsar_message_id_t *startMessageId,
                                                      pulsar_reader_configuration_t *conf,
                                                      pulsar_reader_callback callback, void *ctx);
+/**
+ * Create a table view with given {@code table_view_configuration} for specified topic.
+ *
+ * The TableView provides a key-value map view of a compacted topic. Messages without keys will
+ * be ignored.
+ *
+ * NOTE:
+ * When the result in the callback is `ResultOk`, `*c_tableView` will point to the memory that
+ * is allocated internally. You have to call `pulsar_table_view_free` to free it.
+ *
+ * Example:
+ * ```c
+ * pulsar_table_view_configuration_t *table_view_conf = pulsar_table_view_configuration_create();
+ * pulsar_table_view_configuration_set_subscription_name(table_view_conf, sub_name);
+ * pulsar_table_view_t *table_view;
+ * pulsar_result result = pulsar_client_create_table_view(client, topic_name, table_view_conf, &table_view);
+ *
+ * // do something...
+ *
+ * pulsar_table_view_close(table_view);
+ * pulsar_table_view_free(table_view);
+ * pulsar_table_view_configuration_free(table_view_conf);
+ *
+ * ```
+ *
+ * @param topic The name of the topic.
+ * @param conf The {@code table_view_configuration} pointer.
+ * @param c_tableView The pointer of the table_view pointer
+ * @return Returned when the table_view is successfully linked to the topic and the map is built from a
+ * message that already exists.
+ */
+PULSAR_PUBLIC pulsar_result pulsar_client_create_table_view(pulsar_client_t *client, const char *topic,
+                                                            pulsar_table_view_configuration_t *conf,
+                                                            pulsar_table_view_t **c_tableView);
+
+/**
+ * Async Create a table view with given {@code table_view_configuration} for specified topic.
+ * @param topic The name of the topic.
+ * @param conf The {@code table_view_configuration} pointer.
+ * @param callback
+ * 1. When the result in the callback is `ResultOk`, `tableView` in the callback will point to the memory that
+ * is allocated internally. You have to call `pulsar_table_view_free` to free it.
+ * 2. If the result in the callback is not `ResultOk`, `tableView` in the callback will be nullptr.
+ * @param ctx
+ */
+PULSAR_PUBLIC void pulsar_client_create_table_view_async(pulsar_client_t *client, const char *topic,
+                                                         pulsar_table_view_configuration_t *conf,
+                                                         pulsar_table_view_callback callback, void *ctx);
 
 PULSAR_PUBLIC pulsar_result pulsar_client_get_topic_partitions(pulsar_client_t *client, const char *topic,
                                                                pulsar_string_list_t **partitions);
diff --git a/include/pulsar/c/table_view.h b/include/pulsar/c/table_view.h
@@ -0,0 +1,141 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <pulsar/defines.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <pulsar/c/message.h>
+#include <pulsar/c/messages.h>
+#include <pulsar/c/result.h>
+#include <stdint.h>
+
+typedef struct _pulsar_table_view pulsar_table_view_t;
+
+typedef void (*pulsar_table_view_action)(const char *key, const void *value, size_t value_size, void *ctx);
+typedef void (*pulsar_result_callback)(pulsar_result, void *);
+
+/**
+ * Move the latest value associated with the key.
+ *
+ * NOTE:
+ * 1. Once the value has been retrieved successfully,
+ * the associated value will be removed from the table view until next time the value is updated.
+ * 2. Once the value has been retrieved successfully, `*value` will point to the memory that is allocated
+ * internally. You have to call `free(value)` to free it.
+ *
+ * Example:
+ *
+ * ```c
+ * pulsar_table_view_t *table_view;
+ * void* value;
+ * size_t value_size;
+ * while (true) {
+ *     if (pulsar_table_view_retrieve_value(table_view, "key", &value, &value_size)) {
+ *         for (size_t i = 0; i < value_size; i++) {
+ *             printf("0x%02x%c", ((char*) value)[i], (i + 1 == value_size) ? '\n': ' ');
+ *         }
+ *     } else {
+ *         // sleep for a while or print the message that value is not updated
+ *     }
+ * }
+ * free(value);
+ * ```
+ *
+ * @param table_view
+ * @param key
+ * @param value the value associated with the key
+ * @return true if there is an associated value of the key, otherwise false
+ */
+PULSAR_PUBLIC bool pulsar_table_view_retrieve_value(pulsar_table_view_t *table_view, const char *key,
+                                                    void **value, size_t *value_size);
+
+/**
+ * It's similar with `pulsar_table_view_retrieve_value` except the associated value not will be removed from
+ * the table view.
+ *
+ * NOTE:
+ * Once the value has been get successfully, `*value` will point to the memory that is allocated internally.
+ * You have to call `free(value)` to free it.
+ *
+ * @param table_view
+ * @param key
+ * @param value the value associated with the key
+ * @return true if there is an associated value of the key, otherwise false
+ */
+PULSAR_PUBLIC bool pulsar_table_view_get_value(pulsar_table_view_t *table_view, const char *key, void **value,
+                                               size_t *value_size);
+
+/**
+ * Check if the key exists in the table view.
+ * @param table_view
+ * @param key
+ * @return true if the key exists in the table view
+ */
+PULSAR_PUBLIC bool pulsar_table_view_contain_key(pulsar_table_view_t *table_view, const char *key);
+
+/**
+ * Get the size of the elements.
+ * @param table_view
+ * @return
+ */
+PULSAR_PUBLIC int pulsar_table_view_size(pulsar_table_view_t *table_view);
+
+/**
+ * Performs the given action for each entry in this map until all entries have been processed or the
+ * action throws an exception.
+ */
+PULSAR_PUBLIC void pulsar_table_view_for_each(pulsar_table_view_t *table_view,
+                                              pulsar_table_view_action action, void *ctx);
+
+/**
+ * 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.
+ */
+PULSAR_PUBLIC void pulsar_table_view_for_each_add_listen(pulsar_table_view_t *table_view,
+                                                         pulsar_table_view_action action, void *ctx);
+
+/**
+ * Free the table view.
+ * @param table_view
+ */
+PULSAR_PUBLIC void pulsar_table_view_free(pulsar_table_view_t *table_view);
+
+/**
+ * Close the table view and stop the broker to push more messages
+ * @param table_view
+ * @return
+ */
+PULSAR_PUBLIC pulsar_result pulsar_table_view_close(pulsar_table_view_t *table_view);
+
+/**
+ * Async close the table view and stop the broker to push more messages
+ * @param table_view
+ * @param callback
+ * @param ctx
+ */
+PULSAR_PUBLIC void pulsar_table_view_close_async(pulsar_table_view_t *table_view,
+                                                 pulsar_result_callback callback, void *ctx);
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/include/pulsar/c/table_view_configuration.h b/include/pulsar/c/table_view_configuration.h
@@ -0,0 +1,48 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/defines.h>
+
+#include "producer_configuration.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct _pulsar_table_view_configuration pulsar_table_view_configuration_t;
+
+PULSAR_PUBLIC pulsar_table_view_configuration_t *pulsar_table_view_configuration_create();
+
+PULSAR_PUBLIC void pulsar_table_view_configuration_free(pulsar_table_view_configuration_t *conf);
+
+PULSAR_PUBLIC void pulsar_table_view_configuration_set_schema_info(
+    pulsar_table_view_configuration_t *table_view_configuration_t, pulsar_schema_type schemaType,
+    const char *name, const char *schema, pulsar_string_map_t *properties);
+
+PULSAR_PUBLIC void pulsar_table_view_configuration_set_subscription_name(
+    pulsar_table_view_configuration_t *table_view_configuration_t, const char *subscription_name);
+
+PULSAR_PUBLIC const char *pulsar_table_view_configuration_get_subscription_name(
+    pulsar_table_view_configuration_t *table_view_configuration_t);
+
+#ifdef __cplusplus
+}
+#endif
diff --git a/lib/c/c_Client.cc b/lib/c/c_Client.cc
@@ -192,6 +192,34 @@ void pulsar_client_create_reader_async(pulsar_client_t *client, const char *topi
         std::bind(&handle_reader_callback, std::placeholders::_1, std::placeholders::_2, callback, ctx));
 }
 
+pulsar_result pulsar_client_create_table_view(pulsar_client_t *client, const char *topic,
+                                              pulsar_table_view_configuration_t *conf,
+                                              pulsar_table_view_t **c_tableView) {
+    pulsar::TableView tableView;
+    pulsar::Result res = client->client->createTableView(topic, conf->tableViewConfiguration, tableView);
+    if (res == pulsar::ResultOk) {
+        (*c_tableView) = new pulsar_table_view_t;
+        (*c_tableView)->tableView = std::move(tableView);
+        return pulsar_result_Ok;
+    }
+    return (pulsar_result)res;
+}
+
+void pulsar_client_create_table_view_async(pulsar_client_t *client, const char *topic,
+                                           pulsar_table_view_configuration_t *conf,
+                                           pulsar_table_view_callback callback, void *ctx) {
+    client->client->createTableViewAsync(topic, conf->tableViewConfiguration,
+                                         [callback, ctx](pulsar::Result result, pulsar::TableView tableView) {
+                                             if (result == pulsar::ResultOk) {
+                                                 auto *c_tableView = new pulsar_table_view_t;
+                                                 c_tableView->tableView = std::move(tableView);
+                                                 callback((pulsar_result)result, c_tableView, ctx);
+                                             } else {
+                                                 callback((pulsar_result)result, NULL, ctx);
+                                             }
+                                         });
+}
+
 pulsar_result pulsar_client_get_topic_partitions(pulsar_client_t *client, const char *topic,
                                                  pulsar_string_list_t **partitions) {
     std::vector<std::string> partitionsList;
diff --git a/lib/c/c_TableView.cc b/lib/c/c_TableView.cc
@@ -0,0 +1,89 @@
+/**
+ * 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/c/table_view.h>
+#include <string.h>
+
+#include "c_structs.h"
+
+static void *malloc_and_copy(const char *s, size_t slen) {
+    void *result = (void *)malloc(slen);
+    if (result == NULL) {
+        abort();
+    }
+    memcpy(result, s, slen);
+    return result;
+}
+
+bool pulsar_table_view_retrieve_value(pulsar_table_view_t *table_view, const char *key, void **value,
+                                      size_t *value_size) {
+    std::string v;
+    bool result = table_view->tableView.retrieveValue(key, v);
+    if (result) {
+        *value = malloc_and_copy(v.c_str(), v.size());
+        *value_size = v.size();
+    }
+    return result;
+}
+
+bool pulsar_table_view_get_value(pulsar_table_view_t *table_view, const char *key, void **value,
+                                 size_t *value_size) {
+    std::string v;
+    bool result = table_view->tableView.getValue(key, v);
+    if (result) {
+        *value = malloc_and_copy(v.c_str(), v.size());
+        *value_size = v.size();
+    }
+    return result;
+}
+
+bool pulsar_table_view_contain_key(pulsar_table_view_t *table_view, const char *key) {
+    return table_view->tableView.containsKey(key);
+}
+
+int pulsar_table_view_size(pulsar_table_view_t *table_view) { return table_view->tableView.size(); }
+
+void pulsar_table_view_for_each(pulsar_table_view_t *table_view, pulsar_table_view_action action, void *ctx) {
+    table_view->tableView.forEach([action, ctx](const std::string &key, const std::string &value) {
+        if (action) {
+            action(key.c_str(), value.c_str(), value.size(), ctx);
+        }
+    });
+}
+
+void pulsar_table_view_for_each_add_listen(pulsar_table_view_t *table_view, pulsar_table_view_action action,
+                                           void *ctx) {
+    table_view->tableView.forEachAndListen([action, ctx](const std::string &key, const std::string &value) {
+        if (action) {
+            action(key.c_str(), value.c_str(), value.size(), ctx);
+        }
+    });
+}
+
+void pulsar_table_view_free(pulsar_table_view_t *table_view) { delete table_view; }
+
+pulsar_result pulsar_table_view_close(pulsar_table_view_t *table_view) {
+    return (pulsar_result)table_view->tableView.close();
+}
+
+void pulsar_table_view_close_async(pulsar_table_view_t *table_view, pulsar_result_callback callback,
+                                   void *ctx) {
+    table_view->tableView.closeAsync(
+        [callback, ctx](pulsar::Result result) { return handle_result_callback(result, callback, ctx); });
+}
diff --git a/lib/c/c_TableViewConfiguration.cc b/lib/c/c_TableViewConfiguration.cc
diff --git a/lib/c/c_structs.h b/lib/c/c_structs.h
diff --git a/tests/c/c_TableViewTest.cc b/tests/c/c_TableViewTest.cc
