Use store/tracking lingo instead of commit (#48)

Author: Gsantomaggio

examples/offsetTracking/offsetTracking.go

@@ -69,8 +69,8 @@ func main() {
 	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
 		if atomic.AddInt32(&count, 1)%1000 == 0 {
 			fmt.Printf("cousumed %d  messages \n", atomic.LoadInt32(&count))
-			// AVOID to commit for each single message, it will reduce the performances
-			err := consumerContext.Consumer.Commit()
+			// AVOID to store for each single message, it will reduce the performances
+			err := consumerContext.Consumer.StoreOffset()
 			if err != nil {
 				CheckErr(err)
 			}

perfTest/cmd/silent.go

@@ -310,11 +310,11 @@ func handleConsumerClose(channelClose stream.ChannelClose) {
 func startConsumer(consumerName string, streamName string) error {
 
 	handleMessages := func(consumerContext stream.ConsumerContext, message *amqp.Message) {
-		//logError("consumerMessageCount Commit: %s", consumerMessageCount)
+		//logError("consumerMessageCount StoreOffset: %s", consumerMessageCount)
 		if atomic.AddInt32(&consumerMessageCount, 1)%500 == 0 {
-			err := consumerContext.Consumer.Commit()
+			err := consumerContext.Consumer.StoreOffset()
 			if err != nil {
-				logError("Error Commit: %s", err)
+				logError("Error StoreOffset: %s", err)
 			}
 		}
 	}

pkg/stream/constants.go

@@ -17,7 +17,7 @@ const (
 	commandSubscribe              = 7
 	commandDeliver                = 8
 	commandCredit                 = 9
-	commandCommitOffset           = 10
+	commandStoreOffset            = 10
 	CommandQueryOffset            = 11
 	CommandUnsubscribe            = 12
 	commandCreateStream           = 13

pkg/stream/consumer.go

@@ -150,14 +150,14 @@ func (consumer *Consumer) Close() error {
 	return err.Err
 }
 
-func (consumer *Consumer) Commit() error {
+func (consumer *Consumer) StoreOffset() error {
 	if consumer.options.streamName == "" {
 		return fmt.Errorf("stream Name can't be empty")
 	}
 	length := 2 + 2 + 2 + len(consumer.options.ConsumerName) + 2 +
 		len(consumer.options.streamName) + 8
 	var b = bytes.NewBuffer(make([]byte, 0, length+4))
-	writeProtocolHeader(b, length, commandCommitOffset)
+	writeProtocolHeader(b, length, commandStoreOffset)
 
 	writeString(b, consumer.options.ConsumerName)
 	writeString(b, consumer.options.streamName)

pkg/stream/consumer_test.go

@@ -218,7 +218,7 @@ var _ = Describe("Streaming Consumers", func() {
 		consumer, err := env.NewConsumer(streamName,
 			func(consumerContext ConsumerContext, message *amqp.Message) {
 				atomic.AddInt32(&messagesCount, 1)
-				_ = consumerContext.Consumer.Commit()
+				_ = consumerContext.Consumer.StoreOffset()
 			}, NewConsumerOptions().SetOffset(OffsetSpecification{}.First()).SetConsumerName("consumer_test"))
 		Expect(err).NotTo(HaveOccurred())
 		time.Sleep(500 * time.Millisecond)