// Run sits on a loop reading and processing messages off of the AWS SQS queue.
Run sits on a loop reading and processing messages off of the AWS SQS queue.
// Run sits on a loop reading and processing messages off of the AWS SQS queue .
func (p *MessageProcessor) Run() {
	for {
		r, err := p.client.ReceiveMessage(&sqs.ReceiveMessageInput{
			MaxNumberOfMessages: aws.Int64(p.maxNumberOfMessages),
			MessageAttributeNames: []*string{
				aws.String(messageAttributeNameAll),
			},
			QueueUrl:          aws.String(p.queueURL),
			VisibilityTimeout: aws.Int64(p.visibilityTimeoutSeconds),
			WaitTimeSeconds:   aws.Int64(p.maxWaitTimeSeconds),
		})
		if err != nil {
			log.Errorf("Failed to receive message: %v", err)
			continue
		}
		if len(r.Messages) < 1 {
			continue
		}
		var wg sync.WaitGroup
		wg.Add(len(r.Messages))
		for _, message := range r.Messages {
			go func(message *sqs.Message) {
				defer wg.Done()
				log.Tracef("Processing message with id %q", *message.MessageId)
				var (
					body  string
					topic string
				)
				if message.Body == nil {
					body = ""
				} else {
					body = *message.Body
				}
				if v, ok := message.MessageAttributes[messageAttributeNameTopic]; !ok || v == nil || *v.StringValue == "" {
					topic = p.queueURL
				} else {
					topic = *v.StringValue
				}
				p.controller.InvokeWithContext(buildMessageContext(message), topic, pointers.NewBytes([]byte(body)))
			}(message)
		}
		wg.Wait()
	}
}
func ( p * MessageProcessor ) Run ( ) { for { r , err := p . client . ReceiveMessage ( & sqs . ReceiveMessageInput { MaxNumberOfMessages : aws . Int64 ( p . maxNumberOfMessages ) , MessageAttributeNames : [ ] * string { aws . String ( messageAttributeNameAll ) , } , QueueUrl : aws . String ( p . queueURL ) , VisibilityTimeout : aws . Int64 ( p . visibilityTimeoutSeconds ) , WaitTimeSeconds : aws . Int64 ( p . maxWaitTimeSeconds ) , } ) if err != nil { log . Errorf ( " " , err ) continue } if len ( r . Messages ) < 1 { continue } var wg sync . WaitGroup wg . Add ( len ( r . Messages ) ) for _ , message := range r . Messages { go func ( message * sqs . Message ) { defer wg . Done ( ) log . Tracef ( " " , * message . MessageId ) var ( body string topic string ) if message . Body == nil { body = " " } else { body = * message . Body } if v , ok := message . MessageAttributes [ messageAttributeNameTopic ] ; ! ok || v == nil || * v . StringValue == " " { topic = p . queueURL } else { topic = * v . StringValue } p . controller . InvokeWithContext ( buildMessageContext ( message ) , topic , pointers . NewBytes ( [ ] byte ( body ) ) ) } ( message ) } wg . Wait ( ) } }
Run sits on a loop reading and processing messages off of the AWS SQS queue.
Run sits on a loop reading and processing messages off of the AWS SQS queue .
// Retrieve the message's body (if any). // Retrieve the message's topic (if any). // Invoke the function(s) associated with the topic.