Merge pull request #2 from atlassian/add-channel-id-support-to-owner

check channelId also for job's ownerId equality

Author: Ahmet Halaç

queue/job.go

@@ -74,7 +74,8 @@ func (j *job) Execute() error {
 	messageAttr := j.sqsMessage().MessageAttributes
 
 	if messageAttr == nil ||
-		*messageAttr[ownerId].StringValue != j.ownerId {
+		*messageAttr[ownerId].StringValue != j.ownerId &&
+			*messageAttr[channelId].StringValue != j.ownerId {
 		j.state = jobError
 		return errors.Errorf("Message[%s] is invalid, will not be processed.", messageId)
 	}

queue/job_test.go

@@ -163,7 +163,7 @@ func TestExecuteWithInvalidQueueMessage(t *testing.T) {
 	sqsJob := newJobTest()
 
 	falseIntegrationId := "falseIntegrationId"
-	messageAttr := map[string]*sqs.MessageAttributeValue{ownerId: {StringValue: &falseIntegrationId}}
+	messageAttr := map[string]*sqs.MessageAttributeValue{ownerId: {StringValue: &falseIntegrationId}, channelId: {StringValue: &falseIntegrationId}}
 	sqsJob.message = sqs.Message{MessageAttributes: messageAttr, MessageId: &mockMessageId}
 
 	err := sqsJob.Execute()

queue/message_test.go

@@ -19,6 +19,7 @@ var (
 	mockApiKey    = "mockApiKey"
 	mockBaseUrl   = "mockBaseUrl"
 	mockOwnerId   = "mockOwnerId"
+	mockChannelId = "mockChannelId"
 )
 
 var mockActionSpecs = conf.ActionSpecifications{

queue/sqs_provider.go

@@ -11,6 +11,7 @@ import (
 )
 
 const ownerId = "ownerId"
+const channelId = "channelId"
 
 type SQSClient interface {
 	ChangeMessageVisibility(input *sqs.ChangeMessageVisibilityInput) (*sqs.ChangeMessageVisibilityOutput, error)
@@ -111,6 +112,7 @@ func (qp *sqsProvider) ReceiveMessage(maxNumOfMessage int64, visibilityTimeout i
 	request := &sqs.ReceiveMessageInput{
 		MessageAttributeNames: []*string{
 			aws.String(ownerId),
+			aws.String(channelId),
 		},
 		QueueUrl:            &queueUrl,
 		MaxNumberOfMessages: aws.Int64(maxNumOfMessage),

queue/sqs_provider_test.go

@@ -109,8 +109,9 @@ func TestReceiveMessage(t *testing.T) {
 	assert.Equal(t, mockQueueUrl1, *capturedInput.QueueUrl)
 	assert.Equal(t, int64(0), *capturedInput.WaitTimeSeconds) // because of short polling
 	assert.Equal(t, int64(10), *capturedInput.MaxNumberOfMessages)
-	assert.Equal(t, 1, len(capturedInput.MessageAttributeNames))
+	assert.Equal(t, 2, len(capturedInput.MessageAttributeNames))
 	assert.Equal(t, "ownerId", *capturedInput.MessageAttributeNames[0])
+	assert.Equal(t, "channelId", *capturedInput.MessageAttributeNames[1])
 }
 
 func TestReceiveMessageWithError(t *testing.T) {
@@ -233,7 +234,7 @@ var mockSuccessReceiveFunc = func(numOfMessage int64, visibilityTimeout int64) (
 	messages := make([]*sqs.Message, 0)
 	for i := int64(0); i < numOfMessage; i++ {
 		id := strconv.FormatInt(i, 10)
-		messageAttr := map[string]*sqs.MessageAttributeValue{"ownerId": {StringValue: &mockOwnerId}}
+		messageAttr := map[string]*sqs.MessageAttributeValue{"ownerId": {StringValue: &mockOwnerId}, "channelId": {StringValue: &mockChannelId}}
 		messages = append(messages, &sqs.Message{MessageId: &id, MessageAttributes: messageAttr, Body: &body})
 	}