Add flag to configure the CRC (#110)

Add the perftest flag
Closes #109

Author: Gsantomaggio

README.md

@@ -342,8 +342,11 @@ With `ConsumerOptions` it is possible to customize the consumer behaviour.
 ```golang
   stream.NewConsumerOptions().
   SetConsumerName("my_consumer").                  // set a consumer name
+  SetCRCCheck(false).  // Enable/Disable the CRC control.  
   SetOffset(stream.OffsetSpecification{}.First())) // start consuming from the beginning
 ```
+Disabling the CRC control can increase the performances.
+
 See also "Offset Start" example in the [examples](./examples/) directory
 
 

examples/getting_started.go

@@ -102,7 +102,8 @@ func main() {
 		handleMessages,
 		stream.NewConsumerOptions().
 			SetConsumerName("my_consumer").                  // set a consumer name
-			SetOffset(stream.OffsetSpecification{}.First())) // start consuming from the beginning
+			SetOffset(stream.OffsetSpecification{}.First()). // start consuming from the beginning
+			SetCRCCheck(false))                              // Disable crc control, increase the performances
 	CheckErr(err)
 	channelClose := consumer.NotifyClose()
 	// channelClose receives all the closing events, here you can handle the

perfTest/cmd/commands.go

@@ -43,6 +43,7 @@ var (
 	compression         string
 	exitOnError         bool
 	debugLogs           bool
+	crcCheck            bool
 	runDuration         int
 )
 
@@ -67,6 +68,7 @@ func setupCli(baseCmd *cobra.Command) {
 	baseCmd.PersistentFlags().BoolVarP(&exitOnError, "exit-on-error", "", true, "Close the app in case of error")
 	baseCmd.PersistentFlags().BoolVarP(&printStatsV, "print-stats", "", true, "Print stats")
 	baseCmd.PersistentFlags().BoolVarP(&debugLogs, "debug-logs", "", false, "Enable debug logs")
+	baseCmd.PersistentFlags().BoolVarP(&crcCheck, "crc-check", "", false, "Enable crc control")
 	baseCmd.PersistentFlags().StringSliceVarP(&streams, "streams", "", []string{"perf-test-go"}, "Stream names")
 	baseCmd.PersistentFlags().StringVarP(&maxLengthBytes, "max-length-bytes", "", "0", "Stream max length bytes, e.g. 10MB, 50GB, etc.")
 	baseCmd.PersistentFlags().IntVarP(&maxAge, "max-age", "", 0, "Stream Age in hours, e.g. 1,2.. 24 , etc.")

perfTest/cmd/silent.go

@@ -390,7 +390,7 @@ func startConsumer(consumerName string, streamName string) error {
 		streamName,
 		handleMessages,
 		stream.NewConsumerOptions().
-			SetConsumerName(consumerName).SetOffset(offsetSpec))
+			SetConsumerName(consumerName).SetOffset(offsetSpec).SetCRCCheck(crcCheck))
 	if err != nil {
 		return err
 	}

pkg/stream/consumer.go

@@ -137,20 +137,28 @@ type ConsumerOptions struct {
 	autocommit         bool
 	autoCommitStrategy *AutoCommitStrategy
 	Offset             OffsetSpecification
+	CRCCheck           bool
 }
 
 func NewConsumerOptions() *ConsumerOptions {
 	return &ConsumerOptions{
 		Offset:             OffsetSpecification{}.Last(),
 		autocommit:         false,
-		autoCommitStrategy: NewAutoCommitStrategy()}
+		autoCommitStrategy: NewAutoCommitStrategy(),
+		CRCCheck:           false,
+	}
 }
 
 func (c *ConsumerOptions) SetConsumerName(consumerName string) *ConsumerOptions {
 	c.ConsumerName = consumerName
 	return c
 }
 
+func (c *ConsumerOptions) SetCRCCheck(CRCCheck bool) *ConsumerOptions {
+	c.CRCCheck = CRCCheck
+	return c
+}
+
 func (c *ConsumerOptions) SetAutoCommit(autoCommitStrategy *AutoCommitStrategy) *ConsumerOptions {
 	c.autocommit = true
 	if autoCommitStrategy == nil {

pkg/stream/consumer_test.go

@@ -58,7 +58,8 @@ var _ = Describe("Streaming Consumers", func() {
 	})
 
 	It("Multi Consumers per client", func() {
-		env, err := NewEnvironment(NewEnvironmentOptions().SetMaxConsumersPerClient(2))
+		env, err := NewEnvironment(NewEnvironmentOptions().
+			SetMaxConsumersPerClient(2))
 		Expect(err).NotTo(HaveOccurred())
 		streamName := uuid.New().String()
 		Expect(env.DeclareStream(streamName, nil)).
@@ -161,7 +162,9 @@ var _ = Describe("Streaming Consumers", func() {
 		consumer, err := env.NewConsumer(streamName,
 			func(consumerContext ConsumerContext, message *amqp.Message) {
 				atomic.AddInt32(&messagesReceived, 1)
-			}, NewConsumerOptions().SetOffset(OffsetSpecification{}.Offset(50)))
+			}, NewConsumerOptions().
+				SetOffset(OffsetSpecification{}.Offset(50)).
+				SetCRCCheck(true))
 		Expect(err).NotTo(HaveOccurred())
 
 		Eventually(func() int32 {
@@ -185,6 +188,7 @@ var _ = Describe("Streaming Consumers", func() {
 			}, NewConsumerOptions().
 				SetOffset(OffsetSpecification{}.First()).
 				SetConsumerName("my_auto_consumer").
+				SetCRCCheck(false).
 				SetAutoCommit(NewAutoCommitStrategy().
 					SetCountBeforeStorage(100).
 					SetFlushInterval(50*time.Second))) // here we set a high value to do not trigger the time
@@ -206,6 +210,7 @@ var _ = Describe("Streaming Consumers", func() {
 			}, NewConsumerOptions().
 				SetOffset(OffsetSpecification{}.First()).
 				SetConsumerName("my_auto_consumer_timer").
+				SetCRCCheck(true).
 				SetAutoCommit(NewAutoCommitStrategy().
 					SetCountBeforeStorage(10000000). /// We avoid raising the timer
 					SetFlushInterval(1*time.Second)))

pkg/stream/server_frame.go

@@ -328,12 +328,13 @@ func (c *Client) handleDeliver(r *bufio.Reader) {
 
 	/// headers ---> payload -> messages
 
-	checkSum := crc32.ChecksumIEEE(bytesBuffer)
-
-	if crc != checkSum {
-		logs.LogError("Error during the checkSum, expected %d, checksum %d", crc, checkSum)
-		panic("Error during CRC")
-	} /// ???
+	if consumer.options.CRCCheck {
+		checkSum := crc32.ChecksumIEEE(bytesBuffer)
+		if crc != checkSum {
+			logs.LogError("Error during the checkSum, expected %d, checksum %d", crc, checkSum)
+			panic("Error during CRC")
+		} /// ???
+	}
 
 	bufferReader := bytes.NewReader(bytesBuffer)
 	dataReader := bufio.NewReader(bufferReader)