Method parseRow

consumer.go:597–623

(ctx context.Context, rows *sqlx.Rows)

	var msgs []*MessageIncoming
	for rows.Next() {
		msg, err := c.parseRow(ctx, rows)
		if err != nil {
			return nil, errors.WithStack(err)
		}
		msgs = append(msgs, msg)
	}
	if err := rows.Err(); err != nil {
		return nil, errors.WithStack(err)
	}
	if len(msgs) == 0 {
		return nil, sql.ErrNoRows
	}
	if err := tx.Commit(); err != nil {
		return nil, errors.Wrap(err, "commit message consumption")
	}
	return msgs, nil
}

func (c *Consumer) parseRow(ctx context.Context, rows *sqlx.Rows) (*MessageIncoming, error) {
	var pgMsg pgMessage
	if err := rows.Scan(
		&pgMsg.ID,
		&pgMsg.Payload,
		&pgMsg.Metadata,
		&pgMsg.Attempt,
		&pgMsg.LockedUntil,
	); err != nil {
		if isErrorCode(err, undefinedTableErrCode, undefinedColumnErrCode) {
			return nil, fatalError{Err: err}