dev.arief.transaction

connection.go

@@ -2,25 +2,35 @@ package flexmgo
 
 import (
 	"context"
+	"fmt"
 	"strings"
 	"time"
 
-	"git.eaciitapp.com/sebar/dbflex"
+	"git.kanosolution.net/kano/dbflex"
 	"github.com/eaciit/toolkit"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readconcern"
+	"go.mongodb.org/mongo-driver/mongo/writeconcern"
 )
 
 type Connection struct {
 	dbflex.ConnectionBase `bson:"-" json:"-"`
 	ctx                   context.Context
 	client                *mongo.Client
 	db                    *mongo.Database
+	sess                  mongo.Session
 }
 
 func (c *Connection) Connect() error {
+	configString := "?"
+	for k, v := range c.Config {
+		configString += k + "=" + v.(string) + "&"
+	}
+
 	connURI := "mongodb://"
 	connURI += c.Host + "/"
+	connURI += configString
 
 	opts := options.Client().ApplyURI(connURI)
 	//opts.SetConnectTimeout(5 * time.Second)
@@ -40,6 +50,22 @@ func (c *Connection) Connect() error {
 		case "serverselectiontimeout":
 			opts.SetServerSelectionTimeout(
 				time.Duration(toolkit.ToInt(v, toolkit.RoundingAuto)) * time.Millisecond)
+
+		case "replicaset":
+			opts.SetReplicaSet(v.(string))
+			//opts.SetWriteConcern()
+
+		case "poolsize":
+			poolSize := toolkit.ToInt(v.(string), toolkit.RoundingAuto)
+			if poolSize > 0 {
+				opts.SetMaxPoolSize(uint64(poolSize))
+			}
+
+		case "idle":
+			idle := toolkit.ToInt(v.(string), toolkit.RoundingAuto)
+			if idle > 0 {
+				opts.SetMaxConnIdleTime(time.Duration(idle) * time.Second)
+			}
 		}
 	}
 
@@ -51,7 +77,7 @@ func (c *Connection) Connect() error {
 
 	//toolkit.Logger().Debug("client generated: OK")
 	if c.ctx == nil {
-		c.ctx = context.Background()
+		c.ctx = context.TODO()
 	}
 
 	//toolkit.Logger().Debug("context generated: OK")
@@ -103,48 +129,52 @@ func (c *Connection) DropTable(name string) error {
 	return c.db.Collection(name).Drop(c.ctx)
 }
 
-/*
-func (c *Connection) Prepare(dbflex.ICommand) (dbflex.IQuery, error) {
-	panic("not implemented")
-}
-
-func (c *Connection) Execute(dbflex.ICommand, toolkit.M) (interface{}, error) {
-	panic("not implemented")
-}
+func (c *Connection) BeginTx() error {
+	wc := writeconcern.New(writeconcern.WMajority())
+	rc := readconcern.Snapshot()
+	txnOpts := options.Transaction().SetWriteConcern(wc).SetReadConcern(rc)
 
-func (c *Connection) Cursor(dbflex.ICommand, toolkit.M) dbflex.ICursor {
-	panic("not implemented")
-}
+	if c.sess != nil {
+		return fmt.Errorf("session already exist. Pls commit or rollback last")
+	}
 
-func (c *Connection) NewQuery() dbflex.IQuery {
-	panic("not implemented")
+	sess, err := c.client.StartSession()
+	if err != nil {
+		return fmt.Errorf("unable to start new transaction. %s", err.Error())
+	}
+	sess.StartTransaction(txnOpts)
+	c.sess = sess
+	return nil
 }
 
-func (c *Connection) ObjectNames(dbflex.ObjTypeEnum) []string {
-	panic("not implemented")
-}
+func (c *Connection) Commit() error {
+	if c.sess == nil {
+		return fmt.Errorf("transaction session is not exists yet")
+	}
 
-func (c *Connection) ValidateTable(interface{}, bool) error {
-	panic("not implemented")
-}
+	err := c.sess.CommitTransaction(c.ctx)
+	if err != nil {
+		return fmt.Errorf("unable to commit. %s", err.Error())
+	}
 
-func (c *Connection) DropTable(string) error {
-	panic("not implemented")
+	c.sess = nil
+	return nil
 }
 
-func (c *Connection) SetThis(dbflex.IConnection) dbflex.IConnection {
-	panic("not implemented")
-}
+func (c *Connection) RollBack() error {
+	if c.sess == nil {
+		return fmt.Errorf("transaction session is not exists yet")
+	}
 
-func (c *Connection) This() dbflex.IConnection {
-	panic("not implemented")
-}
+	err := c.sess.AbortTransaction(c.ctx)
+	if err != nil {
+		return fmt.Errorf("unable to rollback. %s", err.Error())
+	}
 
-func (c *Connection) SetFieldNameTag(string) {
-	panic("not implemented")
+	c.sess = nil
+	return nil
 }
 
-func (c *Connection) FieldNameTag() string {
-	panic("not implemented")
+func (c *Connection) IsTx() bool {
+	return c.sess != nil
 }
-*/

cursor.go

@@ -4,8 +4,9 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+	"time"
 
-	"git.eaciitapp.com/sebar/dbflex"
+	"git.kanosolution.net/kano/dbflex"
 	"github.com/eaciit/toolkit"
 	"go.mongodb.org/mongo-driver/mongo"
 )
@@ -20,71 +21,161 @@ type Cursor struct {
 	cursor    *mongo.Cursor
 }
 
-func (cr *Cursor) Close() {
+func (cr *Cursor) Close() error {
+	e := cr.Error()
 	if cr.mc != nil {
 		cr.mc.Close(cr.conn.ctx)
 	}
+	return e
 }
 
 func (cr *Cursor) Count() int {
-	sr := cr.conn.db.RunCommand(cr.conn.ctx, cr.countParm)
-	if sr.Err() != nil {
-		dbflex.Logger().Errorf("unablet to get count. %s", sr.Err().Error())
-		return -1
+	if cr.countParm == nil || len(cr.countParm) == 0 {
+		return 0
 	}
 
-	countModel := new(struct{ N int })
-	if err := sr.Decode(countModel); err != nil {
-		dbflex.Logger().Errorf("unablet to decode count. %s", sr.Err().Error())
-		return -1
+	if cr.countParm.Get("count") == "" {
+		return 0
 	}
-	return countModel.N
+
+	tableName := cr.countParm.GetString("count")
+	where := cr.countParm.Get("query", nil)
+	if where == nil {
+		n, _ := cr.conn.db.Collection(tableName).CountDocuments(cr.conn.ctx, toolkit.M{})
+		return int(n)
+	} else {
+		n, _ := cr.conn.db.Collection(tableName).CountDocuments(cr.conn.ctx, where)
+		return int(n)
+	}
+
+	/*
+		sr := cr.conn.db.RunCommand(cr.conn.ctx, cr.countParm)
+		if sr.Err() != nil {
+			dbflex.Logger().Errorf("unable to get count. %s, countparm: %s",
+				sr.Err().Error(),
+				toolkit.JsonString(cr.countParm))
+			return 0
+		}
+
+		countModel := new(struct{ N int })
+		if err := sr.Decode(countModel); err != nil {
+			dbflex.Logger().Errorf("unablet to decode count. %s", sr.Err().Error())
+			return 0
+		}
+		return countModel.N
+	*/
 }
 
-func (cr *Cursor) Fetch(out interface{}) error {
+func (cr *Cursor) Fetch(out interface{}) dbflex.ICursor {
 	if cr.Error() != nil {
-		return toolkit.Errorf("unable to fetch data. %s", cr.Error())
+		cr.SetError(toolkit.Errorf("unable to fetch data. %s", cr.Error()))
+		return cr
 	}
 
 	if neof := cr.cursor.Next(cr.conn.ctx); !neof {
-		return io.EOF
+		cr.SetError(io.EOF)
+		return cr
 	}
 
-	if err := cr.cursor.Decode(out); err != nil {
-		return toolkit.Errorf("unable to decode output. %s", err.Error())
+	m := toolkit.M{}
+	if err := cr.cursor.Decode(&m); err != nil {
+		cr.SetError(toolkit.Errorf("unable to decode output. %s", err.Error()))
+		return cr
+	}
+	for mk, mv := range m {
+		// update date value to date
+		if mvs, ok := mv.(string); ok && len(mvs) >= 11 {
+			if mvs[4] == '-' && mvs[7] == '-' && mvs[10] == 'T' {
+				if dt, err := time.Parse(time.RFC3339, mvs); err == nil {
+					m.Set(mk, dt)
+				}
+			}
+		}
+	}
+	if reflect.ValueOf(m).Type().String() == reflect.Indirect(reflect.ValueOf(out)).Type().String() {
+		reflect.ValueOf(out).Elem().Set(reflect.ValueOf(m))
+	} else {
+		if err := toolkit.Serde(m, out, ""); err != nil {
+			cr.SetError(toolkit.Errorf("unable to decode output. %s", err.Error()))
+			return cr
+		}
 	}
 
-	return nil
+	return cr
 }
 
-func (cr *Cursor) Fetchs(result interface{}, n int) error {
+func (cr *Cursor) Fetchs(result interface{}, n int) dbflex.ICursor {
 	if cr.Error() != nil {
-		return toolkit.Errorf("unable to fetch data. %s", cr.Error())
+		cr.SetError(toolkit.Errorf("unable to fetch data. %s", cr.Error()))
+		return cr
 	}
 
-	v := reflect.TypeOf(result).Elem().Elem()
-	ivs := reflect.MakeSlice(reflect.SliceOf(v), 0, 0)
-
+	/*
+		v := reflect.TypeOf(result).Elem().Elem()
+		ivs := reflect.MakeSlice(reflect.SliceOf(v), 0, 0)
+
+			read := 0
+			for {
+				if !cr.cursor.Next(cr.conn.ctx) {
+					break
+				}
+
+				iv := reflect.New(v).Interface()
+				err := cr.cursor.Decode(iv)
+				if err != nil {
+					cr.SetError(fmt.Errorf("unable to decode cursor data. %s", err.Error()))
+					return cr
+				}
+				ivs = reflect.Append(ivs, reflect.ValueOf(iv).Elem())
+
+				read++
+				if n != 0 && read == n {
+					break
+				}
+			}
+			reflect.ValueOf(result).Elem().Set(ivs)
+	*/
 	read := 0
+	ms := []toolkit.M{}
 	for {
 		if !cr.cursor.Next(cr.conn.ctx) {
 			break
 		}
 
-		iv := reflect.New(v).Interface()
-		err := cr.cursor.Decode(iv)
+		m := toolkit.M{}
+		err := cr.cursor.Decode(&m)
 		if err != nil {
-			return fmt.Errorf("unable to decode cursor data. %s", err.Error())
+			cr.SetError(fmt.Errorf("unable to decode cursor data. %s", err.Error()))
+			return cr
+		}
+		for mk, mv := range m {
+			// update date value to date
+			if mvs, ok := mv.(string); ok && len(mvs) >= 11 {
+				if mvs[4] == '-' && mvs[7] == '-' && mvs[10] == 'T' {
+					if dt, err := time.Parse(time.RFC3339, mvs); err == nil {
+						m.Set(mk, dt)
+						//fmt.Println(mk, mvs, dt, m, fmt.Sprintf("%t", m.Get("Created")))
+					}
+				}
+			}
 		}
-		ivs = reflect.Append(ivs, reflect.ValueOf(iv).Elem())
+		ms = append(ms, m)
 
 		read++
 		if n != 0 && read == n {
 			break
 		}
 	}
-	reflect.ValueOf(result).Elem().Set(ivs)
-	return nil
+	if reflect.ValueOf(ms).Type().String() == reflect.Indirect(reflect.ValueOf(result)).Type().String() {
+		reflect.ValueOf(result).Elem().Set(reflect.ValueOf(ms))
+	} else {
+		if err := toolkit.Serde(ms, result, ""); err != nil {
+			cr.SetError(fmt.Errorf("unable to decode cursor data. %s", err.Error()))
+			return cr
+		}
+	}
+
+	return cr
 }
 
 /*

flexmgo.go

@@ -1,6 +1,6 @@
 package flexmgo
 
-import "git.eaciitapp.com/sebar/dbflex"
+import "git.kanosolution.net/kano/dbflex"
 
 func init() {
 	dbflex.RegisterDriver("mongodb", func(si *dbflex.ServerInfo) dbflex.IConnection {