in src/ConsumerConfig.cc [92:327]
void ConsumerConfig::InitConfig(std::shared_ptr<ThreadSafeDeferred> deferred,
const Napi::Object &consumerConfig, pulsar_message_listener messageListener) {
if (consumerConfig.Has(CFG_TOPIC) && consumerConfig.Get(CFG_TOPIC).IsString()) {
this->topic = consumerConfig.Get(CFG_TOPIC).ToString().Utf8Value();
}
if (consumerConfig.Has(CFG_TOPICS) && consumerConfig.Get(CFG_TOPICS).IsArray()) {
auto arr = consumerConfig.Get(CFG_TOPICS).As<Napi::Array>();
for (uint32_t i = 0; i < arr.Length(); i++) {
if (arr.Get(i).IsString()) {
this->topics.emplace_back(arr.Get(i).ToString().Utf8Value());
}
}
}
if (consumerConfig.Has(CFG_TOPICS_PATTERN) && consumerConfig.Get(CFG_TOPICS_PATTERN).IsString()) {
this->topicsPattern = consumerConfig.Get(CFG_TOPICS_PATTERN).ToString().Utf8Value();
}
if (this->topic.empty() && this->topics.size() == 0 && this->topicsPattern.empty()) {
deferred->Reject(
std::string("Topic, topics or topicsPattern is required and must be specified as a string when "
"creating consumer"));
return;
}
if (consumerConfig.Has(CFG_SUBSCRIPTION) && consumerConfig.Get(CFG_SUBSCRIPTION).IsString()) {
this->subscription = consumerConfig.Get(CFG_SUBSCRIPTION).ToString().Utf8Value();
}
if (subscription.empty()) {
deferred->Reject(
std::string("Subscription is required and must be specified as a string when creating consumer"));
return;
}
if (consumerConfig.Has(CFG_SUBSCRIPTION_TYPE) && consumerConfig.Get(CFG_SUBSCRIPTION_TYPE).IsString()) {
std::string subscriptionType = consumerConfig.Get(CFG_SUBSCRIPTION_TYPE).ToString().Utf8Value();
if (SUBSCRIPTION_TYPE.count(subscriptionType)) {
pulsar_consumer_configuration_set_consumer_type(this->cConsumerConfig.get(),
SUBSCRIPTION_TYPE.at(subscriptionType));
}
}
if (consumerConfig.Has(CFG_INIT_POSITION) && consumerConfig.Get(CFG_INIT_POSITION).IsString()) {
std::string initPosition = consumerConfig.Get(CFG_INIT_POSITION).ToString().Utf8Value();
if (INIT_POSITION.count(initPosition)) {
pulsar_consumer_set_subscription_initial_position(this->cConsumerConfig.get(),
INIT_POSITION.at(initPosition));
}
}
if (consumerConfig.Has(CFG_REGEX_SUBSCRIPTION_MODE) &&
consumerConfig.Get(CFG_REGEX_SUBSCRIPTION_MODE).IsString()) {
std::string regexSubscriptionMode =
consumerConfig.Get(CFG_REGEX_SUBSCRIPTION_MODE).ToString().Utf8Value();
if (REGEX_SUBSCRIPTION_MODE.count(regexSubscriptionMode)) {
pulsar_consumer_configuration_set_regex_subscription_mode(
this->cConsumerConfig.get(), REGEX_SUBSCRIPTION_MODE.at(regexSubscriptionMode));
}
}
if (consumerConfig.Has(CFG_CONSUMER_NAME) && consumerConfig.Get(CFG_CONSUMER_NAME).IsString()) {
std::string consumerName = consumerConfig.Get(CFG_CONSUMER_NAME).ToString().Utf8Value();
if (!consumerName.empty())
pulsar_consumer_set_consumer_name(this->cConsumerConfig.get(), consumerName.c_str());
}
if (consumerConfig.Has(CFG_ACK_TIMEOUT) && consumerConfig.Get(CFG_ACK_TIMEOUT).IsNumber()) {
this->ackTimeoutMs = consumerConfig.Get(CFG_ACK_TIMEOUT).ToNumber().Int64Value();
if (this->ackTimeoutMs != 0 && ackTimeoutMs < MIN_ACK_TIMEOUT_MILLIS) {
std::string msg("Ack timeout should be 0 or greater than or equal to " +
std::to_string(MIN_ACK_TIMEOUT_MILLIS));
deferred->Reject(msg);
return;
}
pulsar_consumer_set_unacked_messages_timeout_ms(this->cConsumerConfig.get(), this->ackTimeoutMs);
}
if (consumerConfig.Has(CFG_NACK_REDELIVER_TIMEOUT) &&
consumerConfig.Get(CFG_NACK_REDELIVER_TIMEOUT).IsNumber()) {
this->nAckRedeliverTimeoutMs = consumerConfig.Get(CFG_NACK_REDELIVER_TIMEOUT).ToNumber().Int64Value();
if (nAckRedeliverTimeoutMs < 0) {
std::string msg("NAck timeout should be greater than or equal to zero");
deferred->Reject(msg);
return;
}
pulsar_configure_set_negative_ack_redelivery_delay_ms(this->cConsumerConfig.get(),
this->nAckRedeliverTimeoutMs);
}
if (consumerConfig.Has(CFG_RECV_QUEUE) && consumerConfig.Get(CFG_RECV_QUEUE).IsNumber()) {
int32_t receiverQueueSize = consumerConfig.Get(CFG_RECV_QUEUE).ToNumber().Int32Value();
if (receiverQueueSize >= 0) {
pulsar_consumer_configuration_set_receiver_queue_size(this->cConsumerConfig.get(), receiverQueueSize);
}
}
if (consumerConfig.Has(CFG_RECV_QUEUE_ACROSS_PARTITIONS) &&
consumerConfig.Get(CFG_RECV_QUEUE_ACROSS_PARTITIONS).IsNumber()) {
int32_t receiverQueueSizeAcrossPartitions =
consumerConfig.Get(CFG_RECV_QUEUE_ACROSS_PARTITIONS).ToNumber().Int32Value();
if (receiverQueueSizeAcrossPartitions >= 0) {
pulsar_consumer_set_max_total_receiver_queue_size_across_partitions(this->cConsumerConfig.get(),
receiverQueueSizeAcrossPartitions);
}
}
if (consumerConfig.Has(CFG_SCHEMA) && consumerConfig.Get(CFG_SCHEMA).IsObject()) {
SchemaInfo *schemaInfo = new SchemaInfo(consumerConfig.Get(CFG_SCHEMA).ToObject());
schemaInfo->SetConsumerSchema(this->cConsumerConfig);
delete schemaInfo;
}
if (consumerConfig.Has(CFG_PROPS) && consumerConfig.Get(CFG_PROPS).IsObject()) {
Napi::Object propObj = consumerConfig.Get(CFG_PROPS).ToObject();
Napi::Array arr = propObj.GetPropertyNames();
int size = arr.Length();
for (int i = 0; i < size; i++) {
std::string key = arr.Get(i).ToString().Utf8Value();
std::string value = propObj.Get(key).ToString().Utf8Value();
pulsar_consumer_configuration_set_property(this->cConsumerConfig.get(), key.c_str(), value.c_str());
}
}
if (consumerConfig.Has(CFG_LISTENER) && consumerConfig.Get(CFG_LISTENER).IsFunction()) {
this->listener = new MessageListenerCallback();
Napi::ThreadSafeFunction callback = Napi::ThreadSafeFunction::New(
consumerConfig.Env(), consumerConfig.Get(CFG_LISTENER).As<Napi::Function>(), "Listener Callback", 1,
1, (void *)NULL, FinalizeListenerCallback, listener);
this->listener->callback = std::move(callback);
pulsar_consumer_configuration_set_message_listener(this->cConsumerConfig.get(), messageListener,
this->listener);
}
if (consumerConfig.Has(CFG_READ_COMPACTED) && consumerConfig.Get(CFG_READ_COMPACTED).IsBoolean()) {
bool readCompacted = consumerConfig.Get(CFG_READ_COMPACTED).ToBoolean();
if (readCompacted) {
pulsar_consumer_set_read_compacted(this->cConsumerConfig.get(), 1);
}
}
if (consumerConfig.Has(CFG_PRIVATE_KEY_PATH) && consumerConfig.Get(CFG_PRIVATE_KEY_PATH).IsString()) {
std::string publicKeyPath = "";
std::string privateKeyPath = consumerConfig.Get(CFG_PRIVATE_KEY_PATH).ToString().Utf8Value();
pulsar_consumer_configuration_set_default_crypto_key_reader(
this->cConsumerConfig.get(), publicKeyPath.c_str(), privateKeyPath.c_str());
if (consumerConfig.Has(CFG_CRYPTO_FAILURE_ACTION) &&
consumerConfig.Get(CFG_CRYPTO_FAILURE_ACTION).IsString()) {
std::string cryptoFailureAction = consumerConfig.Get(CFG_CRYPTO_FAILURE_ACTION).ToString().Utf8Value();
if (CONSUMER_CRYPTO_FAILURE_ACTION.count(cryptoFailureAction)) {
pulsar_consumer_configuration_set_crypto_failure_action(
this->cConsumerConfig.get(), CONSUMER_CRYPTO_FAILURE_ACTION.at(cryptoFailureAction));
}
}
}
if (consumerConfig.Has(CFG_MAX_PENDING_CHUNKED_MESSAGE) &&
consumerConfig.Get(CFG_MAX_PENDING_CHUNKED_MESSAGE).IsNumber()) {
int32_t maxPendingChunkedMessage =
consumerConfig.Get(CFG_MAX_PENDING_CHUNKED_MESSAGE).ToNumber().Int32Value();
if (maxPendingChunkedMessage >= 0) {
pulsar_consumer_configuration_set_max_pending_chunked_message(this->cConsumerConfig.get(),
maxPendingChunkedMessage);
}
}
if (consumerConfig.Has(CFG_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL) &&
consumerConfig.Get(CFG_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL).IsBoolean()) {
bool autoAckOldestChunkedMessageOnQueueFull =
consumerConfig.Get(CFG_AUTO_ACK_OLDEST_CHUNKED_MESSAGE_ON_QUEUE_FULL).ToBoolean();
pulsar_consumer_configuration_set_auto_ack_oldest_chunked_message_on_queue_full(
this->cConsumerConfig.get(), autoAckOldestChunkedMessageOnQueueFull);
}
if (consumerConfig.Has(CFG_BATCH_INDEX_ACK_ENABLED) &&
consumerConfig.Get(CFG_BATCH_INDEX_ACK_ENABLED).IsBoolean()) {
bool batchIndexAckEnabled = consumerConfig.Get(CFG_BATCH_INDEX_ACK_ENABLED).ToBoolean();
pulsar_consumer_configuration_set_batch_index_ack_enabled(this->cConsumerConfig.get(),
batchIndexAckEnabled);
}
if (consumerConfig.Has(CFG_DEAD_LETTER_POLICY) && consumerConfig.Get(CFG_DEAD_LETTER_POLICY).IsObject()) {
pulsar_consumer_config_dead_letter_policy_t dlq_policy{};
Napi::Object dlqPolicyObject = consumerConfig.Get(CFG_DEAD_LETTER_POLICY).ToObject();
std::string dlq_topic_str;
std::string init_subscription_name;
if (dlqPolicyObject.Has(CFG_DLQ_POLICY_TOPIC) && dlqPolicyObject.Get(CFG_DLQ_POLICY_TOPIC).IsString()) {
dlq_topic_str = dlqPolicyObject.Get(CFG_DLQ_POLICY_TOPIC).ToString().Utf8Value();
dlq_policy.dead_letter_topic = dlq_topic_str.c_str();
}
if (dlqPolicyObject.Has(CFG_DLQ_POLICY_MAX_REDELIVER_COUNT) &&
dlqPolicyObject.Get(CFG_DLQ_POLICY_MAX_REDELIVER_COUNT).IsNumber()) {
dlq_policy.max_redeliver_count =
dlqPolicyObject.Get(CFG_DLQ_POLICY_MAX_REDELIVER_COUNT).ToNumber().Int32Value();
}
if (dlqPolicyObject.Has(CFG_DLQ_POLICY_INIT_SUB_NAME) &&
dlqPolicyObject.Get(CFG_DLQ_POLICY_INIT_SUB_NAME).IsString()) {
init_subscription_name = dlqPolicyObject.Get(CFG_DLQ_POLICY_INIT_SUB_NAME).ToString().Utf8Value();
dlq_policy.initial_subscription_name = init_subscription_name.c_str();
}
pulsar_consumer_configuration_set_dlq_policy(this->cConsumerConfig.get(), &dlq_policy);
}
if (consumerConfig.Has(CFG_BATCH_RECEIVE_POLICY) &&
consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).IsObject()) {
Napi::Object propObj = consumerConfig.Get(CFG_BATCH_RECEIVE_POLICY).ToObject();
int maxNumMessages = -1;
if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES) &&
propObj.Get(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES).IsNumber()) {
maxNumMessages = propObj.Get(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_MESSAGES).ToNumber().Int32Value();
}
int maxNumBytes = 10 * 1024 * 1024;
if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_BYTES) &&
propObj.Get(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_BYTES).IsNumber()) {
maxNumBytes = propObj.Get(CFG_BATCH_RECEIVE_POLICY_MAX_NUM_BYTES).ToNumber().Int64Value();
}
int timeoutMs = 100;
if (propObj.Has(CFG_BATCH_RECEIVE_POLICY_TIMEOUT_MS) &&
propObj.Get(CFG_BATCH_RECEIVE_POLICY_TIMEOUT_MS).IsNumber()) {
timeoutMs = propObj.Get(CFG_BATCH_RECEIVE_POLICY_TIMEOUT_MS).ToNumber().Int64Value();
}
if (maxNumMessages <= 0 && maxNumBytes <= 0 && timeoutMs <= 0) {
std::string msg("At least one of maxNumMessages, maxNumBytes and timeoutMs must be specified.");
deferred->Reject(msg);
return;
}
pulsar_consumer_batch_receive_policy_t batch_receive_policy{maxNumMessages, maxNumBytes, timeoutMs};
int result = pulsar_consumer_configuration_set_batch_receive_policy(this->cConsumerConfig.get(),
&batch_receive_policy);
if (result == -1) {
std::string msg("Set batch receive policy failed: C client returned failure");
deferred->Reject(msg);
return;
}
}
}