Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion include/pulsar/TableView.h
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
51 changes: 51 additions & 0 deletions include/pulsar/c/client.h
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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);

Expand Down Expand Up @@ -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);
Expand Down
141 changes: 141 additions & 0 deletions include/pulsar/c/table_view.h
Original file line number Diff line number Diff line change
@@ -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
48 changes: 48 additions & 0 deletions include/pulsar/c/table_view_configuration.h
Original file line number Diff line number Diff line change
@@ -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
28 changes: 28 additions & 0 deletions lib/c/c_Client.cc
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
89 changes: 89 additions & 0 deletions lib/c/c_TableView.cc
Original file line number Diff line number Diff line change
@@ -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); });
}
Loading