pollQueue polls the given queue as long as the given context is alive.
(ctx context.Context, sqsurl string)
| 190 | |
| 191 | // pollQueue polls the given queue as long as the given context is alive. |
| 192 | func (s *SQS) pollQueue(ctx context.Context, sqsurl string) { |
| 193 | ctxLog := log.WithFields(log.Fields{"f": "SQS.pollQueue", "url": sqsurl}) |
| 194 | backoff := awsutils.DefaultBackoff |
| 195 | for { |
| 196 | resp, err := s.svc.ReceiveMessageWithContext(ctx, &sqs.ReceiveMessageInput{ |
| 197 | QueueUrl: aws.String(sqsurl), |
| 198 | WaitTimeSeconds: aws.Int64(20), |
| 199 | // We ask only for 1 message at a time, because the |
| 200 | // parseFile() call below could block, and we want to |
| 201 | // receive messages and not process them immediately, |
| 202 | // or they could get rescheduled to other readers. |
| 203 | MaxNumberOfMessages: aws.Int64(1), |
| 204 | }) |
| 205 | if ctx.Err() == context.Canceled || ctx.Err() == context.DeadlineExceeded { |
| 206 | return |
| 207 | } |
| 208 | |
| 209 | if err != nil { |
| 210 | ctxLog.WithError(err).Error("error from ReceiveMessage") |
| 211 | time.Sleep(backoff.Duration()) |
| 212 | continue |
| 213 | } |
| 214 | backoff.Reset() |
| 215 | |
| 216 | for _, msg := range resp.Messages { |
| 217 | s3FilePath, err := s.parse(*msg.Body) |
| 218 | if err != nil { |
| 219 | ctxLog.WithError(err).Error("error parsing message") |
| 220 | continue |
| 221 | } |
| 222 | |
| 223 | // Try to unescape URL. Voluntarily ignore an error here since the |
| 224 | // result might be a valid URL anyway. |
| 225 | s3FilePath, _ = url.QueryUnescape(s3FilePath) |
| 226 | |
| 227 | // Skip the file if it doesn't match the filter provided. |
| 228 | if s.filepathRx == nil || s.filepathRx.MatchString(s3FilePath) { |
| 229 | // FIXME: we should check if the bucket matches what was configured |
| 230 | // or even better, change s3Input to not be limited to a single bucket |
| 231 | s.s3Input.ParseFile(s3FilePath) |
| 232 | } |
| 233 | |
| 234 | _, err = s.svc.DeleteMessageWithContext(ctx, &sqs.DeleteMessageInput{ |
| 235 | QueueUrl: aws.String(sqsurl), |
| 236 | ReceiptHandle: msg.ReceiptHandle, |
| 237 | }) |
| 238 | if ctx.Err() == context.Canceled || ctx.Err() == context.DeadlineExceeded { |
| 239 | return |
| 240 | } |
| 241 | if err != nil { |
| 242 | ctxLog.WithError(err).Error("error from DeleteMessage") |
| 243 | } |
| 244 | } |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | func (s *SQS) Run(inch chan<- *baker.Data) error { |
| 249 | s.s3Input.SetOutputChannel(inch) |
no test coverage detected