Add metadata info

Author: Gsantomaggio

perfTest/cmd/silent.go

@@ -156,6 +156,9 @@ func initStreams() error {
 	}
 
 	for _, streamName := range streams {
+		streamMetadata, err := env.StreamMetaData(streamName)
+		checkErr(err)
+		logInfo("stream %s, meta data: %s", streamName, streamMetadata)
 
 		err = env.DeclareStream(
 			streamName,

pkg/stream/brokers.go

@@ -132,13 +132,21 @@ type StreamMetadata struct {
 	stream       string
 	responseCode uint16
 	Leader       *Broker
-	replicas     []*Broker
+	Replicas     []*Broker
+}
+
+func (sm StreamMetadata) String() string {
+	replicas := ""
+	for _, replica := range sm.Replicas {
+		replicas += fmt.Sprintf("%s:%s", replica.Host, replica.Port)
+	}
+	return fmt.Sprintf("leader %s:%s, followers %s ", sm.Leader.Host, sm.Leader.Port, replicas)
 }
 
 func (StreamMetadata) New(stream string, responseCode uint16,
 	leader *Broker, replicas []*Broker) *StreamMetadata {
 	return &StreamMetadata{stream: stream, responseCode: responseCode,
-		Leader: leader, replicas: replicas}
+		Leader: leader, Replicas: replicas}
 }
 
 type StreamsMetadata struct {

pkg/stream/client.go

@@ -604,7 +604,7 @@ func (c *Client) BrokerForConsumer(stream string) (*Broker, error) {
 	}
 	var brokers []*Broker
 	brokers = append(brokers, streamMetadata.Leader)
-	brokers = append(brokers, streamMetadata.replicas...)
+	brokers = append(brokers, streamMetadata.Replicas...)
 	rand.Seed(time.Now().UnixNano())
 	n := rand.Intn(len(brokers))
 	return brokers[n], nil

pkg/stream/enviroment_test.go

@@ -249,7 +249,7 @@ var _ = Describe("Environment test", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(metaData.Leader.Host).To(Equal("localhost"))
 		Expect(metaData.Leader.Port).To(Equal("5552"))
-		Expect(len(metaData.replicas)).To(Equal(0))
+		Expect(len(metaData.Replicas)).To(Equal(0))
 		err = env.DeleteStream(stream)
 		Expect(err).NotTo(HaveOccurred())
 		exists, err = env.StreamExists(stream)