package intimate import ( "database/sql" "encoding/binary" "fmt" "log" "reflect" "strconv" "time" ) // StoreExtractorDB 全局的Extractor DB 库链接 var StoreExtractorDB *Store // TStreamer 全局的Streamer. 在config init 完成初始化 var TStreamer *Table // TClog 全局的Clog var TClog *Table // TStreamerList 全局的streamer list 这个表存的url. 进去可以找到主播的列表. 便于动态更新 var TStreamerList *Table /*Store 结构体. 必须使用tag. field 数据库字段标签 uid 唯一id字段标签必须存在 */ type Store struct { db *sql.DB } type Table struct { store *Store name string setting interface{} updatesql string selectsql string insertsql string } // const updatesql = "UPDATE %s SET %s WHERE %s = ?" func NewStore(uri string) *Store { db, err := sql.Open("mysql", uri) if err != nil { panic(err) } s := &Store{db: db} return s } // Table 选择表. func (store *Store) Table(name string) *Table { table := &Table{store: store} table.name = name table.insertsql = `INSERT INTO ` + table.name + `(%s) values(%s)` table.updatesql = `UPDATE ` + table.name + ` SET %s WHERE %s = ?` // table.selectsql = `FROM ` + table.name + `WHERE operator` table.selectsql = `SELECT %s FROM ` + table.name + ` WHERE %s ` return table } // Queue mysql 队列结构 type Queue struct { table *Table obj reflect.Type selected string condition string uidname string uididx int } // OperatorType 字典Operator 标志位的类型 type OperatorType string const ( // OpOK 正常 OpOK OperatorType = "0" // OpWAIT 等待处理 OpWAIT OperatorType = "1000" // OpERROR 错误处理 OpERROR OperatorType = "10000" ) // Queue 根据Table生成一个队列. 处理结构. 每次弹出一个 obj 是要处理的结构体 自定义的whereCondition条件 func (t *Table) Queue(obj interface{}, whereCondition string) *Queue { q := &Queue{} q.condition = whereCondition q.obj = reflect.TypeOf(obj) q.table = t for i := 0; i < q.obj.NumField(); i++ { field := q.obj.Field(i) if fname, ok := field.Tag.Lookup("field"); ok { q.selected += fname + "," if _, ok := field.Tag.Lookup("uid"); ok { q.uididx = i q.uidname = fname } } } q.selected = q.selected[:len(q.selected)-1] return q } // Pop 队列弹出一个数据(任务). 参考队列处理 不支持嵌套. func (queue *Queue) Pop() (result interface{}, err error) { db := queue.table.store.db tx, err := db.Begin() if err != nil { return nil, err } defer func() { cerr := tx.Commit() if cerr != nil { log.Println(cerr) log.Println(tx.Rollback()) } }() selectsql := `SELECT ` + queue.selected + ` FROM ` + queue.table.name + ` WHERE ` + queue.condition + " limit 1 for update" rows, err := tx.Query(selectsql) if err != nil { return nil, fmt.Errorf("table: %s queue is empty", queue.table.name) } var fields = make([]interface{}, queue.obj.NumField()) for i := range fields { var iv interface{} fields[i] = &iv } if rows.Next() { err = rows.Scan(fields...) if err != nil { return nil, err } } columntypes, err := rows.ColumnTypes() if err != nil { return nil, err } if err = rows.Close(); err != nil { return nil, err } _, err = tx.Exec("UPDATE "+queue.table.name+" SET operator = "+string(OpWAIT)+" WHERE "+queue.uidname+" = ?", fields[queue.uididx]) if err != nil { log.Println(err) return nil, err } obj := reflect.New(queue.obj).Elem() for i := 0; i < obj.NumField(); i++ { field := obj.Field(i) convert(*fields[i].(*interface{}), field, columntypes[i]) } return obj.Addr().Interface(), err } // Insert nil 不插入. 不支持嵌套. func (t *Table) Insert(obj interface{}) error { ov := reflect.ValueOf(obj).Elem() ot := reflect.TypeOf(obj) fieldsql := "" argssql := "" var args []interface{} for i := 0; i < ov.NumField(); i++ { field := ov.Field(i) ftype := ot.Elem().Field(i) if fname, ok := ftype.Tag.Lookup("field"); ok { if flag, ok := ftype.Tag.Lookup("uid"); ok { if flag == "auto" { continue } } k := ftype.Type.Kind() if k == reflect.Ptr || k == reflect.Interface { if !field.IsNil() { felem := field.Elem() args = append(args, felem.Interface()) fieldsql += fname + "," argssql += "?," } } else { args = append(args, field.Interface()) fieldsql += fname + "," argssql += "?," } } } ssql := fmt.Sprintf(t.insertsql, fieldsql[:len(fieldsql)-1], argssql[:len(argssql)-1]) _, err := t.store.db.Exec(ssql, args...) return err } // Update 结构体更新 func (t *Table) Update(obj interface{}) error { ov := reflect.ValueOf(obj).Elem() ot := reflect.TypeOf(obj) fieldsql := "" var uidname string var uidvalue interface{} var args []interface{} for i := 0; i < ov.NumField(); i++ { field := ov.Field(i) ftype := ot.Elem().Field(i) if fname, ok := ftype.Tag.Lookup("field"); ok { if _, ok := ftype.Tag.Lookup("uid"); ok { if uidvalue != nil { panic(fmt.Errorf("uid must unique, %s and %s", uidname, fname)) } uidname = fname uidvalue = field.Interface() continue } k := ftype.Type.Kind() if k == reflect.Ptr || k == reflect.Interface { if !field.IsNil() { felem := field.Elem() args = append(args, felem.Interface()) fieldsql += fname + " = ?," } } else { args = append(args, field.Interface()) fieldsql += fname + " = ?," } } } if uidvalue == nil { panic(fmt.Errorf("update must contain `uid` tag")) } usql := fmt.Sprintf(t.updatesql, fieldsql[:len(fieldsql)-1], uidname) args = append(args, uidvalue) _, err := t.store.db.Exec(usql, args...) return err } func assign(field reflect.Value, src interface{}) (bool, error) { switch field.Kind() { case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: s := asString(src) i64, err := strconv.ParseInt(s, 10, field.Type().Bits()) if err != nil { err = strconvErr(err) return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err) } field.SetInt(i64) case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64: s := asString(src) u64, err := strconv.ParseUint(s, 10, field.Type().Bits()) if err != nil { err = strconvErr(err) return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err) } field.SetUint(u64) case reflect.Float32, reflect.Float64: s := asString(src) f64, err := strconv.ParseFloat(s, field.Type().Bits()) if err != nil { err = strconvErr(err) return false, fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err) } field.SetFloat(f64) case reflect.String: field.SetString(string(src.([]byte))) case reflect.Interface: return true, nil } return false, nil } func convert(src interface{}, field reflect.Value, columntype *sql.ColumnType) error { // log.Println("type:", field.Type(), ",kind:", field.Kind(), ",field:", field, "scanType:", columntype.ScanType(), "databaseType:", columntype.DatabaseTypeName()) if field.Kind() == reflect.Ptr { fn := field.Type().Elem() // New 一个 field.Type().Elem() . 然后判断 columntype 转化 成 NullString Time field.Set(reflect.New(fn)) field = field.Elem() // log.Println("type:", fn.Name(), ",kind:", field.Kind(), ",fieldtype:", field.Type()) } // log.Println(field.Kind(), field, reflect.TypeOf(field).Elem().Name(), columntype.ScanType().Kind()) if src == nil { return fmt.Errorf("converting NULL to %s is unsupported", field.Kind()) } switch columntype.DatabaseTypeName() { case "TINYINT", "SMALLINT", "MEDIUMINT", "INT", "BIGINT": isdefault, err := assign(field, src) if err != nil { return err } if isdefault { s := asString(src) i64, err := strconv.ParseInt(s, 10, 64) if err != nil { err = strconvErr(err) return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err) } // reflect.New(reflect.TypeOf(i64)) field.Set(reflect.ValueOf(i64)) } case "FLOAT", "DOUBLE", "DECIMAL": isdefault, err := assign(field, src) if err != nil { return err } if isdefault { s := asString(src) f64, err := strconv.ParseFloat(s, 64) if err != nil { err = strconvErr(err) return fmt.Errorf("converting driver.Value type %T (%q) to a %s: %v", src, s, field.Kind(), err) } field.Set(reflect.ValueOf(f64)) } case "BINARY", "VARBINARY", "TINYBLOB", "BLOB", "MEDIUMBLOB", "LONGBLOB", "JSON": isdefault, err := assign(field, src) if err != nil { return err } if isdefault { field.Set(reflect.ValueOf(src.([]byte))) } case "CHAR", "VARCHAR", "TINYTEXT", "TEXT", "MEDIUMTEXT", "LONGTEXT": isdefault, err := assign(field, src) if err != nil { return err } if isdefault { field.Set(reflect.ValueOf(string(src.([]byte)))) } case "BIT": var bits []byte = make([]byte, 8) copy(bits, src.([]byte)) switch field.Kind() { case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64: field.SetInt(int64(binary.LittleEndian.Uint64(bits))) case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64: field.SetUint(binary.LittleEndian.Uint64(bits)) case reflect.Interface: field.Set(reflect.ValueOf(binary.LittleEndian.Uint64(bits))) } case "YEAR", "TIME", "DATE", "DATETIME", "TIMESTAMP": s := src.(time.Time) switch field.Interface().(type) { case time.Time: field.Set(reflect.ValueOf(src)) case string: field.SetString(s.Format(time.RFC3339Nano)) case []byte: field.SetBytes([]byte(s.Format(time.RFC3339Nano))) default: } } // log.Println(fv, columntype.ScanType().Kind()) if iscan, ok := field.Addr().Interface().(sql.Scanner); ok { err := iscan.Scan(src) if err != nil { return err } } return nil }