Add manual commit

Signed-off-by: Gabriele Santomaggio <g.santomaggio@gmail.com>

Author: Gsantomaggio

examples/offsetTracking/offsetTracking.go

@@ -71,6 +71,7 @@ func main() {
 		streamName,
 		handleMessages,
 		stream.NewConsumerOptions().
+			SetManualCommit().                               // disable auto commit
 			SetConsumerName("my_consumer").                  // set a consumer name
 			SetOffset(stream.OffsetSpecification{}.First())) // start consuming from the beginning
 	CheckErr(err)

pkg/stream/super_stream_consumer.go

@@ -15,6 +15,7 @@ type SuperStreamConsumerOptions struct {
 	SingleActiveConsumer *SingleActiveConsumer
 	ConsumerName         string
 	AutoCommitStrategy   *AutoCommitStrategy
+	Autocommit           bool
 }
 
 func NewSuperStreamConsumerOptions() *SuperStreamConsumerOptions {
@@ -49,10 +50,16 @@ func (s *SuperStreamConsumerOptions) SetConsumerName(consumerName string) *Super
 }
 
 func (s *SuperStreamConsumerOptions) SetAutoCommit(autoCommitStrategy *AutoCommitStrategy) *SuperStreamConsumerOptions {
+	s.Autocommit = true
 	s.AutoCommitStrategy = autoCommitStrategy
 	return s
 }
 
+func (s *SuperStreamConsumerOptions) SetManualCommit() *SuperStreamConsumerOptions {
+	s.Autocommit = false
+	return s
+}
+
 // CPartitionClose is a struct that is used to notify the user when a partition from a consumer is closed
 // The user can use the NotifyPartitionClose to get the channel
 type CPartitionClose struct {
@@ -167,8 +174,11 @@ func (s *SuperStreamConsumer) ConnectPartition(partition string, offset OffsetSp
 	}
 
 	options = options.SetFilter(s.SuperStreamConsumerOptions.Filter)
-	if s.SuperStreamConsumerOptions.AutoCommitStrategy != nil {
+
+	if s.SuperStreamConsumerOptions.Autocommit {
 		options = options.SetAutoCommit(s.SuperStreamConsumerOptions.AutoCommitStrategy)
+	} else {
+		options = options.SetManualCommit()
 	}
 
 	if s.SuperStreamConsumerOptions.SingleActiveConsumer != nil {