Merge pull request #25 from DocMerlin/master

return an error when concurrency is set to less than 1

Author: twang-rs

firehose.go

@@ -60,6 +60,9 @@ func (p *Producer) firehoseFlush(counter *int, timer *time.Time) bool {
 
 // Initialize a producer for Kinesis Firehose with the params supplied in the configuration file
 func (p *Producer) Firehose() (*Producer, error) {
+	if conf.Concurrency.Producer < 1 {
+		return nil, BadConcurrencyError
+	}
 	p.setConcurrency(conf.Concurrency.Producer)
 	p.initChannels()
 	auth, err := authenticate(conf.AWS.AccessKey, conf.AWS.SecretKey)
@@ -78,6 +81,9 @@ func (p *Producer) Firehose() (*Producer, error) {
 
 // Initialize a producer for Kinesis Firehose with the specified params
 func (p *Producer) FirehoseC(stream, accessKey, secretKey, region string, concurrency int) (*Producer, error) {
+	if concurrency < 1 {
+		return nil, BadConcurrencyError
+	}
 	if stream == "" {
 		return nil, NullStreamError
 	}

listener.go

@@ -43,7 +43,9 @@ type Listener struct {
 
 func (l *Listener) init(stream, shard, shardIterType, accessKey, secretKey, region string, concurrency int) (*Listener, error) {
 	var err error
-
+	if concurrency < 1 {
+		return nil, BadConcurrencyError
+	}
 	if stream == "" {
 		return nil, NullStreamError
 	}

producer.go

@@ -24,6 +24,7 @@ const (
 var (
 	ThroughputExceededError = errors.New("Configured AWS Kinesis throughput has been exceeded!")
 	KinesisFailureError     = errors.New("AWS Kinesis internal failure.")
+	BadConcurrencyError     = errors.New("Concurrency must be greater than zero.")
 )
 
 // Producer keeps a queue of messages on a channel and continually attempts
@@ -54,7 +55,9 @@ type Producer struct {
 
 func (p *Producer) init(stream, shard, shardIterType, accessKey, secretKey, region string, concurrency int) (*Producer, error) {
 	var err error
-
+	if concurrency < 1 {
+		return nil, BadConcurrencyError
+	}
 	if stream == "" {
 		return nil, NullStreamError
 	}