diff options
Diffstat (limited to 'weed/mq/client/cmd')
| -rw-r--r-- | weed/mq/client/cmd/weed_pub/publisher.go | 13 |
1 files changed, 10 insertions, 3 deletions
diff --git a/weed/mq/client/cmd/weed_pub/publisher.go b/weed/mq/client/cmd/weed_pub/publisher.go index 2873ba21f..e9227130a 100644 --- a/weed/mq/client/cmd/weed_pub/publisher.go +++ b/weed/mq/client/cmd/weed_pub/publisher.go @@ -16,6 +16,8 @@ var ( concurrency = flag.Int("c", 4, "concurrent publishers") partitionCount = flag.Int("p", 6, "partition count") + clientName = flag.String("client", "c1", "client name") + namespace = flag.String("ns", "test", "namespace") t = flag.String("t", "test", "t") seedBrokers = flag.String("brokers", "localhost:17777", "seed brokers") @@ -25,16 +27,20 @@ func doPublish(publisher *pub_client.TopicPublisher, id int) { startTime := time.Now() for i := 0; i < *messageCount / *concurrency; i++ { // Simulate publishing a message - key := []byte(fmt.Sprintf("key-%d-%d", id, i)) - value := []byte(fmt.Sprintf("value-%d-%d", id, i)) + key := []byte(fmt.Sprintf("key-%s-%d-%d", *clientName, id, i)) + value := []byte(fmt.Sprintf("value-%s-%d-%d", *clientName, id, i)) if err := publisher.Publish(key, value); err != nil { fmt.Println(err) break } + time.Sleep(time.Second) // println("Published", string(key), string(value)) } + if err := publisher.FinishPublish(); err != nil { + fmt.Println(err) + } elapsed := time.Since(startTime) - log.Printf("Publisher %d finished in %s", id, elapsed) + log.Printf("Publisher %s-%d finished in %s", *clientName, id, elapsed) } func main() { @@ -44,6 +50,7 @@ func main() { CreateTopic: true, CreateTopicPartitionCount: int32(*partitionCount), Brokers: strings.Split(*seedBrokers, ","), + PublisherName: *clientName, } publisher := pub_client.NewTopicPublisher(config) |