Add limits check

fixes: #60

Author: Gsantomaggio

Makefile

@@ -23,15 +23,21 @@ check: $(STATICCHECK)
 	$(STATICCHECK) ./pkg/stream
 
 test: vet fmt check
-	go test --tags=debug -v  ./pkg/stream -race -coverprofile=coverage.txt -covermode=atomic  #-ginkgo.v
+	go test --tags=debug -v  ./pkg/stream -coverprofile=coverage.txt -covermode=atomic  #-ginkgo.v
+
+build-all: vet fmt check build-darwin build-windows build-linux
+	 go test --tags=debug -v -race ./pkg/stream -coverprofile=coverage.txt -covermode=atomic  #-ginkgo.v
 
 integration-test: vet fmt check
 	cd ./pkg/system_integration && go test -v  . -race -coverprofile=coverage.txt -covermode=atomic -tags debug -timeout 99999s
 
+build-%: vet fmt check
+	GOOS=$(*) GOARCH=amd64 go build -ldflags=$(LDFLAGS) -v ./...
+
 build: vet fmt check
 	go build -ldflags=$(LDFLAGS) -v ./...
 
-PERFTEST_FLAGS ?= --producers 1 --consumers 1
+PERFTEST_FLAGS ?= --publishers 1 --consumers 1
 perf-test-run: perf-test-build
 	go run perfTest/perftest.go silent $(PERFTEST_FLAGS)
 

examples/publishersError/publisherError.go

@@ -4,7 +4,6 @@ import (
 	"bufio"
 	"fmt"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/amqp"
-	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/message"
 	"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
 
 	//"github.com/rabbitmq/rabbitmq-stream-go-client/pkg/stream"
@@ -20,14 +19,6 @@ func CheckErr(err error) {
 	}
 }
 
-func CreateArrayMessagesForTesting(bacthMessages int) []message.StreamMessage {
-	var arr []message.StreamMessage
-	for z := 0; z < bacthMessages; z++ {
-		arr = append(arr, amqp.NewMessage([]byte("hello_world_"+strconv.Itoa(z))))
-	}
-	return arr
-}
-
 func main() {
 	reader := bufio.NewReader(os.Stdin)
 
@@ -41,45 +32,43 @@ func main() {
 			SetUser("test").
 			SetPassword("test"))
 	CheckErr(err)
-	streamName := "no"
-	//err = env.DeclareStream(streamName,
-	//	&stream.StreamOptions{
-	//		MaxLengthBytes: stream.ByteCapacity{}.GB(2),
-	//	},
-	//)
-	//CheckErr(err)
+	streamName := "pub-error"
+	err = env.DeclareStream(streamName,
+		&stream.StreamOptions{
+			MaxLengthBytes: stream.ByteCapacity{}.GB(2),
+		},
+	)
 
-	producer, err := env.NewProducer(streamName, &stream.ProducerOptions{Name: "myProducer"})
+	CheckErr(err)
+
+	producer, err := env.NewProducer(streamName, stream.NewProducerOptions().SetProducerName("myProducer"))
 	CheckErr(err)
 
 	// This channel receives the callback in case the is some error during the
 	// publisher.
 	chPublishError := producer.NotifyPublishError()
 	handlePublishError(chPublishError)
 
-	go func() {
-		for i := 0; i < 100; i++ {
-			err := producer.BatchSend(CreateArrayMessagesForTesting(2))
-			CheckErr(err)
-		}
-	}()
-	// Here we close the producer during the publishing
-	// so an publish error is raised
-
-	err = producer.Close()
-	CheckErr(err)
+	for i := 0; i < 100; i++ {
+		msg := amqp.NewMessage([]byte("hello_world_" + strconv.Itoa(i)))
+		err := producer.Send(msg)
+		CheckErr(err)
+	}
 
 	fmt.Println("Press any key to stop ")
 	_, _ = reader.ReadString('\n')
 	CheckErr(err)
+	err = env.DeleteStream(streamName)
+	CheckErr(err)
+	err = env.Close()
+	CheckErr(err)
 
 }
 
 func handlePublishError(publishError stream.ChannelPublishError) {
 	go func() {
 		var totalMessages int32
-		for {
-			pError := <-publishError
+		for pError := range publishError {
 			atomic.AddInt32(&totalMessages, 1)
 			var data [][]byte
 			if pError.UnConfirmedMessage != nil {

pkg/stream/client.go

@@ -440,6 +440,17 @@ func (c *Client) DeclarePublisher(streamName string, options *ProducerOptions) (
 	if options == nil {
 		options = NewProducerOptions()
 	}
+
+	if options.QueueSize < minQueuePublisherSize || options.QueueSize > maxQueuePublisherSize {
+		return nil, fmt.Errorf("QueueSize values must be between %d and %d",
+			minQueuePublisherSize, maxQueuePublisherSize)
+	}
+
+	if options.BatchSize < minBatchSize || options.BatchSize > maxBatchSize {
+		return nil, fmt.Errorf("BatchSize values must be between %d and %d",
+			minBatchSize, maxBatchSize)
+	}
+
 	producer, err := c.coordinator.NewProducer(&ProducerOptions{
 		client:     c,
 		streamName: streamName,
@@ -621,6 +632,10 @@ func (c *Client) DeclareSubscriber(streamName string,
 		options = NewConsumerOptions()
 	}
 
+	if options.Offset.typeOfs <= 0 || options.Offset.typeOfs > 6 {
+		return nil, fmt.Errorf("specify a valid Offset")
+	}
+
 	options.client = c
 	options.streamName = streamName
 	consumer := c.coordinator.NewConsumer(messagesHandler, options)

pkg/stream/constants.go

@@ -76,8 +76,14 @@ const (
 	LocalhostUriConnection = "rabbitmq-stream://guest:guest@localhost:5552/%2f"
 
 	///
-	defaultSocketBuffer = 4096
-
+	defaultSocketBuffer       = 4096
+	defaultQueuePublisherSize = 10000
+	minQueuePublisherSize     = 100
+	maxQueuePublisherSize     = 1_000_000
+
+	minBatchSize     = 1
+	maxBatchSize     = 10_000
+	defaultBatchSize = 100
 	//
 	ClientVersion = "0.10-alpha"
 

pkg/stream/consumer.go

@@ -74,7 +74,7 @@ type ConsumerContext struct {
 
 type MessagesHandler func(consumerContext ConsumerContext, message *amqp.Message)
 
-type ConsumerOptions struct {
+type /**/ ConsumerOptions struct {
 	client       *Client
 	ConsumerName string
 	streamName   string

pkg/stream/consumer_test.go

@@ -266,4 +266,14 @@ var _ = Describe("Streaming Consumers", func() {
 
 	})
 
+	It("Validation", func() {
+		_, err := env.NewConsumer(streamName,
+			func(consumerContext ConsumerContext, message *amqp.Message) {
+			}, &ConsumerOptions{
+				Offset: OffsetSpecification{},
+			})
+		Expect(err).To(HaveOccurred())
+
+	})
+
 })

pkg/stream/enviroment.go

@@ -189,14 +189,6 @@ func (env *Environment) NewConsumer(streamName string,
 	return env.consumers.NewSubscriber(client, streamName, messagesHandler, options)
 }
 
-//func (env *Environment) NewStreamMessage(data []byte) message.StreamMessage {
-//	env.options.Codec
-//	return &AMQP10{
-//		message:      newMessage(data),
-//		publishingId: -1,
-//	}
-//}
-
 func (env *Environment) Close() error {
 	_ = env.producers.close()
 	_ = env.consumers.close()

pkg/stream/producer.go

@@ -71,8 +71,8 @@ func (po *ProducerOptions) SetBatchSize(size int) *ProducerOptions {
 
 func NewProducerOptions() *ProducerOptions {
 	return &ProducerOptions{
-		QueueSize: 10000,
-		BatchSize: 100,
+		QueueSize: defaultQueuePublisherSize,
+		BatchSize: defaultBatchSize,
 	}
 }
 

pkg/stream/producer_test.go

@@ -183,7 +183,7 @@ var _ = Describe("Streaming Producers", func() {
 		Expect(err).To(HaveOccurred())
 	})
 
-	It("Smart Send Split frame", func() {
+	It("Smart Send Split frame/BatchSize", func() {
 		producer, err := testEnvironment.NewProducer(testProducerStream,
 			NewProducerOptions().SetBatchSize(50))
 		Expect(err).NotTo(HaveOccurred())
@@ -206,9 +206,32 @@ var _ = Describe("Streaming Producers", func() {
 		s := make([]byte, 1148576)
 		err = producer.Send(amqp.NewMessage(s))
 		Expect(err).To(HaveOccurred())
+		err = producer.Close()
+		Expect(err).NotTo(HaveOccurred())
+
+		producer, err = testEnvironment.NewProducer(testProducerStream,
+			NewProducerOptions().SetBatchSize(2))
+		Expect(err).NotTo(HaveOccurred())
+		var messagesCountBatch int32
+		chConfirmBatch := producer.NotifyPublishConfirmation()
+		go func(ch ChannelPublishConfirm) {
+			for ids := range ch {
+				atomic.AddInt32(&messagesCountBatch, int32(len(ids)))
+			}
+		}(chConfirmBatch)
+
+		for i := 0; i < 100; i++ {
+			s := make([]byte, 11)
+			err = producer.Send(amqp.NewMessage(s))
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		time.Sleep(800 * time.Millisecond)
+		Expect(atomic.LoadInt32(&messagesCountBatch)).To(Equal(int32(100)))
 
 		err = producer.Close()
 		Expect(err).NotTo(HaveOccurred())
+
 	})
 
 	It("Smart Send send after", func() {
@@ -244,7 +267,7 @@ var _ = Describe("Streaming Producers", func() {
 		Expect(err).NotTo(HaveOccurred())
 	})
 
-	It("Already Closed", func() {
+	It("Already Closed/Limits", func() {
 		env, err := NewEnvironment(NewEnvironmentOptions().SetMaxProducersPerClient(5))
 		Expect(err).NotTo(HaveOccurred())
 		producer, err := env.NewProducer(testProducerStream, nil)
@@ -254,6 +277,27 @@ var _ = Describe("Streaming Producers", func() {
 
 		err = producer.Close()
 		Expect(err).To(Equal(AlreadyClosed))
+
+		/// validation limits
+		/// options.QueueSize and options.BatchSize
+		_, err = env.NewProducer(testProducerStream, &ProducerOptions{
+			QueueSize: 1,
+		})
+		Expect(err).To(HaveOccurred())
+
+		_, err = env.NewProducer(testProducerStream, NewProducerOptions().SetQueueSize(5000000))
+		Expect(err).To(HaveOccurred())
+
+		_, err = env.NewProducer(testProducerStream, &ProducerOptions{
+			BatchSize: 0,
+		})
+		Expect(err).To(HaveOccurred())
+
+		_, err = env.NewProducer(testProducerStream, &ProducerOptions{
+			BatchSize: 5000000,
+		})
+		Expect(err).To(HaveOccurred())
+
 		err = env.Close()
 		Expect(err).NotTo(HaveOccurred())
 	})