add publisher name for debugging

1 files changed, 7 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..0950af187 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,17 @@ 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))
 	}
 	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 +47,7 @@ func main() {
 		CreateTopic:               true,
 		CreateTopicPartitionCount: int32(*partitionCount),
 		Brokers:                   strings.Split(*seedBrokers, ","),
+		PublisherName:             *clientName,
 	}
 	publisher := pub_client.NewTopicPublisher(config)