run.go

package database

// Stop, when returned by fn parameter of Run, instructs
// Run to stop iterating rows.
var Stop = errors.New("stop Run execution")

// Run executes a query and calls fn for each row.
//
// fn must be a function with parameters matching the rows
// returned by the query. For example, for a query
//
//   SELECT id, name from datacenters
//
// fn can be
//
//   func(id int, name string)
//
// fn may return an error. If the returned error is Stop,
// Run stop reading rows and returns nil.
// Otherwise, if it is not nil, Run returns that error as is.
func Run(c context.Context, query string, fn interface{}) error {
	fv := reflect.ValueOf(fn)
	ft := fv.Type()
	switch {
	case fv.Kind() != reflect.Func:
		panic("fn is not a function")
	case ft.NumOut() > 1:
		panic("fn has more than one return value")
	case ft.NumOut() == 1 && ft.Out(0) != reflect.TypeOf((error)(nil)):
		panic("fn return type is not an error")
	}

	db := Get(c)
	rows, err := db.QueryContext(c, query)
	if err != nil {
		return err
	}
	defer rows.Close()

	valuePtrs := make([]*interface{}, ft.NumIn())
	for i := range valuePtrs {
		valuePtrs[i] = new(interface{})
	}
	reflectValues := make([]reflect.Value, len(valuePtrs))

	for rows.Next() {
		if err = rows.Scan(valuePtrs...); err != nil {
			return errors.Annotate(err, "failed to scan a row").Err()
		}

		for i, ptr := range valuePtrs {
			reflectValues[i] = reflect.ValueOf(*ptr)
		}
		if ret := fv.Call(reflectValues); len(ret) > 0 {
			switch err := ret[0].Interface().(error); {
			case err == Stop:
				return nil
			case err != nil:
				return err
			}
		}
	}
	return nil
}