diff --git a/include/pulsar/BatchReceivePolicy.h b/include/pulsar/BatchReceivePolicy.h index bc8b791d..f40c29d9 100644 --- a/include/pulsar/BatchReceivePolicy.h +++ b/include/pulsar/BatchReceivePolicy.h @@ -58,9 +58,10 @@ class PULSAR_PUBLIC BatchReceivePolicy { /** * - * @param maxNumMessage Max num message, if less than 0, it means no limit. - * @param maxNumBytes Max num bytes, if less than 0, it means no limit. - * @param timeoutMs If less than 0, it means no limit. + * @param maxNumMessage Max num message, a non-positive value means no limit. + * @param maxNumBytes Max num bytes, a non-positive value means no limit. + * @param timeoutMs The receive timeout, a non-positive value means no limit. + * @throws std::invalid_argument if all arguments are non-positive */ BatchReceivePolicy(int maxNumMessage, long maxNumBytes, long timeoutMs); diff --git a/include/pulsar/c/consumer_configuration.h b/include/pulsar/c/consumer_configuration.h index dea92f50..0282889a 100644 --- a/include/pulsar/c/consumer_configuration.h +++ b/include/pulsar/c/consumer_configuration.h @@ -89,12 +89,14 @@ typedef enum pulsar_consumer_regex_sub_mode_AllTopics = 2 } pulsar_consumer_regex_subscription_mode; +// Though any field could be non-positive, if all of them are non-positive, this policy will be treated as +// invalid typedef struct { - // Max num message, if less than 0, it means no limit. - int maxNumMessage; - // Max num bytes, if less than 0, it means no limit. + // Max num messages, a non-positive value means no limit. + int maxNumMessages; + // Max num bytes, a non-positive value means no limit. long maxNumBytes; - // If less than 0, it means no limit. + // The receive timeout, a non-positive value means no limit. long timeoutMs; } pulsar_consumer_batch_receive_policy_t; @@ -340,18 +342,33 @@ pulsar_consumer_configuration_get_regex_subscription_mode( /** * Set batch receive policy. * - * The default value: {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100} - * @param consumer_configuration the consumer conf object. - * @param maxNumMessage default: Max num message, if less than 0, it means no limit. - * @param maxNumBytes Max num bytes, if less than 0, it means no limit. - * @param timeoutMs If less than 0, it means no limit. + * @param [in] consumer_configuration a non-null pointer of the consumer configuration + * @param [in] batch_receive_policy + * @return 0 on success and -1 on failure + * + * The possible failed reasons are: + * - batch_receive_policy is null + * - batch_receive_policy points to an invalid policy */ -PULSAR_PUBLIC void pulsar_consumer_configuration_set_batch_receive_policy( +PULSAR_PUBLIC int pulsar_consumer_configuration_set_batch_receive_policy( pulsar_consumer_configuration_t *consumer_configuration, const pulsar_consumer_batch_receive_policy_t *batch_receive_policy); -PULSAR_PUBLIC pulsar_consumer_batch_receive_policy_t pulsar_consumer_configuration_get_batch_receive_policy( - pulsar_consumer_configuration_t *consumer_configuration); +/** + * Get the batch receive policy. + * + * @param [in] consumer_configuration a non-null pointer of the consumer configuration + * @param [out] batch_receive_policy + * + * If batch_receive_policy is not null, the instance that it points to will be updated to the batch receive + * policy of the consumer configuration. + * + * If the policy was never set before, the batch_receive_policy will be set with the following value: + * {maxNumMessage: -1, maxNumBytes: 10 * 1024 * 1024, timeoutMs: 100} + */ +PULSAR_PUBLIC void pulsar_consumer_configuration_get_batch_receive_policy( + pulsar_consumer_configuration_t *consumer_configuration, + pulsar_consumer_batch_receive_policy_t *batch_receive_policy); // const CryptoKeyReaderPtr getCryptoKeyReader() // diff --git a/lib/c/c_ConsumerConfiguration.cc b/lib/c/c_ConsumerConfiguration.cc index 6b808b65..dd9a3b8e 100644 --- a/lib/c/c_ConsumerConfiguration.cc +++ b/lib/c/c_ConsumerConfiguration.cc @@ -251,19 +251,31 @@ pulsar_consumer_regex_subscription_mode pulsar_consumer_configuration_get_regex_ consumer_configuration->consumerConfiguration.getRegexSubscriptionMode(); } -void pulsar_consumer_configuration_set_batch_receive_policy( +int pulsar_consumer_configuration_set_batch_receive_policy( pulsar_consumer_configuration_t *consumer_configuration, const pulsar_consumer_batch_receive_policy_t *batch_receive_policy_t) { - pulsar::BatchReceivePolicy batchReceivePolicy(batch_receive_policy_t->maxNumMessage, + if (!batch_receive_policy_t) { + return -1; + } + if (batch_receive_policy_t->maxNumMessages <= 0 && batch_receive_policy_t->maxNumBytes <= 0 && + batch_receive_policy_t->timeoutMs <= 0) { + return -1; + } + pulsar::BatchReceivePolicy batchReceivePolicy(batch_receive_policy_t->maxNumMessages, batch_receive_policy_t->maxNumBytes, batch_receive_policy_t->timeoutMs); consumer_configuration->consumerConfiguration.setBatchReceivePolicy(batchReceivePolicy); + return 0; } -pulsar_consumer_batch_receive_policy_t pulsar_consumer_configuration_get_batch_receive_policy( - pulsar_consumer_configuration_t *consumer_configuration) { +void pulsar_consumer_configuration_get_batch_receive_policy( + pulsar_consumer_configuration_t *consumer_configuration, pulsar_consumer_batch_receive_policy_t *policy) { + if (!policy) { + return; + } pulsar::BatchReceivePolicy batchReceivePolicy = consumer_configuration->consumerConfiguration.getBatchReceivePolicy(); - return {batchReceivePolicy.getMaxNumMessages(), batchReceivePolicy.getMaxNumBytes(), - batchReceivePolicy.getTimeoutMs()}; + policy->maxNumMessages = batchReceivePolicy.getMaxNumMessages(); + policy->maxNumBytes = batchReceivePolicy.getMaxNumBytes(); + policy->timeoutMs = batchReceivePolicy.getTimeoutMs(); } diff --git a/tests/c/c_ConsumerConfigurationTest.cc b/tests/c/c_ConsumerConfigurationTest.cc index 629c6f37..450b6a17 100644 --- a/tests/c/c_ConsumerConfigurationTest.cc +++ b/tests/c/c_ConsumerConfigurationTest.cc @@ -43,13 +43,34 @@ TEST(C_ConsumerConfigurationTest, testCApiConfig) { ASSERT_EQ(pulsar_consumer_configuration_get_regex_subscription_mode(consumer_conf), pulsar_consumer_regex_sub_mode_NonPersistentOnly); - pulsar_consumer_batch_receive_policy_t batch_receive_policy{10, 1000, 1000}; - pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, &batch_receive_policy); - pulsar_consumer_batch_receive_policy_t get_batch_receive_policy = - pulsar_consumer_configuration_get_batch_receive_policy(consumer_conf); - ASSERT_EQ(get_batch_receive_policy.maxNumMessage, 10); - ASSERT_EQ(get_batch_receive_policy.maxNumBytes, 1000); - ASSERT_EQ(get_batch_receive_policy.timeoutMs, 1000); + pulsar_consumer_batch_receive_policy_t batch_receive_policy; + pulsar_consumer_configuration_get_batch_receive_policy(consumer_conf, &batch_receive_policy); + ASSERT_EQ(batch_receive_policy.maxNumMessages, -1); + ASSERT_EQ(batch_receive_policy.maxNumBytes, 10 * 1024 * 1024L); + ASSERT_EQ(batch_receive_policy.timeoutMs, 100L); + + pulsar_consumer_batch_receive_policy_t new_batch_receive_policy{-1, -1, -1}; + ASSERT_EQ(-1, pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, NULL)); + ASSERT_EQ( + -1, pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, &new_batch_receive_policy)); + + new_batch_receive_policy.maxNumMessages = 100; + ASSERT_EQ( + 0, pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, &new_batch_receive_policy)); + pulsar_consumer_configuration_get_batch_receive_policy(consumer_conf, &batch_receive_policy); + ASSERT_EQ(batch_receive_policy.maxNumMessages, 100); + + new_batch_receive_policy.maxNumBytes = 100L * 1024 * 1024 * 1024; + ASSERT_EQ( + 0, pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, &new_batch_receive_policy)); + pulsar_consumer_configuration_get_batch_receive_policy(consumer_conf, &batch_receive_policy); + ASSERT_EQ(batch_receive_policy.maxNumBytes, 100L * 1024 * 1024 * 1024); + + new_batch_receive_policy.timeoutMs = 365L * 24 * 3600 * 1000; + ASSERT_EQ( + 0, pulsar_consumer_configuration_set_batch_receive_policy(consumer_conf, &new_batch_receive_policy)); + pulsar_consumer_configuration_get_batch_receive_policy(consumer_conf, &batch_receive_policy); + ASSERT_EQ(batch_receive_policy.timeoutMs, 365L * 24 * 3600 * 1000); pulsar_consumer_configuration_free(consumer_conf); }