merlimat closed pull request #2447: Added producer/consumer properties in Go
client
URL: https://github.com/apache/incubator-pulsar/pull/2447
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
index 45154c5c6a..565a6ab0ea 100644
--- a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
+++ b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h
@@ -128,7 +128,7 @@ class ProducerConfiguration {
ProducerConfiguration& addEncryptionKey(std::string key);
/**
- * Check whether the message has a specific property attached.
+ * Check whether the producer has a specific property attached.
*
* @param name the name of the property to check
* @return true if the message has the specified property
@@ -150,7 +150,8 @@ class ProducerConfiguration {
std::map<std::string, std::string>& getProperties() const;
/**
- * Sets a new property on a message.
+ * Sets a new property on the producer
+ * .
* @param name the name of the property
* @param value the associated value
*/
diff --git a/pulsar-client-cpp/include/pulsar/c/consumer_configuration.h
b/pulsar-client-cpp/include/pulsar/c/consumer_configuration.h
index 7299867f38..fca47eb06d 100644
--- a/pulsar-client-cpp/include/pulsar/c/consumer_configuration.h
+++ b/pulsar-client-cpp/include/pulsar/c/consumer_configuration.h
@@ -152,6 +152,9 @@ int
pulsar_consumer_is_read_compacted(pulsar_consumer_configuration_t *consumer_
void pulsar_consumer_set_read_compacted(pulsar_consumer_configuration_t
*consumer_configuration,
int compacted);
+void
pulsar_consumer_configuration_set_property(pulsar_consumer_configuration_t
*conf, const char *name,
+ const char *value);
+
// const CryptoKeyReaderPtr getCryptoKeyReader()
//
// const;
diff --git a/pulsar-client-cpp/include/pulsar/c/producer_configuration.h
b/pulsar-client-cpp/include/pulsar/c/producer_configuration.h
index ae88198f9a..670bf50ed8 100644
--- a/pulsar-client-cpp/include/pulsar/c/producer_configuration.h
+++ b/pulsar-client-cpp/include/pulsar/c/producer_configuration.h
@@ -134,6 +134,9 @@ void
pulsar_producer_configuration_set_batching_max_publish_delay_ms(pulsar_prod
unsigned long pulsar_producer_configuration_get_batching_max_publish_delay_ms(
pulsar_producer_configuration_t *conf);
+void
pulsar_producer_configuration_set_property(pulsar_producer_configuration_t
*conf, const char *name,
+ const char *value);
+
// const CryptoKeyReaderPtr getCryptoKeyReader() const;
// ProducerConfiguration &setCryptoKeyReader(CryptoKeyReaderPtr
cryptoKeyReader);
//
diff --git a/pulsar-client-cpp/lib/c/c_ConsumerConfiguration.cc
b/pulsar-client-cpp/lib/c/c_ConsumerConfiguration.cc
index c8d545383b..75cdc47ec0 100644
--- a/pulsar-client-cpp/lib/c/c_ConsumerConfiguration.cc
+++ b/pulsar-client-cpp/lib/c/c_ConsumerConfiguration.cc
@@ -113,3 +113,8 @@ void
pulsar_consumer_set_read_compacted(pulsar_consumer_configuration_t *consume
int compacted) {
consumer_configuration->consumerConfiguration.setReadCompacted(compacted);
}
+
+void
pulsar_consumer_configuration_set_property(pulsar_consumer_configuration_t
*conf, const char *name,
+ const char *value) {
+ conf->consumerConfiguration.setProperty(name, value);
+}
diff --git a/pulsar-client-cpp/lib/c/c_ProducerConfiguration.cc
b/pulsar-client-cpp/lib/c/c_ProducerConfiguration.cc
index 914fc0ac13..a8eb5be3ec 100644
--- a/pulsar-client-cpp/lib/c/c_ProducerConfiguration.cc
+++ b/pulsar-client-cpp/lib/c/c_ProducerConfiguration.cc
@@ -174,3 +174,8 @@ 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);
+}
diff --git a/pulsar-client-go/pulsar/c_consumer.go
b/pulsar-client-go/pulsar/c_consumer.go
index 7f613b29d6..1b41a71a49 100644
--- a/pulsar-client-go/pulsar/c_consumer.go
+++ b/pulsar-client-go/pulsar/c_consumer.go
@@ -120,6 +120,18 @@ func subscribeAsync(client *client, options
ConsumerOptions, callback func(Consu
C.pulsar_consumer_set_consumer_name(conf, name)
}
+ if options.Properties != nil {
+ for key, value := range options.Properties {
+ cKey := C.CString(key)
+ cValue := C.CString(value)
+
+ C.pulsar_consumer_configuration_set_property(conf,
cKey, cValue)
+
+ C.free(unsafe.Pointer(cKey))
+ C.free(unsafe.Pointer(cValue))
+ }
+ }
+
C.pulsar_consumer_set_read_compacted(conf, cBool(options.ReadCompacted))
subName := C.CString(options.SubscriptionName)
diff --git a/pulsar-client-go/pulsar/c_producer.go
b/pulsar-client-go/pulsar/c_producer.go
index b4cd2c5e14..284315dbb2 100644
--- a/pulsar-client-go/pulsar/c_producer.go
+++ b/pulsar-client-go/pulsar/c_producer.go
@@ -124,6 +124,18 @@ func createProducerAsync(client *client, options
ProducerOptions, callback func(
C.pulsar_producer_configuration_set_batching_max_messages(conf,
C.uint(options.BatchingMaxMessages))
}
+ if options.Properties != nil {
+ for key, value := range options.Properties {
+ cKey := C.CString(key)
+ cValue := C.CString(value)
+
+ C.pulsar_producer_configuration_set_property(conf,
cKey, cValue)
+
+ C.free(unsafe.Pointer(cKey))
+ C.free(unsafe.Pointer(cValue))
+ }
+ }
+
topicName := C.CString(options.Topic)
defer C.free(unsafe.Pointer(topicName))
diff --git a/pulsar-client-go/pulsar/consumer.go
b/pulsar-client-go/pulsar/consumer.go
index b9f2616c07..030ba1b3a8 100644
--- a/pulsar-client-go/pulsar/consumer.go
+++ b/pulsar-client-go/pulsar/consumer.go
@@ -64,6 +64,10 @@ type ConsumerOptions struct {
// This argument is required when subscribing
SubscriptionName string
+ // Attach a set of application defined properties to the consumer
+ // This properties will be visible in the topic stats
+ Properties map[string]string
+
// Set the timeout for unacked messages
// Message not acknowledged within the give time, will be replayed by
the broker to the same or a different consumer
// Default is 0, which means message are not being replayed based on
ack time
diff --git a/pulsar-client-go/pulsar/producer.go
b/pulsar-client-go/pulsar/producer.go
index 2cfd141633..46d6dd68ed 100644
--- a/pulsar-client-go/pulsar/producer.go
+++ b/pulsar-client-go/pulsar/producer.go
@@ -71,6 +71,10 @@ type ProducerOptions struct {
// a topic.
Name string
+ // Attach a set of application defined properties to the producer
+ // This properties will be visible in the topic stats
+ Properties map[string]string
+
// Set the send timeout (default: 30 seconds)
// If a message is not acknowledged by the server before the
sendTimeout expires, an error will be reported.
// Setting the timeout to -1, will set the timeout to infinity, which
can be useful when using Pulsar's message
diff --git a/pulsar-client-go/pulsar/producer_test.go
b/pulsar-client-go/pulsar/producer_test.go
index 940be85be1..cfa0bcbd1a 100644
--- a/pulsar-client-go/pulsar/producer_test.go
+++ b/pulsar-client-go/pulsar/producer_test.go
@@ -20,9 +20,9 @@
package pulsar
import (
- "testing"
- "fmt"
"context"
+ "fmt"
+ "testing"
"time"
)
@@ -77,6 +77,10 @@ func TestProducer(t *testing.T) {
MaxPendingMessages: 100,
BlockIfQueueFull: true,
CompressionType: LZ4,
+ Properties: map[string]string{
+ "my-name": "test",
+ "key": "value",
+ },
})
assertNil(t, err)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
