Merge pull request #7 from kayac/fix/credentials

Fix bugs.

examples/dynamodb.tf

@@ -0,0 +1,31 @@
+resource "aws_dynamodb_table" "cron" {
+  name           = "Cron"
+  read_capacity  = 5
+  write_capacity = 5
+  hash_key       = "Id"
+  range_key      = "Type"
+
+  attribute {
+    name = "Id"
+    type = "S"
+  }
+
+  attribute {
+    name = "Type"
+    type = "S"
+  }
+
+  attribute {
+    name = "Expired"
+    type = "N"
+  }
+
+  global_secondary_index {
+    name               = "TypeExpiredIndex"
+    hash_key           = "Type"
+    range_key          = "Expired"
+    write_capacity     = 5
+    read_capacity      = 5
+    projection_type    = "ALL"
+  }
+}

job.go

@@ -88,7 +88,7 @@ func (j *DefaultJob) Execute(lkr lock.Locker) ([]byte, error) {
 	}
 
 	// 2. Locks Job (if job's lockID have been locked already, retry to Execute() after 5sec).
-	if j.lockID != "" && lkr != nil {
+	if j.lockID != "" && j.eventID != "" && lkr != nil {
 		err := lkr.Lock(j.lockID, j.eventID)
 		if err != nil {
 			logger.Errorf(err.Error())
@@ -114,15 +114,12 @@ func (j *DefaultJob) Execute(lkr lock.Locker) ([]byte, error) {
 	cmd.Env = env
 	output, err := cmd.CombinedOutput()
 
-	// return if not set locker
-	if lkr == nil {
-		return output, err
-	}
-
 	// 5. Unlocks job.
-	if derr := lkr.Unlock(j.lockID); derr != nil {
-		// TODO: should implement notification
-		logger.Errorf(derr.Error())
+	if j.lockID != "" && lkr != nil {
+		if derr := lkr.Unlock(j.lockID); derr != nil {
+			// TODO: should implement notification
+			logger.Errorf(derr.Error())
+		}
 	}
 
 	return output, err

lock/dynamodb_lock.go

@@ -99,12 +99,19 @@ func (dl DynamodbLock) Unlock(lockID string) error {
 
 // NewDynamodbLock build DynamodbLock
 func NewDynamodbLock(profile, region, table string) Locker {
-	cred := credentials.NewSharedCredentials("", profile)
+	var conf *aws.Config
+	if profile != "" {
+		conf = &aws.Config{
+			Region:      aws.String(region),
+			Credentials: credentials.NewSharedCredentials("", profile),
+		}
+	} else {
+		conf = &aws.Config{
+			Region: aws.String(region),
+		}
+	}
 	// configure Dynamodb
-	ddb := dynamodb.New(session.New(), &aws.Config{
-		Region:      &region,
-		Credentials: cred,
-	})
+	ddb := dynamodb.New(session.New(), conf)
 	return DynamodbLock{
 		TableName: table,
 		dynamodb:  ddb,

throttle/dynamodb_throttle.go

@@ -150,13 +150,20 @@ func (dt *DynamodbThrottle) GetWriteCapacity() (int64, error) {
 
 // NewDynamodbThrottle build DynamodbThrottle
 func NewDynamodbThrottle(ctx context.Context, profile, region, table string, retention time.Duration) Throttler {
-	cred := credentials.NewSharedCredentials("", profile)
+	var conf *aws.Config
+	if profile != "" {
+		conf = &aws.Config{
+			Region:      aws.String(region),
+			Credentials: credentials.NewSharedCredentials("", profile),
+		}
+	} else {
+		conf = &aws.Config{
+			Region: aws.String(region),
+		}
+	}
 
 	// configure Dynamodb
-	ddb := dynamodb.New(session.New(), &aws.Config{
-		Region:      aws.String(region),
-		Credentials: cred,
-	})
+	ddb := dynamodb.New(session.New(), conf)
 
 	dt := &DynamodbThrottle{
 		TableName:       table,