| /** |
| * 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/message.h> |
| #include <pulsar/c/producer_configuration.h> |
| |
| #include "c_structs.h" |
| |
| pulsar_producer_configuration_t *pulsar_producer_configuration_create() { |
| pulsar_producer_configuration_t *c_conf = new pulsar_producer_configuration_t; |
| c_conf->conf = pulsar::ProducerConfiguration(); |
| return c_conf; |
| } |
| |
| void pulsar_producer_configuration_free(pulsar_producer_configuration_t *conf) { delete conf; } |
| |
| void pulsar_producer_configuration_set_producer_name(pulsar_producer_configuration_t *conf, |
| const char *producerName) { |
| conf->conf.setProducerName(producerName); |
| } |
| |
| const char *pulsar_producer_configuration_get_producer_name(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getProducerName().c_str(); |
| } |
| |
| void pulsar_producer_configuration_set_send_timeout(pulsar_producer_configuration_t *conf, |
| int sendTimeoutMs) { |
| conf->conf.setSendTimeout(sendTimeoutMs); |
| } |
| |
| int pulsar_producer_configuration_get_send_timeout(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getSendTimeout(); |
| } |
| |
| void pulsar_producer_configuration_set_initial_sequence_id(pulsar_producer_configuration_t *conf, |
| int64_t initialSequenceId) { |
| conf->conf.setInitialSequenceId(initialSequenceId); |
| } |
| |
| int64_t pulsar_producer_configuration_get_initial_sequence_id(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getInitialSequenceId(); |
| } |
| |
| void pulsar_producer_configuration_set_compression_type(pulsar_producer_configuration_t *conf, |
| pulsar_compression_type compressionType) { |
| conf->conf.setCompressionType((pulsar::CompressionType)compressionType); |
| } |
| |
| pulsar_compression_type pulsar_producer_configuration_get_compression_type( |
| pulsar_producer_configuration_t *conf) { |
| return (pulsar_compression_type)conf->conf.getCompressionType(); |
| } |
| |
| void pulsar_producer_configuration_set_schema_info(pulsar_producer_configuration_t *conf, |
| pulsar_schema_type schemaType, const char *name, |
| const char *schema, pulsar_string_map_t *properties) { |
| auto schemaInfo = pulsar::SchemaInfo((pulsar::SchemaType)schemaType, name, schema, properties->map); |
| conf->conf.setSchema(schemaInfo); |
| } |
| |
| void pulsar_producer_configuration_set_max_pending_messages(pulsar_producer_configuration_t *conf, |
| int maxPendingMessages) { |
| conf->conf.setMaxPendingMessages(maxPendingMessages); |
| } |
| |
| int pulsar_producer_configuration_get_max_pending_messages(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getMaxPendingMessages(); |
| } |
| |
| void pulsar_producer_configuration_set_max_pending_messages_across_partitions( |
| pulsar_producer_configuration_t *conf, int maxPendingMessagesAcrossPartitions) { |
| conf->conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions); |
| } |
| |
| int pulsar_producer_configuration_get_max_pending_messages_across_partitions( |
| pulsar_producer_configuration_t *conf) { |
| return conf->conf.getMaxPendingMessagesAcrossPartitions(); |
| } |
| |
| void pulsar_producer_configuration_set_partitions_routing_mode(pulsar_producer_configuration_t *conf, |
| pulsar_partitions_routing_mode mode) { |
| conf->conf.setPartitionsRoutingMode((pulsar::ProducerConfiguration::PartitionsRoutingMode)mode); |
| } |
| |
| pulsar_partitions_routing_mode pulsar_producer_configuration_get_partitions_routing_mode( |
| pulsar_producer_configuration_t *conf) { |
| return (pulsar_partitions_routing_mode)conf->conf.getPartitionsRoutingMode(); |
| } |
| |
| void pulsar_producer_configuration_set_hashing_scheme(pulsar_producer_configuration_t *conf, |
| pulsar_hashing_scheme scheme) { |
| conf->conf.setHashingScheme((pulsar::ProducerConfiguration::HashingScheme)scheme); |
| } |
| |
| pulsar_hashing_scheme pulsar_producer_configuration_get_hashing_scheme( |
| pulsar_producer_configuration_t *conf) { |
| return (pulsar_hashing_scheme)conf->conf.getHashingScheme(); |
| } |
| |
| class MessageRoutingPolicy : public pulsar::MessageRoutingPolicy { |
| pulsar_message_router _router; |
| void *_ctx; |
| |
| public: |
| MessageRoutingPolicy(pulsar_message_router router, void *ctx) : _router(router), _ctx(ctx) {} |
| |
| int getPartition(const pulsar::Message &msg, const pulsar::TopicMetadata &topicMetadata) { |
| pulsar_message_t message; |
| message.message = msg; |
| |
| pulsar_topic_metadata_t metadata; |
| metadata.metadata = &topicMetadata; |
| |
| return _router(&message, &metadata, _ctx); |
| } |
| }; |
| |
| void pulsar_producer_configuration_set_message_router(pulsar_producer_configuration_t *conf, |
| pulsar_message_router router, void *ctx) { |
| conf->conf.setMessageRouter(std::make_shared<MessageRoutingPolicy>(router, ctx)); |
| } |
| |
| void pulsar_producer_configuration_set_lazy_start_partitioned_producers( |
| pulsar_producer_configuration_t *conf, int useLazyStartPartitionedProducers) { |
| conf->conf.setLazyStartPartitionedProducers(useLazyStartPartitionedProducers); |
| } |
| |
| int pulsar_producer_configuration_get_lazy_start_partitioned_producers( |
| pulsar_producer_configuration_t *conf) { |
| return conf->conf.getLazyStartPartitionedProducers(); |
| } |
| |
| void pulsar_producer_configuration_set_block_if_queue_full(pulsar_producer_configuration_t *conf, |
| int blockIfQueueFull) { |
| conf->conf.setBlockIfQueueFull(blockIfQueueFull); |
| } |
| |
| int pulsar_producer_configuration_get_block_if_queue_full(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getBlockIfQueueFull(); |
| } |
| |
| void pulsar_producer_configuration_set_batching_enabled(pulsar_producer_configuration_t *conf, |
| int batchingEnabled) { |
| conf->conf.setBatchingEnabled(batchingEnabled); |
| } |
| |
| int pulsar_producer_configuration_get_batching_enabled(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getBatchingEnabled(); |
| } |
| |
| void pulsar_producer_configuration_set_batching_max_messages(pulsar_producer_configuration_t *conf, |
| unsigned int batchingMaxMessages) { |
| conf->conf.setBatchingMaxMessages(batchingMaxMessages); |
| } |
| |
| unsigned int pulsar_producer_configuration_get_batching_max_messages(pulsar_producer_configuration_t *conf) { |
| return conf->conf.getBatchingMaxMessages(); |
| } |
| |
| void pulsar_producer_configuration_set_batching_max_allowed_size_in_bytes( |
| pulsar_producer_configuration_t *conf, unsigned long batchingMaxAllowedSizeInBytes) { |
| conf->conf.setBatchingMaxAllowedSizeInBytes(batchingMaxAllowedSizeInBytes); |
| } |
| |
| unsigned long pulsar_producer_configuration_get_batching_max_allowed_size_in_bytes( |
| pulsar_producer_configuration_t *conf) { |
| return conf->conf.getBatchingMaxAllowedSizeInBytes(); |
| } |
| |
| void pulsar_producer_configuration_set_batching_max_publish_delay_ms( |
| pulsar_producer_configuration_t *conf, unsigned long batchingMaxPublishDelayMs) { |
| conf->conf.setBatchingMaxPublishDelayMs(batchingMaxPublishDelayMs); |
| } |
| |
| unsigned long pulsar_producer_configuration_get_batching_max_publish_delay_ms( |
| pulsar_producer_configuration_t *conf) { |
| return conf->conf.getBatchingMaxPublishDelayMs(); |
| } |
| |
| void pulsar_producer_configuration_set_property(pulsar_producer_configuration_t *conf, const char *name, |
| const char *value) { |
| conf->conf.setProperty(name, value); |
| } |
| |
| int pulsar_producer_is_encryption_enabled(pulsar_producer_configuration_t *conf) { |
| return conf->conf.isEncryptionEnabled(); |
| } |
| |
| void pulsar_producer_configuration_set_default_crypto_key_reader(pulsar_producer_configuration_t *conf, |
| const char *public_key_path, |
| const char *private_key_path) { |
| std::shared_ptr<pulsar::DefaultCryptoKeyReader> keyReader = |
| std::make_shared<pulsar::DefaultCryptoKeyReader>(public_key_path, private_key_path); |
| conf->conf.setCryptoKeyReader(keyReader); |
| } |
| |
| pulsar_producer_crypto_failure_action pulsar_producer_configuration_get_crypto_failure_action( |
| pulsar_producer_configuration_t *conf) { |
| return (pulsar_producer_crypto_failure_action)conf->conf.getCryptoFailureAction(); |
| } |
| |
| void pulsar_producer_configuration_set_crypto_failure_action( |
| pulsar_producer_configuration_t *conf, pulsar_producer_crypto_failure_action cryptoFailureAction) { |
| conf->conf.setCryptoFailureAction((pulsar::ProducerCryptoFailureAction)cryptoFailureAction); |
| } |
| |
| void pulsar_producer_configuration_set_encryption_key(pulsar_producer_configuration_t *conf, |
| const char *key) { |
| conf->conf.addEncryptionKey(key); |
| } |
| |
| void pulsar_producer_configuration_set_chunking_enabled(pulsar_producer_configuration_t *conf, |
| int chunkingEnabled) { |
| conf->conf.setChunkingEnabled(chunkingEnabled); |
| } |
| |
| int pulsar_producer_configuration_is_chunking_enabled(pulsar_producer_configuration_t *conf) { |
| return conf->conf.isChunkingEnabled(); |
| } |
| |
| pulsar_producer_access_mode pulsar_producer_configuration_get_access_mode( |
| pulsar_producer_configuration_t *conf) { |
| return (pulsar_producer_access_mode)conf->conf.getAccessMode(); |
| } |
| |
| void pulsar_producer_configuration_set_access_mode(pulsar_producer_configuration_t *conf, |
| pulsar_producer_access_mode accessMode) { |
| conf->conf.setAccessMode((pulsar::ProducerConfiguration::ProducerAccessMode)accessMode); |
| } |