diff options
Diffstat (limited to 'weed/mq/client/cmd/weed_pub/publisher.go')
| -rw-r--r-- | weed/mq/client/cmd/weed_pub/publisher.go | 11 |
1 files changed, 8 insertions, 3 deletions
diff --git a/weed/mq/client/cmd/weed_pub/publisher.go b/weed/mq/client/cmd/weed_pub/publisher.go index ee00be9f8..59469e66b 100644 --- a/weed/mq/client/cmd/weed_pub/publisher.go +++ b/weed/mq/client/cmd/weed_pub/publisher.go @@ -12,7 +12,8 @@ import ( var ( messageCount = flag.Int("n", 1000, "message count") - concurrency = flag.Int("c", 4, "concurrency count") + concurrency = flag.Int("c", 4, "concurrent publishers") + partitionCount = flag.Int("p", 6, "partition count") namespace = flag.String("ns", "test", "namespace") topic = flag.String("topic", "test", "topic") @@ -25,7 +26,10 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) { // Simulate publishing a message key := []byte(fmt.Sprintf("key-%d-%d", id, i)) value := []byte(fmt.Sprintf("value-%d-%d", id, i)) - publisher.Publish(key, value) // Call your publisher function here + if err := publisher.Publish(key, value); err != nil { + fmt.Println(err) + break + } // println("Published", string(key), string(value)) } elapsed := time.Since(startTime) @@ -35,7 +39,8 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) { func main() { flag.Parse() config := &pub_client.PublisherConfiguration{ - CreateTopic: true, + CreateTopic: true, + CreateTopicPartitionCount: int32(*partitionCount), } publisher := pub_client.NewTopicPublisher(*namespace, *topic, config) brokers := strings.Split(*seedBrokers, ",") |