Merge branch 'master' into master

Author: twang-rs

auth.go

@@ -0,0 +1,19 @@
+package kinetic
+
+import (
+	"errors"
+	gokinesis "github.com/rewardStyle/go-kinesis"
+)
+
+var MetaAuthenticationErr = errors.New("Authentication error: failed to auth from meta.  Your IAM roles are bad, or you need to specify an AccessKey and SecretKey")
+
+func authenticate(accessKey, secretKey string) (auth *gokinesis.AuthCredentials, err error) {
+	if accessKey == "" || secretKey == "" {
+		if auth, err = gokinesis.NewAuthFromMetadata(); err != nil {
+			return nil, MetaAuthenticationErr
+		}
+	} else {
+		auth = gokinesis.NewAuth(accessKey, secretKey)
+	}
+	return
+}

firehose.go

@@ -65,10 +65,13 @@ func (p *Producer) Firehose() (*Producer, error) {
 	}
 	p.setConcurrency(conf.Concurrency.Producer)
 	p.initChannels()
-
+	auth, err := authenticate(conf.AWS.AccessKey, conf.AWS.SecretKey)
 	p.kinesis = &kinesis{
 		stream: conf.Firehose.Stream,
-		client: gokinesis.NewWithEndpoint(gokinesis.NewAuth(conf.AWS.AccessKey, conf.AWS.SecretKey), conf.AWS.Region, fmt.Sprintf(firehoseURL, conf.AWS.Region)),
+		client: gokinesis.NewWithEndpoint(auth, conf.AWS.Region, fmt.Sprintf(firehoseURL, conf.AWS.Region)),
+	}
+	if err != nil {
+		return p, err
 	}
 
 	p.producerType = firehoseType
@@ -87,10 +90,13 @@ func (p *Producer) FirehoseC(stream, accessKey, secretKey, region string, concur
 
 	p.setConcurrency(concurrency)
 	p.initChannels()
-
+	auth, err := authenticate(accessKey, secretKey)
 	p.kinesis = &kinesis{
 		stream: stream,
-		client: gokinesis.NewWithEndpoint(gokinesis.NewAuth(accessKey, secretKey), region, fmt.Sprintf(firehoseURL, region)),
+		client: gokinesis.NewWithEndpoint(auth, region, fmt.Sprintf(firehoseURL, region)),
+	}
+	if err != nil {
+		return p, err
 	}
 
 	p.producerType = firehoseType

kinesis.go

@@ -65,14 +65,19 @@ type kinesis struct {
 }
 
 func (k *kinesis) init(stream, shard, shardIteratorType, accessKey, secretKey, region string) (*kinesis, error) {
+
+	auth, err := authenticate(accessKey, secretKey)
 	k = &kinesis{
 		stream:            stream,
 		shard:             shard,
 		shardIteratorType: shardIteratorType,
-		client:            gokinesis.New(gokinesis.NewAuth(accessKey, secretKey), region),
+		client:            gokinesis.New(auth, region),
+	}
+	if err != nil {
+		return k, err
 	}
 
-	err := k.initShardIterator()
+	err = k.initShardIterator()
 	if err != nil {
 		return k, err
 	}

kinesis_test.go

@@ -0,0 +1,24 @@
+package kinetic
+
+import (
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+// TestKineticCreation tests to make sure that the Kinetic creation doesn't return early or return a nil.
+func TestKineticCreation(t *testing.T) {
+	kinesisKinetic, err := new(kinesis).init("fake", "ShardId-00000001", "TRIM_HORIZON", "BADaccessKey", "BADsecretKey", "region")
+
+	Convey("Given an badly configured init-ed kinetic", t, func() {
+		Convey("the error returned should not be nil", func() {
+			So(err, ShouldNotBeNil)
+		})
+		Convey("the returned kinesis struct pointer should not be nil", func() {
+			So(kinesisKinetic, ShouldNotBeNil)
+		})
+		Convey("it should also have some data in it", func() {
+			So(kinesisKinetic.stream, ShouldEqual, "fake")
+			So(kinesisKinetic.shard, ShouldEqual, "ShardId-00000001")
+		})
+	})
+}