Add limit for the batch entry size

Author: Gsantomaggio

pkg/amqp/buffer.go

@@ -21,7 +21,6 @@
 //OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 //SOFTWARE
 
-
 package amqp
 
 import (

pkg/amqp/encode.go

@@ -21,8 +21,6 @@
 //OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 //SOFTWARE
 
-
-
 package amqp
 
 import (

pkg/amqp/error_stdlib.go

@@ -21,7 +21,6 @@
 //OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 //SOFTWARE
 
-
 //go:build !pkgerrors
 // +build !pkgerrors
 

pkg/amqp/types.go

@@ -21,7 +21,6 @@
 //OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 //SOFTWARE
 
-
 package amqp
 
 import (

pkg/stream/client.go

@@ -464,9 +464,9 @@ func (c *Client) DeclarePublisher(streamName string, options *ProducerOptions) (
 			minBatchPublishingDelay, maxBatchPublishingDelay)
 	}
 
-	if options.SubEntrySize < minSubEntrySize {
-		return nil, fmt.Errorf("SubEntrySize value must equal or bigger than %d",
-			minSubEntrySize)
+	if options.SubEntrySize < minSubEntrySize || options.SubEntrySize > maxSubEntrySize {
+		return nil, fmt.Errorf("SubEntrySize values must be between %d and %d",
+			minSubEntrySize, maxSubEntrySize)
 	}
 
 	if !options.isSubEntriesBatching() {

pkg/stream/constants.go

@@ -90,6 +90,7 @@ const (
 	maxBatchSize = 10_000
 
 	minSubEntrySize = 1
+	maxSubEntrySize = 65535
 
 	minBatchPublishingDelay = 50
 	maxBatchPublishingDelay = 500

pkg/stream/producer.go

@@ -45,7 +45,6 @@ func (cs *ConfirmationStatus) GetMessage() message.StreamMessage {
 	return cs.message
 }
 
-
 func (cs *ConfirmationStatus) GetErrorCode() uint16 {
 	return cs.errorCode
 }

pkg/stream/producer_test.go

@@ -485,6 +485,11 @@ var _ = Describe("Streaming Producers", func() {
 			SetSubEntrySize(1).SetCompression(Compression{}.Gzip()))
 		Expect(err).To(HaveOccurred())
 
+		_, err = env.NewProducer(testProducerStream, &ProducerOptions{
+			SubEntrySize: 65_539,
+		})
+		Expect(err).To(HaveOccurred())
+
 		err = env.Close()
 		Expect(err).NotTo(HaveOccurred())
 	})