封装一个mgo的简单使用

mongoDao

package dao

import (
	"gopkg.in/mgo.v2/bson"
	"time"
	"gopkg.in/mgo.v2"
	"strings"

	"log"
)

type Mongodb struct {
	Standard      string
	Userid        string
	IsJob         bool
	IsGroupJob    bool
//	Job           *Job
	RstNum        int
	RstKeyArray   []string
	IsMailJobRst  bool
	MaxExecutionTime time.Duration
//	Ctx           *ripple.Context
	Head          bson.M
	Output        interface{}
	Url           string
}

const  (
	TDD_URL    			=  ""
	FDD_URL    			=  ""
	VIP_USER_MAX_EXECUTION_TIME	=  time.Second * 55
	GENERAL_USER_MAX_EXECUTION_TIME =  time.Second * 30
	MAX_BULK_WRITE_FILE_LEN         =  1024
)

var (
	mongodb     *Mongodb
	MgoSession  *mgo.Session
)

func GetMongodbInstance() *Mongodb {
	if nil  == mongodb{
		mongodb = new(Mongodb)
	}
	return mongodb
}

func (this *Mongodb) HandleError(err error, parentInfo string, returnVal []bson.M)  {
	errInfo := err.Error()
	if strings.Contains(errInfo, "Sort operation used more than the maximum 33554432 bytes of RAM. Add an index, or specify a smaller limit.") {
		returnVal[0]["statusCode"] = "500"
		returnVal[0]["errorInfo"] = "本次操作结果大小超过32M，系统排序失败，请缩小 筛选范围 或 增加筛选条件 后再尝试此操作"
		this.Output = returnVal
		log.Fatal("panic in location: ", parentInfo)
		panic("Fatal error in " + parentInfo)
 	}else if strings.Contains(errInfo, "operation exceeded time limit") || strings.Contains(errInfo, "i/o timeout") {
		returnVal[0]["statusCode"] = "500"
		returnVal[0]["errorInfo"] = "目前系统繁忙，导致您本次操作耗时太长而失败，请稍后再尝试此操作"
		this.Output = returnVal
		log.Fatal("panic in location: ", parentInfo)
		panic("Fatal error in " + parentInfo)
	}else if strings.Contains(errInfo, "not found") {
		log.Fatal("error in location: " + parentInfo)
		return
	}else {
		log.Fatal("panic in location: " + parentInfo)
		panic(err)
	}
}

func (this *Mongodb) HandleErrorTwo(err error, parentInfo string, returnVal bson.M) {
	errInfo := err.Error()
	if strings.Contains(errInfo, "Sort operation used more than the maximum 33554432 bytes of RAM. Add an index, or specify a smaller limit.") {
		returnVal["statusCode"] = "500"
		returnVal["errorInfo"] = "本次操作结果大小超过32M，系统排序失败，请缩小 筛选范围 或 增加筛选条件 后再尝试此操作"
		this.Output = returnVal
		log.Fatal("panic in location:", parentInfo)
		panic("Fatal error in " + parentInfo)
	} else if strings.Contains(errInfo, "operation exceeded time limit") || strings.Contains(errInfo, "i/o timeout") {
		returnVal["statusCode"] = "失败"
		returnVal["errorInfo"] = "目前系统繁忙，导致您本次操作耗时太长而失败，请稍后再尝试此操作"
		this.Output = returnVal
		log.Fatal("panic in location:", parentInfo)
		panic("Fatal error in " + parentInfo)
	} else if strings.Contains(errInfo, "not found") {
		log.Fatal("error in location:", parentInfo)
		return
	} else {
		returnVal["statusCode"] = "失败"
		returnVal["errorInfo"] = "未知错误"
		this.Output = returnVal
		log.Fatal("panic in location:", parentInfo)
		panic("Fatal error in " + parentInfo)
	}
}


func (this *Mongodb) HandleMutiRoutineError(err error, parentInfo string) {
	errInfo := err.Error()
	if strings.Contains(errInfo, "Sort operation used more than the maximum 33554432 bytes of RAM. Add an index, or specify a smaller limit.") {
		log.Fatal("panic in location:", parentInfo)
		panic(err)
	} else if strings.Contains(errInfo, "operation exceeded time limit") || strings.Contains(errInfo, "i/o timeout") {
		log.Fatal("panic in location:", parentInfo)
		panic(err)
	} else if strings.Contains(errInfo, "not found") {
		log.Fatal("error in location:", parentInfo)
		return
	} else {
		log.Fatal("panic in location:", parentInfo)
		panic(err)
	}
}

func (this *Mongodb) Catch(methodName string) {
	if r := recover(); nil != r {
		log.Println("At method:", methodName,
			", Runtime error caught:", r)
	}
}


func (this *Mongodb) SetMaxExecutionTime (userid string) {
	this.Userid = userid
	if true == this.IsJob{
		 this.MaxExecutionTime = time.Minute * 20
		return 
	}
	this.MaxExecutionTime = GENERAL_USER_MAX_EXECUTION_TIME
}

// 公共方法，获取session 如果存在则拷贝一份
func (this *Mongodb) GetSession() *mgo.Session {
	if MgoSession  ==  nil{
		var err error
		MgoSession, err = mgo.Dial(this.Url)
		if err != nil {
			panic(err)
		}
		MgoSession.SetSocketTimeout(time.Minute * 20)
	}
	// 最大连接池默认4096
	return MgoSession.Clone()
}

//获取collection对象执行操作
func (this *Mongodb) WithCollection(dataBase string, collection string, s func(*mgo.Collection)error) error {
	session := this.GetSession()
	defer session.Close()
	c := session.DB(dataBase).C(collection)
	return s(c)
}

//获取GridFS对象执行操作
func (this *Mongodb) WithGridFS(dataBase string, s func(*mgo.GridFS) error) error {
	session := this.GetSession()
	defer session.Close()
	gridsFS := session.DB(dataBase).GridFS("fs")
	return s(gridsFS)
}


func (this *Mongodb) GetOneRecord(dataBase, collection string, queryCondition, selector bson.M) (result bson.M, err error) {
	query := func(c *mgo.Collection) error {
		return c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Select(selector).One(&result)
	}	
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb GetOneRecord, query failed! db:", dataBase,
			", Collection:", collection,
			", queryCondition:", queryCondition,
			", selector:", selector,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) GetOneRecordWithSort(dataBase, collection string, queryCondition, selector bson.M, fields ...string) (result bson.M, err error) {
	query := func(c *mgo.Collection) error {
		return c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Select(selector).Sort(fields...).One(&result)
	}
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb GetOneRecordWithSort, query failed! db:", dataBase,
			", Collection:", collection,
			", queryCondition:", queryCondition,
			", selector:", selector,
			", sort:", fields,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) GetMultiRecord(dataBase, collection string, queryCondition, selector bson.M) (result []bson.M, err error) {
	query := func(c *mgo.Collection) error {
		return c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Select(selector).All(&result)
	}
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb GetMultiRecord, query failed! db:", dataBase,
			", Collection:", collection,
			", queryCondition:", queryCondition,
			", selector:", selector,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) GetMultiRecordWithSort(dataBase, collection string, queryCondition, selector bson.M, fields ...string) (result []bson.M, err error) {
	query := func(c *mgo.Collection) error {
		return c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Select(selector).Sort(fields...).All(&result)
	}
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb GetMultiRecordWithSort, query failed! db:", dataBase,
			", Collection:", collection,
			", queryCondition:", queryCondition,
			", selector:", selector,
			", sort:", fields,
			", ErrInfo:", err)
	}
	return
}


func (this *Mongodb) DeleteCollection(dataBase, collection string) (err error) {
	deleteOperetion := func(c *mgo.Collection) error {
		return c.DropCollection()
	}

	err = this.WithCollection(dataBase, collection, deleteOperetion)
	if err != nil {
		log.Println("mongodb DeleteCollection, deleteOperation failed! db:", dataBase,
			", Collection:", collection,
			", ErrInfo:", err)
		return
	}
	return
}




func (this *Mongodb) Count(dataBase, collection string, queryCondition bson.M) (total int, err error) {
	query := func(c *mgo.Collection) (err error) {
		total, err = c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Count()
		return
	}
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb Count, count failed, db:", dataBase,
			", Collection:", collection,
			", QueryCondition:", queryCondition,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) Pipe(dataBase, collection string, pipelineStage []bson.M) (result []interface{}, err error) {
	aggregate := func(c *mgo.Collection) error {
		return c.Pipe(pipelineStage).All(&result)
	}
	err = this.WithCollection(dataBase, collection, aggregate)
	if err != nil {
		log.Println("mongodb Pipe, pipe failed, db:", dataBase,
			", Collection:", collection,
			", PipelineStage:", pipelineStage,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) PipeAllowDiskUse(dataBase, collection string, pipelineStage []bson.M) (result []interface{}, err error) {
	aggregate := func(c *mgo.Collection) error {
		return c.Pipe(pipelineStage).AllowDiskUse().All(&result)
	}
	err = this.WithCollection(dataBase, collection, aggregate)
	if err != nil {
		log.Println("mongodb Pipe, PipeAllowDiskUse failed, db:", dataBase,
			", Collection:", collection,
			", PipelineStage:", pipelineStage,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) Distinct(dataBase, collection, fieldName string, queryCondition bson.M) (out []interface{}, err error) {
	query := func(c *mgo.Collection) error {
		return c.Find(queryCondition).SetMaxTime(this.MaxExecutionTime).Distinct(fieldName, &out)
	}
	err = this.WithCollection(dataBase, collection, query)
	if err != nil {
		log.Println("mongodb Distinct, distinct failed, db:", dataBase,
			", Collection:", collection,
			", FieldName:", fieldName,
			", QueryCondition:", queryCondition,
			", ErrInfo:", err)
	}
	return
}


func (this *Mongodb) UpdateOne(dataBase, collection string, selector interface{}, updater interface{}) (err error) {
	update := func(c *mgo.Collection) error {
		return c.Update(selector, updater)
	}
	err = this.WithCollection(dataBase, collection, update)
	if err != nil {
		log.Println("mongodb UpdateOne, updateOne Failed, db:", dataBase,
			", Collection:", collection,
			", Selector:", selector,
			", Updater:", updater,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) UpdateAll(dataBase, collection string, selector interface{}, updater interface{}) (err error) {
	update := func(c *mgo.Collection) error {
		_, err := c.UpdateAll(selector, updater)
		return err
	}
	err = this.WithCollection(dataBase, collection, update)
	if err != nil {
		log.Println("mongodb UpdateAll, UpdateAll Failed, db:", dataBase,
			", Collection:", collection,
			", Selector:", selector,
			", Updater:", updater,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) InsertOne(dataBase, collection string, oneRecord interface{}) (err error) {
	insert := func(c *mgo.Collection) error {
		return c.Insert(oneRecord)
	}
	err = this.WithCollection(dataBase, collection, insert)
	if err != nil {
		log.Println("mongodb InsertOne, InsertOne Failed, db:", dataBase,
			", Collection:", collection,
			", OneRecord:", oneRecord,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) Upsert(dataBase, collection string, selector interface{}, updater interface{}) (err error) {
	upsert := func(c *mgo.Collection) error {
		_, err := c.Upsert(selector, updater)
		return err
	}
	err = this.WithCollection(dataBase, collection, upsert)
	if err != nil {
		log.Println("mongodb Upsert, Upsert Failed, db:", dataBase,
			", Collection:", collection,
			", Selector:", selector,
			", Updater:", updater,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) BulkInsert(dataBase, collection string, allRecords []interface{}) (err error) {
	if 0 == len(allRecords) {
		log.Println("bulkInsert: 0 == len(allRecords)! db:", dataBase, "collection:", collection)
		return
	}
	bulkInsert := func(c *mgo.Collection) error {
		bulk := c.Bulk()
		bulk.Unordered()
		bulk.Insert(allRecords...)
		_, err := bulk.Run()
		return err
	}
	err = this.WithCollection(dataBase, collection, bulkInsert)
	if err != nil {
		log.Println("mongodb BulkInsert, BulkInsert Failed, db:", dataBase,
			", Collection:", collection,
			", ErrInfo:", err)
	}
	return
}

func (this *Mongodb) RemoveOne(dataBase, collection string, selector interface{}) (err error) {
	remove := func(c *mgo.Collection) error {
		return c.Remove(selector)
	}
	err = this.WithCollection(dataBase, collection, remove)
	if err != nil {
		log.Println("mongodb RemoveOne, RemoveOne Failed, db:", dataBase,
			", Collection:", collection,
			", Selector:", selector,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) RemoveAll(dataBase, collection string, selector interface{}) (err error) {
	removeAll := func(c *mgo.Collection) (err error) {
		removeAllInfo, err := c.RemoveAll(selector)
		if nil != err {
			log.Println("mongodb RemoveAll, RemoveAll Failed, info:", *removeAllInfo)
		}
		return
	}
	err = this.WithCollection(dataBase, collection, removeAll)
	if err != nil {
		log.Println("mongodb RemoveAll, RemoveAll Failed, db:", dataBase,
			", Collection:", collection,
			", Selector:", selector,
			", ErrInfo:", err)
		return
	}
	return
}


func (this *Mongodb) InsertOneFile(dataBase, fileName string, fileData []byte) (fileSize int, err error) {
	create := func(gridFS *mgo.GridFS) error {
		file, err := gridFS.Create(fileName)
		if nil != err {
			return err
		}
		fileSize, err = file.Write(fileData)
		if nil != err {
			return err
		}
		return file.Close()
	}
	err = this.WithGridFS(dataBase, create)
	if nil != err {
		log.Println("mongodb GetOneFile, GetOneFile Failed, db:", dataBase,
			", FileName:", fileName,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) GetOneFileInfo(dataBase string, queryCondition, selector bson.M) (result bson.M, err error) {
	query := func(gridFS *mgo.GridFS) error {
		return gridFS.Find(queryCondition).Select(selector).One(&result)
	}
	err = this.WithGridFS(dataBase, query)
	if nil != err {
		log.Println("mongodb GetOneFile, GetOneFile Failed, db:", dataBase,
			", QueryCondition:", queryCondition,
			", Selector:", selector,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) GetMultiFileInfo(dataBase string, queryCondition, selector bson.M) (result []bson.M, err error) {
	query := func(gridFS *mgo.GridFS) error {
		return gridFS.Find(queryCondition).Select(selector).All(&result)
	}
	err = this.WithGridFS(dataBase, query)
	if nil != err {
		log.Println("mongodb GetOneFile, GetOneFile Failed, db:", dataBase,
			", QueryCondition:", queryCondition,
			", Selector:", selector,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) OpenFileByName(dataBase, fileName string, fileSize int) (result []byte, err error) {
	openFile := func(gridFS *mgo.GridFS) error {
		file, err := gridFS.Open(fileName)
		if nil != err {
			return err
		}
		result = make([]byte, fileSize)
		_, err = file.Read(result)
		if nil != err {
			return err
		}
		return file.Close()
	}
	err = this.WithGridFS(dataBase, openFile)
	if nil != err {
		log.Println("mongodb OpenFile, OpenFile Failed, db:", dataBase,
			", FileName:", fileName,
			", FileSize:", fileSize,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) RemoveFileByName(dataBase, fileName string) (err error) {
	removeFile := func(gridFS *mgo.GridFS) error {
		return gridFS.Remove(fileName)
	}
	err = this.WithGridFS(dataBase, removeFile)
	if nil != err {
		log.Println("mongodb RemoveFile, RemoveFile Failed, db:", dataBase,
			", FileName:", fileName,
			", ErrInfo:", err)
		return
	}
	return
}

func (this *Mongodb) RemoveFileById(dataBase string, id interface{}) (err error) {
	removeFile := func(gridFS *mgo.GridFS) error {
		return gridFS.RemoveId(id)
	}
	err = this.WithGridFS(dataBase, removeFile)
	if nil != err {
		log.Println("mongodb RemoveFile, RemoveFile Failed, db:", dataBase,
			", Id:", id,
			", ErrInfo:", err)
		return
	}
	return
}