close finalizer

crud.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 	"time"
 
-	mgo "gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
 )
 
@@ -28,7 +27,7 @@ func (e *ErrorWithStack) Stack() string {
 }
 
 // CreateDocument creates a document from specified model
-func CreateDocument(db *mgo.Database, m ModelInterface) error {
+func (db *MgoDatabase) CreateDocument(m ModelInterface) error {
 	m.PrepareInsert()
 
 	c := db.C(GetCollectionName(m))
@@ -38,7 +37,7 @@ func CreateDocument(db *mgo.Database, m ModelInterface) error {
 }
 
 // ReadDocument gets one document via its id
-func ReadDocument(db *mgo.Database, m ModelInterface, selector bson.M) error {
+func (db *MgoDatabase) ReadDocument(m ModelInterface, selector bson.M) error {
 	c := db.C(GetCollectionName(m))
 
 	q := c.FindId(m.GetID())
@@ -137,7 +136,7 @@ func idToObjectID(filter interface{}) {
 }
 
 // ReadCollection gets the filtered collection of the model
-func ReadCollection(db *mgo.Database, results interface{}, filter bson.M, selector bson.M, offset int, limit int, sort []string, pipelineModifier PipelineModifierFunction) (err error) {
+func (db *MgoDatabase) ReadCollection(results interface{}, filter bson.M, selector bson.M, offset int, limit int, sort []string, pipelineModifier PipelineModifierFunction) (err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			err = &ErrorWithStack{Message: fmt.Sprintf("%v", r), StackTrace: string(debug.Stack())}
@@ -212,8 +211,7 @@ func ReadCollection(db *mgo.Database, results interface{}, filter bson.M, select
 			pipeline = pipelineModifier(pipeline)
 		}
 
-		q := c.Pipe(pipeline).AllowDiskUse().Iter()
-		_err = q.All(results)
+		_err = c.Pipe(pipeline).All(results)
 	} else {
 		// search without pipe is faster
 
@@ -245,7 +243,7 @@ func ReadCollection(db *mgo.Database, results interface{}, filter bson.M, select
 }
 
 // ReadCollectionCount gets the count of elements in filtered collection
-func ReadCollectionCount(db *mgo.Database, m ModelInterface, filter bson.M) (count int, err error) {
+func (db *MgoDatabase) ReadCollectionCount(m ModelInterface, filter bson.M) (count int, err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			err = fmt.Errorf("%v", r)
@@ -257,7 +255,7 @@ func ReadCollectionCount(db *mgo.Database, m ModelInterface, filter bson.M) (cou
 }
 
 // UpdateDocument updates a document from specified model
-func UpdateDocument(db *mgo.Database, m ModelInterface, changes bson.M) error {
+func (db *MgoDatabase) UpdateDocument(m ModelInterface, changes bson.M) error {
 	m.PrepareUpdate()
 	changes["updateTime"] = time.Now()
 
@@ -268,7 +266,7 @@ func UpdateDocument(db *mgo.Database, m ModelInterface, changes bson.M) error {
 }
 
 // UpsertDocument updates a document from specified model or inserts it, of not found
-func UpsertDocument(db *mgo.Database, m ModelInterface, changes bson.M) error {
+func (db *MgoDatabase) UpsertDocument(m ModelInterface, changes bson.M) error {
 	m.PrepareUpdate()
 	changes["updateTime"] = time.Now()
 
@@ -279,7 +277,7 @@ func UpsertDocument(db *mgo.Database, m ModelInterface, changes bson.M) error {
 }
 
 // DeleteDocument deletes one document via its id
-func DeleteDocument(db *mgo.Database, m ModelInterface) error {
+func (db *MgoDatabase) DeleteDocument(m ModelInterface) error {
 	c := db.C(GetCollectionName(m))
 
 	err := c.RemoveId(m.GetID())
@@ -288,12 +286,12 @@ func DeleteDocument(db *mgo.Database, m ModelInterface) error {
 }
 
 // DeleteDocuments deletes documents found by filter
-func DeleteDocuments(db *mgo.Database, m ModelInterface, filter bson.M) (removed int, err error) {
+func (db *MgoDatabase) DeleteDocuments(m ModelInterface, filter bson.M) (removed int, err error) {
 	c := db.C(GetCollectionName(m))
 
 	info, err := c.RemoveAll(filter)
 	if info != nil {
-		removed = info.Removed
+		removed = info.Removed()
 	}
 
 	return removed, err

go.mod

@@ -1,4 +1,4 @@
-module gitbase.de/gopackage/mgocrud
+module gitbase.de/gopackage/mgocrud/v2
 
 go 1.16
 

lookup.go

@@ -4,12 +4,11 @@ import (
 	"fmt"
 	"reflect"
 
-	mgo "gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
 )
 
 // Lookup extends results with data for inline structs
-func Lookup(db *mgo.Database, structField string, results interface{}, selector bson.M) error {
+func (db *MgoDatabase) Lookup(structField string, results interface{}, selector bson.M) error {
 
 	t := reflect.TypeOf(results)
 	v := reflect.ValueOf(results)
@@ -94,7 +93,7 @@ func Lookup(db *mgo.Database, structField string, results interface{}, selector
 		// no entries to map
 		return nil
 	}
-	sArr := make([]bson.M, lArr, lArr)
+	sArr := make([]bson.M, lArr)
 	aI := 0
 	for sID := range objectIDs {
 		sArr[aI] = bson.M{
@@ -123,7 +122,7 @@ func Lookup(db *mgo.Database, structField string, results interface{}, selector
 		return fmt.Errorf("ID type in objects struct %+v is not bson.ObjectId", fieldType)
 	}
 
-	err := ReadCollection(db, objectResults, sQuery, selector, 0, 0, nil, nil)
+	err := db.ReadCollection(objectResults, sQuery, selector, 0, 0, nil, nil)
 	if err != nil {
 		panic(err)
 	}

mgo.go

@@ -0,0 +1,224 @@
+package mgocrud
+
+import (
+	"errors"
+	"fmt"
+	"runtime"
+
+	mgo "gopkg.in/mgo.v2"
+)
+
+type MgoConnection struct {
+	connection *mgo.Session
+	closed     bool
+}
+
+func NewMgoConnection(dial string) (Connection, error) {
+	fmt.Println("CONNECTION CREATE")
+	connection, err := mgo.Dial(dial)
+	if err != nil {
+		return nil, err
+	}
+	connection.SetMode(mgo.Monotonic, true)
+	c := &MgoConnection{connection: connection}
+	runtime.SetFinalizer(c, func(c *MgoConnection) {
+		fmt.Println("CONNECTION CLOSE FINALIZER")
+		if !c.closed {
+			c.Close()
+		}
+	})
+	return c, nil
+}
+
+func (c *MgoConnection) Close() {
+	fmt.Println("CONNECTION CLOSE MANUALLY")
+	if !c.closed {
+		c.connection.Close()
+		c.closed = true
+	}
+	runtime.SetFinalizer(c, nil)
+}
+
+type MgoSession struct {
+	session *mgo.Session
+	closed  bool
+}
+
+func (s *MgoSession) Close() {
+	fmt.Println("SESSION CLOSE MANUALLY")
+	if !s.closed {
+		s.session.Close()
+		s.closed = true
+	}
+	runtime.SetFinalizer(s, nil)
+}
+
+func (c *MgoConnection) NewSession() Session {
+	fmt.Println("SESSION CREATE")
+	s := &MgoSession{session: c.connection.Copy()}
+	runtime.SetFinalizer(s, func(s *MgoSession) {
+		fmt.Println("SESSION CLOSE FINALIZER")
+		if !s.closed {
+			s.Close()
+		}
+	})
+	return s
+}
+
+type MgoDatabase struct {
+	database *mgo.Database
+	session  Session
+}
+
+func (s *MgoSession) DB(name string) Database {
+	return &MgoDatabase{database: s.session.DB(name), session: s}
+}
+
+type MgoCollection struct {
+	collection *mgo.Collection
+}
+
+func (db *MgoDatabase) Session() Session {
+	return db.session
+}
+
+func (db *MgoDatabase) C(name string) Collection {
+	return &MgoCollection{collection: db.database.C(name)}
+}
+
+func (db *MgoDatabase) Name() string {
+	return db.database.Name
+}
+
+func (c *MgoCollection) Insert(docs ...interface{}) error {
+	return c.collection.Insert(docs...)
+}
+
+func (c *MgoCollection) UpdateId(id interface{}, update interface{}) error {
+	return c.collection.UpdateId(id, update)
+}
+
+func (c *MgoCollection) RemoveId(id interface{}) error {
+	return c.collection.RemoveId(id)
+}
+
+type MgoChangeInfo struct {
+	changeInfo *mgo.ChangeInfo
+}
+
+func (ci *MgoChangeInfo) Matched() int {
+	return ci.changeInfo.Matched
+}
+
+func (ci *MgoChangeInfo) Removed() int {
+	return ci.changeInfo.Removed
+}
+
+func (ci *MgoChangeInfo) Updated() int {
+	return ci.changeInfo.Updated
+}
+
+func (c *MgoCollection) Upsert(selector interface{}, update interface{}) (ChangeInfo, error) {
+	ci, err := c.collection.Upsert(selector, update)
+	if err != nil {
+		return nil, err
+	}
+	return &MgoChangeInfo{changeInfo: ci}, nil
+}
+
+func (c *MgoCollection) RemoveAll(filter interface{}) (ChangeInfo, error) {
+	ci, err := c.collection.RemoveAll(filter)
+	if err != nil {
+		return nil, err
+	}
+	return &MgoChangeInfo{changeInfo: ci}, nil
+}
+
+type MgoQuery struct {
+	query *mgo.Query
+}
+
+func (c *MgoCollection) FindId(id interface{}) Query {
+	return &MgoQuery{query: c.collection.FindId(id)}
+}
+
+func (c *MgoCollection) Find(query interface{}) Query {
+	return &MgoQuery{query: c.collection.Find(query)}
+}
+
+func (q *MgoQuery) Select(selector interface{}) Query {
+	q.query = q.query.Select(selector)
+	return q
+}
+
+func (q *MgoQuery) One(result interface{}) error {
+	err := q.query.One(result)
+	if err == mgo.ErrNotFound {
+		err = ErrNotFound
+	}
+	return err
+}
+
+func (q *MgoQuery) Sort(fields ...string) Query {
+	q.query = q.query.Sort(fields...)
+	return q
+}
+
+func (q *MgoQuery) Skip(n int) Query {
+	q.query = q.query.Skip(n)
+	return q
+}
+
+func (q *MgoQuery) Limit(n int) Query {
+	q.query = q.query.Limit(n)
+	return q
+}
+
+func (q *MgoQuery) All(result interface{}) error {
+	err := q.query.All(result)
+	if err == mgo.ErrNotFound {
+		err = ErrNotFound
+	}
+	return err
+}
+
+func (q *MgoQuery) Count() (int, error) {
+	c, err := q.query.Count()
+	if err == mgo.ErrNotFound {
+		err = ErrNotFound
+	}
+	return c, err
+}
+
+type MgoIndex struct {
+	index *mgo.Index
+}
+
+func NewMgoIndex(index mgo.Index) Index {
+	return &MgoIndex{index: &index}
+}
+
+func (c *MgoCollection) EnsureIndex(index Index) error {
+	if i, ok := index.(*MgoIndex); ok {
+		if i != nil && i.index != nil {
+			return c.collection.EnsureIndex(*i.index)
+		}
+	}
+	return errors.New("index parameter not initialized with mgo.Index")
+}
+
+type MgoPipe struct {
+	pipe *mgo.Pipe
+}
+
+func (p *MgoPipe) All(result interface{}) error {
+	err := p.pipe.All(result)
+	if err == mgo.ErrNotFound {
+		err = ErrNotFound
+	}
+	return err
+}
+
+func (c *MgoCollection) Pipe(pipeline interface{}) Pipe {
+	return &MgoPipe{pipe: c.collection.Pipe(pipeline).AllowDiskUse()}
+}

setup.go

@@ -9,7 +9,7 @@ import (
 )
 
 // EnsureIndex ensured mongodb index reflecting model struct index tag
-func EnsureIndex(db *mgo.Database, m ModelInterface) error {
+func (db *MgoDatabase) EnsureIndex(m ModelInterface) error {
 	colName := GetCollectionName(m)
 	col := db.C(colName)
 
@@ -57,14 +57,14 @@ func EnsureIndex(db *mgo.Database, m ModelInterface) error {
 							case indexEl == "text":
 								textFields = append(textFields, "$text:"+fieldbase+bsonField)
 							default:
-								return fmt.Errorf("invalid index tag on collection %s.%s for field %s%s in model %+v", db.Name, colName, fieldbase, bsonField, t)
+								return fmt.Errorf("invalid index tag on collection %s.%s for field %s%s in model %+v", db.Name(), colName, fieldbase, bsonField, t)
 							}
 
 						}
 						if len(index.Key) > 0 {
 							// fmt.Println(bsonField, index)
-							fmt.Printf("ensure index on collection %s.%s for field %s%s\n", db.Name, colName, fieldbase, bsonField)
-							err := col.EnsureIndex(index)
+							fmt.Printf("ensure index on collection %s.%s for field %s%s\n", db.Name(), colName, fieldbase, bsonField)
+							err := col.EnsureIndex(NewMgoIndex(index))
 							if err != nil {
 								return err
 							}
@@ -93,13 +93,13 @@ func EnsureIndex(db *mgo.Database, m ModelInterface) error {
 
 	if len(textFields) > 0 {
 		// fmt.Println("$text", textFields)
-		fmt.Printf("ensure text index on collection %s.%s for fields %v\n", db.Name, GetCollectionName(m), textFields)
-		err := col.EnsureIndex(mgo.Index{
+		fmt.Printf("ensure text index on collection %s.%s for fields %v\n", db.Name(), GetCollectionName(m), textFields)
+		err := col.EnsureIndex(NewMgoIndex(mgo.Index{
 			Name:            "textindex",
 			Key:             textFields,
 			DefaultLanguage: "german",
 			Background:      false,
-		})
+		}))
 		if err != nil {
 			return err
 		}

types.go

@@ -0,0 +1,71 @@
+package mgocrud
+
+import (
+	"errors"
+
+	"gopkg.in/mgo.v2/bson"
+)
+
+var (
+	ErrNotFound = errors.New("not found")
+)
+
+type Connection interface {
+	Close()
+	NewSession() Session
+}
+
+type Session interface {
+	Close()
+	DB(name string) Database
+}
+
+type Database interface {
+	Session() Session
+	C(name string) Collection
+	Name() string
+	EnsureIndex(m ModelInterface) error
+	ValidateObject(m ModelInterface, changes bson.M) error
+	ReadDocument(m ModelInterface, selector bson.M) error
+	CreateDocument(m ModelInterface) error
+	ReadCollection(results interface{}, filter bson.M, selector bson.M, offset int, limit int, sort []string, pipelineModifier PipelineModifierFunction) error
+	ReadCollectionCount(m ModelInterface, filter bson.M) (count int, err error)
+	UpdateDocument(m ModelInterface, changes bson.M) error
+	UpsertDocument(m ModelInterface, changes bson.M) error
+	DeleteDocument(m ModelInterface) error
+	DeleteDocuments(m ModelInterface, filter bson.M) (removed int, err error)
+}
+
+type Collection interface {
+	Insert(docs ...interface{}) error
+	UpdateId(id interface{}, update interface{}) error
+	RemoveId(id interface{}) error
+	Upsert(selector interface{}, update interface{}) (ChangeInfo, error)
+	RemoveAll(filter interface{}) (ChangeInfo, error)
+	FindId(id interface{}) Query
+	Find(query interface{}) Query
+	EnsureIndex(index Index) error
+	Pipe(pipeline interface{}) Pipe
+}
+
+type ChangeInfo interface {
+	Matched() int
+	Removed() int
+	Updated() int
+}
+
+type Query interface {
+	Select(selector interface{}) Query
+	One(result interface{}) error
+	Sort(fields ...string) Query
+	Skip(n int) Query
+	Limit(n int) Query
+	All(result interface{}) error
+	Count() (int, error)
+}
+
+type Index interface{}
+
+type Pipe interface {
+	All(result interface{}) error
+}

validator.go

@@ -2,12 +2,11 @@ package mgocrud
 
 import (
 	validator "gopkg.in/go-playground/validator.v8"
-	mgo "gopkg.in/mgo.v2"
 	"gopkg.in/mgo.v2/bson"
 )
 
 // ValidateObject validates object via validator tag and custom method
-func ValidateObject(db *mgo.Database, m ModelInterface, changes bson.M) error {
+func (db *MgoDatabase) ValidateObject(m ModelInterface, changes bson.M) error {
 	// first validate via struct tag
 	validator := validator.New(&validator.Config{
 		TagName: "validator",
@@ -19,7 +18,7 @@ func ValidateObject(db *mgo.Database, m ModelInterface, changes bson.M) error {
 
 	// next execute custom model validator if exists
 	if i, ok := m.(interface {
-		Validate(db *mgo.Database, changes bson.M) error
+		Validate(db Database, changes bson.M) error
 	}); ok {
 		return i.Validate(db, changes)
 	}