Refactored the signature of the function option methods to operate on the main (public) objects instead of the respective (private) option configuration structs.

Author: Jason Yu

checkpointer.go

@@ -10,7 +10,7 @@ import (
 	"time"
 )
 
-// checkpointElement is the struct used to store checkpointing information.
+// checkpointElement is the struct used to store checkpointing information in the doubly linked list.
 type checkpointElement struct {
 	seqNum int
 	done   bool
@@ -46,7 +46,7 @@ func (c *checkpointList) insert(seqNum int) error {
 	return nil
 }
 
-// find is a method used to retreive the element in the doubly linked list for a given sequence number.  find is
+// find is a method used to retrieve the element in the doubly linked list for a given sequence number.  find is
 // optimized for searching oldest numbers first as it traverse the linked list starting from the beginning.
 func (c *checkpointList) find(seqNum int) (*list.Element, bool) {
 	for e := c.Front(); e != nil; e = e.Next() {
@@ -83,11 +83,11 @@ func defaultCheckpointOptions() *checkpointOptions {
 
 // checkpointOptionsFn is a function signature used to define function options for configuring all of the
 // configurable options of a checkpoint object.
-type checkpointOptionsFn func(*checkpointOptions) error
+type checkpointOptionsFn func(*checkpointer) error
 
 // checkpointAutoCheckpointCount is a functional option method for configuring the checkpoint's auto checkpoint count.
 func checkpointAutoCheckpointCount(count int) checkpointOptionsFn {
-	return func(o *checkpointOptions) error {
+	return func(o *checkpointer) error {
 		o.autoCheckpointCount = count
 		return nil
 	}
@@ -96,23 +96,23 @@ func checkpointAutoCheckpointCount(count int) checkpointOptionsFn {
 // checkpointAutoCheckpointFreq is a functional option method for configuring the checkpoint's auto checkpoint
 // frequency.
 func checkpointAutoCheckpointFreq(freq time.Duration) checkpointOptionsFn {
-	return func(o *checkpointOptions) error {
+	return func(o *checkpointer) error {
 		o.autoCheckpointFreq = freq
 		return nil
 	}
 }
 
 // checkpointCheckpointFn is a functional option method for configuring the checkpoint's checkpoint callback function.
 func checkpointCheckpointFn(fn func(string) error) checkpointOptionsFn {
-	return func(o *checkpointOptions) error {
+	return func(o *checkpointer) error {
 		o.checkpointFn = fn
 		return nil
 	}
 }
 
 // checkpointCountCheckFreq is a functional option method for configuring the checkpoint's count check frequency.
 func checkpointCountCheckFreq(freq time.Duration) checkpointOptionsFn {
-	return func(o *checkpointOptions) error {
+	return func(o *checkpointer) error {
 		o.countCheckFreq = freq
 		return nil
 	}
@@ -138,14 +138,15 @@ type checkpointer struct {
 // newCheckpoint instantiates a new checkpoint object with default configuration settings unless the function option
 // methods are provided to change the default values.
 func newCheckpointer(optionFns ...checkpointOptionsFn) *checkpointer {
-	checkpointOptions := defaultCheckpointOptions()
-	for _, optionFn := range optionFns {
-		optionFn(checkpointOptions)
-	}
-	return &checkpointer{
-		checkpointOptions: checkpointOptions,
+	checkpointer := &checkpointer{
+		checkpointOptions: defaultCheckpointOptions(),
 		list:              &checkpointList{list.New()},
 	}
+	for _, optionFn := range optionFns {
+		optionFn(checkpointer)
+	}
+
+	return checkpointer
 }
 
 // startup is a method used to enable automatic checkpointing.
@@ -231,7 +232,8 @@ func (c *checkpointer) insert(seqNumStr string) error {
 	return nil
 }
 
-// markDone safely marks the given sequence number as done.
+// markDone safely marks the given sequence number as done and attempts to remove it's previous element if the
+// previous element is also marked done or attempts to remove itself if it's next element is also marked done
 func (c *checkpointer) markDone(seqNumStr string) error {
 	c.listMu.Lock()
 	defer c.listMu.Unlock()

consumer.go

@@ -29,19 +29,19 @@ func defaultConsumerOptions() *consumerOptions {
 }
 
 // ConsumerOptionsFn is a method signature for defining functional option methods for configuring the Consumer.
-type ConsumerOptionsFn func(*consumerOptions) error
+type ConsumerOptionsFn func(*Consumer) error
 
 // ConsumerReader is a functional option method for configuring the consumer's stream reader.
 func ConsumerReader(r StreamReader) ConsumerOptionsFn {
-	return func(o *consumerOptions) error {
+	return func(o *Consumer) error {
 		o.reader = r
 		return nil
 	}
 }
 
 // ConsumerQueueDepth is a functional option method for configuring the consumer's queueDepth.
 func ConsumerQueueDepth(depth int) ConsumerOptionsFn {
-	return func(o *consumerOptions) error {
+	return func(o *Consumer) error {
 		if depth > 0 {
 			o.queueDepth = depth
 			return nil
@@ -52,7 +52,7 @@ func ConsumerQueueDepth(depth int) ConsumerOptionsFn {
 
 // ConsumerConcurrency is a functional option method for configuring the consumer's concurrency.
 func ConsumerConcurrency(count int) ConsumerOptionsFn {
-	return func(o *consumerOptions) error {
+	return func(o *Consumer) error {
 		if count > 0 {
 			o.concurrency = count
 			return nil
@@ -63,15 +63,15 @@ func ConsumerConcurrency(count int) ConsumerOptionsFn {
 
 // ConsumerLogLevel is a functional option method for configuring the consumer's log level.
 func ConsumerLogLevel(ll aws.LogLevelType) ConsumerOptionsFn {
-	return func(o *consumerOptions) error {
+	return func(o *Consumer) error {
 		o.logLevel = ll & 0xffff0000
 		return nil
 	}
 }
 
 // ConsumerStats is a functional option method for configuring the consumer's stats collector.
 func ConsumerStats(sc ConsumerStatsCollector) ConsumerOptionsFn {
-	return func(o *consumerOptions) error {
+	return func(o *Consumer) error {
 		o.Stats = sc
 		return nil
 	}
@@ -93,24 +93,25 @@ type Consumer struct {
 
 // NewConsumer creates a new Consumer object for retrieving and listening to message(s) on a StreamReader.
 func NewConsumer(c *aws.Config, stream string, shard string, optionFns ...ConsumerOptionsFn) (*Consumer, error) {
-	consumerOptions := defaultConsumerOptions()
+	consumer := &Consumer{consumerOptions: defaultConsumerOptions()}
 	for _, optionFn := range optionFns {
-		optionFn(consumerOptions)
+		optionFn(consumer)
 	}
-	if consumerOptions.reader == nil {
+
+	if consumer.reader == nil {
 		r, err := NewKinesisReader(c, stream, shard)
 		if err != nil {
 			return nil, err
 		}
-		consumerOptions.reader = r
+		consumer.reader = r
+	}
+
+	consumer.LogHelper = &LogHelper{
+		LogLevel: consumer.logLevel,
+		Logger:   c.Logger,
 	}
-	return &Consumer{
-		consumerOptions: consumerOptions,
-		LogHelper: &LogHelper{
-			LogLevel: consumerOptions.logLevel,
-			Logger:   c.Logger,
-		},
-	}, nil
+
+	return consumer, nil
 }
 
 // startConsuming will initialize the message channel and set consuming to true if there is not already another consume

firehose_writer.go

@@ -39,12 +39,12 @@ func defaultFirehoseWriterOptions() *firehoseWriterOptions {
 
 // FirehoseWriterOptionsFn is a method signature for defining functional option methods for configuring
 // the FirehoseWriter.
-type FirehoseWriterOptionsFn func(*firehoseWriterOptions) error
+type FirehoseWriterOptionsFn func(*FirehoseWriter) error
 
 // FirehoseWriterMsgCountRateLimit is a functional option method for configuring the FirehoseWriter's
 // message count rate limit.
 func FirehoseWriterMsgCountRateLimit(limit int) FirehoseWriterOptionsFn {
-	return func(o *firehoseWriterOptions) error {
+	return func(o *FirehoseWriter) error {
 		if limit > 0 && limit <= firehoseMsgCountRateLimit {
 			o.msgCountRateLimit = limit
 			return nil
@@ -56,7 +56,7 @@ func FirehoseWriterMsgCountRateLimit(limit int) FirehoseWriterOptionsFn {
 // FirehoseWriterMsgSizeRateLimit is a functional option method for configuring the FirehoseWriter's
 // messsage size rate limit.
 func FirehoseWriterMsgSizeRateLimit(limit int) FirehoseWriterOptionsFn {
-	return func(o *firehoseWriterOptions) error {
+	return func(o *FirehoseWriter) error {
 		if limit > 0 && limit <= firehoseMsgSizeRateLimit {
 			o.msgSizeRateLimit = limit
 			return nil
@@ -68,7 +68,7 @@ func FirehoseWriterMsgSizeRateLimit(limit int) FirehoseWriterOptionsFn {
 // FirehoseWriterThroughputMultiplier is a functional option method for configuring the FirehoseWriter's
 // throughput multiplier.
 func FirehoseWriterThroughputMultiplier(multiplier int) FirehoseWriterOptionsFn {
-	return func(o *firehoseWriterOptions) error {
+	return func(o *FirehoseWriter) error {
 		if multiplier > 0 {
 			o.throughputMultiplier = multiplier
 			return nil
@@ -79,15 +79,15 @@ func FirehoseWriterThroughputMultiplier(multiplier int) FirehoseWriterOptionsFn
 
 // FirehoseWriterLogLevel is a functional option method for configuring the FirehoseWriter's log level.
 func FirehoseWriterLogLevel(ll aws.LogLevelType) FirehoseWriterOptionsFn {
-	return func(o *firehoseWriterOptions) error {
+	return func(o *FirehoseWriter) error {
 		o.logLevel = ll & 0xffff0000
 		return nil
 	}
 }
 
 // FirehoseWriterStats is a functional option method for configuring the FirehoseWriter's stats collector.
 func FirehoseWriterStats(sc ProducerStatsCollector) FirehoseWriterOptionsFn {
-	return func(o *firehoseWriterOptions) error {
+	return func(o *FirehoseWriter) error {
 		o.Stats = sc
 		return nil
 	}
@@ -103,23 +103,26 @@ type FirehoseWriter struct {
 
 // NewFirehoseWriter creates a new stream writer to write records to a Kinesis.
 func NewFirehoseWriter(c *aws.Config, stream string, optionFns ...FirehoseWriterOptionsFn) (*FirehoseWriter, error) {
-	firehoseWriterOptions := defaultFirehoseWriterOptions()
-	for _, optionFn := range optionFns {
-		optionFn(firehoseWriterOptions)
-	}
 	sess, err := session.NewSession(c)
 	if err != nil {
 		return nil, err
 	}
-	return &FirehoseWriter{
+
+	firehoseWriter := &FirehoseWriter{
+		firehoseWriterOptions: defaultFirehoseWriterOptions(),
 		stream:                stream,
 		client:                firehose.New(sess),
-		firehoseWriterOptions: firehoseWriterOptions,
-		LogHelper: &LogHelper{
-			LogLevel: firehoseWriterOptions.logLevel,
-			Logger:   c.Logger,
-		},
-	}, nil
+	}
+	for _, optionFn := range optionFns {
+		optionFn(firehoseWriter)
+	}
+
+	firehoseWriter.LogHelper = &LogHelper{
+		LogLevel: firehoseWriter.logLevel,
+		Logger:   c.Logger,
+	}
+
+	return firehoseWriter, nil
 }
 
 // PutRecords sends a batch of records to Firehose and returns a list of records that need to be retried.

kcl_reader.go

@@ -43,11 +43,11 @@ func defaultKclReaderOptions() *kclReaderOptions {
 }
 
 // KclReaderOptionsFn is a method signature for defining functional option methods for configuring the KclReader.
-type KclReaderOptionsFn func(*kclReaderOptions) error
+type KclReaderOptionsFn func(*KclReader) error
 
 // kclReaderBatchSize is a functional option method for configuring the KclReader's batch size
 func kclReaderBatchSize(size int) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		if size > 0 && size <= kclReaderMaxBatchSize {
 			o.batchSize = size
 			return nil
@@ -58,15 +58,15 @@ func kclReaderBatchSize(size int) KclReaderOptionsFn {
 
 // KclReaderAutoCheckpointCount is a functional option method for configuring the KclReader's checkpoint count
 func KclReaderAutoCheckpointCount(count int) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.autoCheckpointCount = count
 		return nil
 	}
 }
 
 // KclReaderAutoCheckpointFreq is a functional option method for configuring the KclReader's checkpoint frequency
 func KclReaderAutoCheckpointFreq(freq time.Duration) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.autoCheckpointFreq = freq
 		return nil
 	}
@@ -75,7 +75,7 @@ func KclReaderAutoCheckpointFreq(freq time.Duration) KclReaderOptionsFn {
 // KclReaderUpdateCheckpointSizeFreq is a functional option method for configuring the KclReader's
 // update checkpoint size stats frequency
 func KclReaderUpdateCheckpointSizeFreq(freq time.Duration) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.updateCheckpointSizeFreq = freq
 		return nil
 	}
@@ -84,7 +84,7 @@ func KclReaderUpdateCheckpointSizeFreq(freq time.Duration) KclReaderOptionsFn {
 // KclReaderOnInitCallbackFn is a functional option method for configuring the KclReader's
 // onInitCallbackFn.
 func KclReaderOnInitCallbackFn(fn func() error) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.onInitCallbackFn = fn
 		return nil
 	}
@@ -93,7 +93,7 @@ func KclReaderOnInitCallbackFn(fn func() error) KclReaderOptionsFn {
 // KclReaderOnCheckpointCallbackFn is a functional option method for configuring the KclReader's
 // onCheckpointCallbackFn.
 func KclReaderOnCheckpointCallbackFn(fn func(seqNum string, err string) error) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.onCheckpointCallbackFn = fn
 		return nil
 	}
@@ -102,23 +102,23 @@ func KclReaderOnCheckpointCallbackFn(fn func(seqNum string, err string) error) K
 // KclReaderOnShutdownCallbackFn is a functional option method for configuring the KclReader's
 // onShutdownCallbackFn.
 func KclReaderOnShutdownCallbackFn(fn func() error) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.onShutdownCallbackFn = fn
 		return nil
 	}
 }
 
 // KclReaderLogLevel is a functional option method for configuring the KclReader's log level.
 func KclReaderLogLevel(ll aws.LogLevelType) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.logLevel = ll
 		return nil
 	}
 }
 
 // KclReaderStats is a functional option method for configuring the KclReader's stats collector.
 func KclReaderStats(sc ConsumerStatsCollector) KclReaderOptionsFn {
-	return func(o *kclReaderOptions) error {
+	return func(o *KclReader) error {
 		o.Stats = sc
 		return nil
 	}
@@ -140,17 +140,17 @@ type KclReader struct {
 
 // NewKclReader creates a new stream reader to read records from KCL
 func NewKclReader(c *aws.Config, optionFns ...KclReaderOptionsFn) (*KclReader, error) {
-	kclReaderOptions := defaultKclReaderOptions()
+	kclReader := &KclReader{kclReaderOptions: defaultKclReaderOptions()}
 	for _, optionFn := range optionFns {
-		optionFn(kclReaderOptions)
+		optionFn(kclReader)
 	}
-	return &KclReader{
-		kclReaderOptions: kclReaderOptions,
-		LogHelper: &LogHelper{
-			LogLevel: kclReaderOptions.logLevel,
-			Logger:   c.Logger,
-		},
-	}, nil
+
+	kclReader.LogHelper = &LogHelper{
+		LogLevel: kclReader.logLevel,
+		Logger:   c.Logger,
+	}
+
+	return kclReader, nil
 }
 
 func (r *KclReader) process(ctx context.Context) {