(exec execer, msgID pgtype.UUID)
| 713 | if !isJSONObject(pgMsg.Metadata) { |
| 714 | return nil, errors.New("metadata is invalid JSON object") |
| 715 | } |
| 716 | return nil, errors.Wrap(err, "parsing metadata") |
| 717 | } |
| 718 | return m, nil |
| 719 | } |
| 720 | |
| 721 | func isJSONObject(b json.RawMessage) bool { |
| 722 | if !json.Valid(b) { |
| 723 | return false |
| 724 | } |
| 725 | // remove insignificant characters. |
| 726 | b = bytes.TrimLeftFunc(b, unicode.IsSpace) |
| 727 | return bytes.HasPrefix(b, []byte{'{'}) |
| 728 | } |
| 729 | |
| 730 | type execer interface { |
| 731 | ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) |
| 732 | } |
| 733 | |
| 734 | func (c *Consumer) ackMessage(exec execer, msgID pgtype.UUID) func(ctx context.Context) error { |
| 735 | query := `UPDATE ` + pg.QuoteIdentifier(c.queueName) + ` SET locked_until = NULL, processed_at = CURRENT_TIMESTAMP WHERE id = $1` |
| 736 | return func(ctx context.Context) error { |
| 737 | if _, err := exec.ExecContext(ctx, query, msgID); err != nil { |
| 738 | c.metrics.failedProcessingCounter.Add(ctx, 1, |
no test coverage detected