Return an error in the leader is not ready yet (#98)

* Return an error in the
leader is not ready yet, add also a small fix in the aggregation
Co-authored-by: Victor Elias <[email protected]>

Author: Gsantomaggio

pkg/stream/aggregation.go

@@ -40,15 +40,15 @@ func (compression Compression) Gzip() Compression {
 }
 
 func (compression Compression) Snappy() Compression {
-	return Compression{value: SNAPPY}
+	return Compression{value: SNAPPY, enabled: true}
 }
 
 func (compression Compression) Zstd() Compression {
-	return Compression{value: ZSTD}
+	return Compression{value: ZSTD, enabled: true}
 }
 
 func (compression Compression) Lz4() Compression {
-	return Compression{value: LZ4}
+	return Compression{value: LZ4, enabled: true}
 }
 
 type subEntry struct {

pkg/stream/client.go

@@ -590,7 +590,7 @@ func (c *Client) BrokerLeader(stream string) (*Broker, error) {
 		return nil, lookErrorCode(streamMetadata.responseCode)
 	}
 	if streamMetadata.Leader == nil {
-		return nil, fmt.Errorf("leader error for stream for stream: %s", stream)
+		return nil, LeaderNotReady
 	}
 
 	streamMetadata.Leader.advPort = streamMetadata.Leader.Port
@@ -610,16 +610,28 @@ func (c *Client) StreamExists(stream string) bool {
 func (c *Client) BrokerForConsumer(stream string) (*Broker, error) {
 	streamsMetadata := c.metaData(stream)
 	if streamsMetadata == nil {
-		return nil, fmt.Errorf("leader error for stream for stream: %s", stream)
+		return nil, fmt.Errorf("leader error for stream: %s", stream)
 	}
 
 	streamMetadata := streamsMetadata.Get(stream)
 	if streamMetadata.responseCode != responseCodeOk {
 		return nil, lookErrorCode(streamMetadata.responseCode)
 	}
-	var brokers []*Broker
+
+	if streamMetadata.Leader == nil {
+		return nil, LeaderNotReady
+	}
+
+	brokers := make([]*Broker, 0, 1+len(streamMetadata.Replicas))
 	brokers = append(brokers, streamMetadata.Leader)
-	brokers = append(brokers, streamMetadata.Replicas...)
+	for idx, replica := range streamMetadata.Replicas {
+		if replica == nil {
+			logs.LogWarn("Stream %s replica not ready: %d", stream, idx)
+			continue
+		}
+		brokers = append(brokers, replica)
+	}
+
 	rand.Seed(time.Now().UnixNano())
 	n := rand.Intn(len(brokers))
 	return brokers[n], nil

pkg/stream/constants.go

@@ -2,9 +2,10 @@ package stream
 
 import (
 	"fmt"
+	"time"
+
 	"github.com/pkg/errors"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/logs"
-	"time"
 )
 
 // is needed to indicate the general status
@@ -112,6 +113,11 @@ var PublisherDoesNotExist = errors.New("Publisher Does Not Exist")
 var FrameTooLarge = errors.New("Frame Too Large, the buffer is too big")
 var CodeAccessRefused = errors.New("Resources Access Refused")
 var ConnectionClosed = errors.New("Can't send the message, connection closed")
+var StreamNotAvailable = errors.New("Stream Not Available")
+var UnknownFrame = errors.New("Unknown Frame")
+var InternalError = errors.New("Internal Error")
+
+var LeaderNotReady = errors.New("Leader not Ready yet")
 
 func lookErrorCode(errorCode uint16) error {
 	switch errorCode {
@@ -135,6 +141,12 @@ func lookErrorCode(errorCode uint16) error {
 		return FrameTooLarge
 	case responseCodeAccessRefused:
 		return CodeAccessRefused
+	case responseCodeStreamNotAvailable:
+		return StreamNotAvailable
+	case responseCodeUnknownFrame:
+		return UnknownFrame
+	case responseCodeInternalError:
+		return InternalError
 	default:
 		{
 			logs.LogWarn("Error not handled %d", errorCode)

pkg/stream/consumer_test.go

@@ -426,7 +426,7 @@ var _ = Describe("Streaming Consumers", func() {
 
 		Eventually(func() int32 {
 			return atomic.LoadInt32(&messagesReceived)
-		}, 9*time.Second).Should(Equal(int32((2000*5)+(50*len(batchMessages)))),
+		}, 15*time.Second).Should(Equal(int32((2000*5)+(50*len(batchMessages)))),
 			"consumer should be the same sent from different publishers settings")
 
 		Expect(producer1.Close()).NotTo(HaveOccurred())

pkg/stream/enviroment.go

@@ -174,7 +174,7 @@ func (env *Environment) StreamMetaData(streamName string) (*StreamMetadata, erro
 
 	if streamMetadata.Leader == nil {
 
-		return nil, errors.New("can't find leader for stream: " + streamName)
+		return nil, LeaderNotReady
 	}
 
 	return streamMetadata, nil