Merge pull request #27 from nashtsai/processors

implementation of before/after processors

base_test.go

@@ -1706,6 +1706,211 @@ func testPrefixTableName(engine *Engine, t *testing.T) {
 	}
 }
 
+type ProcessorsStruct struct {
+	Id int64
+	
+	B4InsertFlag int
+	AfterInsertedFlag int 
+	B4UpdateFlag int
+	AfterUpdatedFlag int
+	B4DeleteFlag int `xorm:"-"`
+	AfterDeletedFlag int `xorm:"-"`
+
+	B4InsertViaExt int
+	AfterInsertedViaExt int 
+	B4UpdateViaExt int
+	AfterUpdatedViaExt int
+	B4DeleteViaExt int `xorm:"-"`
+	AfterDeletedViaExt int `xorm:"-"`
+}
+
+func (p *ProcessorsStruct) BeforeInsert() {
+	p.B4InsertFlag = 1
+}
+
+func (p *ProcessorsStruct) BeforeUpdate() {
+	p.B4UpdateFlag = 1
+}
+
+func (p *ProcessorsStruct) BeforeDelete() {
+	p.B4DeleteFlag = 1
+}
+
+func (p *ProcessorsStruct) AfterInsert() {
+	p.AfterInsertedFlag = 1
+}
+
+func (p *ProcessorsStruct) AfterUpdate() {
+	p.AfterUpdatedFlag = 1
+}
+
+func (p *ProcessorsStruct) AfterDelete() {
+	p.AfterDeletedFlag = 1
+}
+
+func testProcessors(engine *Engine, t *testing.T) {
+	tempEngine, err := NewEngine(engine.DriverName, engine.DataSourceName)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	}
+	
+	tempEngine.ShowSQL = true
+	err = tempEngine.DropTables(&ProcessorsStruct{})
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	}
+
+	err = tempEngine.CreateTables(&ProcessorsStruct{})
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	}
+
+	p := &ProcessorsStruct{}
+
+	b4InsertFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.B4InsertViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+
+	afterInsertFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.AfterInsertedViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+		
+	_, err = tempEngine.Before(b4InsertFunc).After(afterInsertFunc).Insert(p)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if p.B4InsertFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+		if p.AfterInsertedFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+		if p.B4InsertViaExt == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+		if p.AfterInsertedViaExt == 0 { t.Error(errors.New("AfterInsertedViaExt not set")) }
+	}
+		
+	p2 := &ProcessorsStruct{}
+	_, err = tempEngine.Id(p.Id).Get(p2)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if p2.B4InsertFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+		if p2.AfterInsertedFlag != 0 { t.Error(errors.New("AfterInsertedFlag is set")) }
+		if p2.B4InsertViaExt == 0 { t.Error(errors.New("B4InsertViaExt not set")) }
+		if p2.AfterInsertedViaExt != 0 { t.Error(errors.New("AfterInsertedViaExt is set")) }
+	}
+
+	b4UpdateFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.B4UpdateViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+
+	afterUpdateFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.AfterUpdatedViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+	
+	p = p2 // reset
+
+	_, err = tempEngine.Before(b4UpdateFunc).After(afterUpdateFunc).Update(p)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if p.B4UpdateFlag == 0 { t.Error(errors.New("B4UpdateFlag not set")) }
+		if p.AfterUpdatedFlag == 0 { t.Error(errors.New("AfterUpdatedFlag not set")) }
+		if p.B4UpdateViaExt == 0 { t.Error(errors.New("B4UpdateViaExt not set")) }
+		if p.AfterUpdatedViaExt == 0 { t.Error(errors.New("AfterUpdatedViaExt not set")) }
+	}
+	
+	p2 = &ProcessorsStruct{}	
+	_, err = tempEngine.Id(p.Id).Get(p2)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if p2.B4UpdateFlag == 0 { t.Error(errors.New("B4UpdateFlag not set")) }
+		if p2.AfterUpdatedFlag != 0 { t.Error(errors.New("AfterUpdatedFlag is set: " + string(p.AfterUpdatedFlag))) }
+		if p2.B4UpdateViaExt == 0 { t.Error(errors.New("B4UpdateViaExt not set")) }
+		if p2.AfterUpdatedViaExt != 0 { t.Error(errors.New("AfterUpdatedViaExt is set: " + string(p.AfterUpdatedViaExt))) }
+	}
+
+	b4DeleteFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.B4DeleteViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+
+	afterDeleteFunc := func(bean interface{}) {
+		if v, ok := (bean).(*ProcessorsStruct); ok {
+			v.AfterDeletedViaExt = 1
+		} else {
+			t.Error(errors.New("cast to ProcessorsStruct failed, how can this be!?"))
+		}
+	}
+	
+	p = p2 // reset
+	_, err = tempEngine.Before(b4DeleteFunc).After(afterDeleteFunc).Delete(p)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if p.B4DeleteFlag == 0 { t.Error(errors.New("B4DeleteFlag not set")) }
+		if p.AfterDeletedFlag == 0 { t.Error(errors.New("AfterDeletedFlag not set")) }
+		if p.B4DeleteViaExt == 0 { t.Error(errors.New("B4DeleteViaExt not set")) }
+		if p.AfterDeletedViaExt == 0 { t.Error(errors.New("AfterDeletedViaExt not set")) }
+	}
+
+	// test insert multi
+	pslice := make([]*ProcessorsStruct, 0);
+	pslice = append(pslice, &ProcessorsStruct{})
+	pslice = append(pslice, &ProcessorsStruct{})
+	cnt, err := tempEngine.Before(b4InsertFunc).After(afterInsertFunc).Insert(&pslice)
+	if err != nil {
+		t.Error(err)
+		panic(err)
+	} else {
+		if cnt != 2 { t.Error(errors.New("incorrect insert count")) }
+		for _, elem := range pslice {
+			if elem.B4InsertFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+			if elem.AfterInsertedFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+			if elem.B4InsertViaExt == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+			if elem.AfterInsertedViaExt == 0 { t.Error(errors.New("AfterInsertedViaExt not set")) }
+		}
+	}
+	
+	for _, elem := range pslice {
+		p = &ProcessorsStruct{}
+		_, err = tempEngine.Id(elem.Id).Get(p)
+		if err != nil {
+			t.Error(err)
+			panic(err)
+		} else {
+			if p2.B4InsertFlag == 0 { t.Error(errors.New("B4InsertFlag not set")) }
+			if p2.AfterInsertedFlag != 0 { t.Error(errors.New("AfterInsertedFlag is set")) }
+			if p2.B4InsertViaExt == 0 { t.Error(errors.New("B4InsertViaExt not set")) }
+			if p2.AfterInsertedViaExt != 0 { t.Error(errors.New("AfterInsertedViaExt is set")) }
+		}
+	}
+}
+
 func testAll(engine *Engine, t *testing.T) {
 	fmt.Println("-------------- directCreateTable --------------")
 	directCreateTable(engine, t)
@@ -1800,6 +2005,8 @@ func testAll2(engine *Engine, t *testing.T) {
 	testTime(engine, t)
 	fmt.Println("-------------- testPrefixTableName --------------")
 	testPrefixTableName(engine, t)
+	fmt.Println("-------------- processors --------------")
+	testProcessors(engine, t)
 	fmt.Println("-------------- transaction --------------")
 	transaction(engine, t)
 }

engine.go

@@ -272,6 +272,20 @@ func (engine *Engine) Id(id int64) *Session {
 	return session.Id(id)
 }
 
+// Apply before Processor, affected bean is passed to closure arg
+func (engine *Engine) Before(closures func(interface{})) *Session {
+	session := engine.NewSession()
+	session.IsAutoClose = true
+	return session.Before(closures)
+} 
+
+// Apply after insert Processor, affected bean is passed to closure arg
+func (engine *Engine) After(closures func(interface{})) *Session {
+	session := engine.NewSession()
+	session.IsAutoClose = true
+	return session.After(closures)
+} 
+
 // set charset when create table, only support mysql now
 func (engine *Engine) Charset(charset string) *Session {
 	session := engine.NewSession()

processors.go

@@ -0,0 +1,39 @@
+package xorm
+
+// Executed before an object is initially persisted to the database
+type BeforeInsertProcessor interface {
+	BeforeInsert()
+}
+
+// Executed before an object is updated
+type BeforeUpdateProcessor interface {
+	BeforeUpdate()
+}
+
+// Executed before an object is deleted
+type BeforeDeleteProcessor interface {
+	BeforeDelete()
+}
+
+// !nashtsai! TODO enable BeforeValidateProcessor when xorm start to support validations
+//// Executed before an object is validated
+//type BeforeValidateProcessor interface {
+//	BeforeValidate()
+//}
+// --
+
+// Executed after an object is persisted to the database
+type AfterInsertProcessor interface {
+	AfterInsert()
+}
+
+// Executed after an object has been updated
+type AfterUpdateProcessor interface {
+	AfterUpdate()
+}
+
+// Executed after an object has been deleted
+type AfterDeleteProcessor interface {
+	AfterDelete()
+}
+

session.go

@@ -22,6 +22,15 @@ type Session struct {
 	IsCommitedOrRollbacked bool
 	TransType              string
 	IsAutoClose            bool
+	
+    // !nashtsai! storing these beans due to yet committed tx
+	afterInsertBeans []interface{}
+	afterUpdateBeans []interface{}
+	afterDeleteBeans []interface{}
+	// --
+	
+	beforeClosures []func(interface{})
+	afterClosures []func(interface{})
 }
 
 // Method Init reset the session as the init status.
@@ -31,6 +40,13 @@ func (session *Session) Init() {
 	session.IsAutoCommit = true
 	session.IsCommitedOrRollbacked = false
 	session.IsAutoClose = false
+	
+	// !nashtsai! is lazy init better?
+	session.afterInsertBeans = make([]interface{}, 0)
+	session.afterUpdateBeans = make([]interface{}, 0)
+	session.afterDeleteBeans = make([]interface{}, 0)
+	session.beforeClosures = make([]func(interface{}), 0)
+	session.afterClosures = make([]func(interface{}), 0)
 }
 
 // Method Close release the connection from pool
@@ -76,6 +92,22 @@ func (session *Session) Id(id int64) *Session {
 	return session
 }
 
+// Apply before Processor, affected bean is passed to closure arg
+func (session *Session) Before(closures func(interface{})) *Session {
+	if closures != nil {
+		session.beforeClosures = append(session.beforeClosures, closures)
+	}
+	return session
+} 
+
+// Apply after Processor, affected bean is passed to closure arg
+func (session *Session) After(closures func(interface{})) *Session {
+	if closures != nil {
+		session.afterClosures = append(session.afterClosures, closures)
+	}
+	return session
+}
+
 // Method Table can input a string or pointer to struct for special a table to operate.
 func (session *Session) Table(tableNameOrBean interface{}) *Session {
 	session.Statement.Table(tableNameOrBean)
@@ -251,7 +283,52 @@ func (session *Session) Commit() error {
 	if !session.IsAutoCommit && !session.IsCommitedOrRollbacked {
 		session.Engine.LogSQL("COMMIT")
 		session.IsCommitedOrRollbacked = true
-		return session.Tx.Commit()
+		var err error
+		if err = session.Tx.Commit(); err == nil { 
+			// handle processors after tx committed
+			for _, elem := range session.afterInsertBeans {
+				for _, closure := range session.afterClosures {
+					closure(elem)
+				}
+				
+				if processor, ok := interface{}(elem).(AfterInsertProcessor); ok {
+					processor.AfterInsert()
+				}
+			}
+			for _, elem := range session.afterUpdateBeans {
+				for _, closure := range session.afterClosures {
+					closure(elem)
+				}
+				if processor, ok := interface{}(elem).(AfterUpdateProcessor); ok {
+					processor.AfterUpdate()
+				}
+			}
+			for _, elem := range session.afterDeleteBeans {
+				for _, closure := range session.afterClosures {
+					closure(elem)
+				}			 
+				if processor, ok := interface{}(elem).(AfterDeleteProcessor); ok {
+					processor.AfterDelete()
+				}
+			}
+			cleanUpFunc := func(slices *[]interface{}) {
+				if len(*slices) > 0 {
+					*slices = make([]interface{}, 0)
+				}
+			}
+			cleanUpProcessorsFunc := func(slices *[]func(interface{})) {
+				if len(*slices) > 0 {
+					*slices = make([]func(interface{}), 0)
+				}
+			}
+			cleanUpFunc(&session.afterInsertBeans)
+			cleanUpFunc(&session.afterUpdateBeans)
+			cleanUpFunc(&session.afterDeleteBeans)
+
+			// !nash! should session based processors get cleanup?
+			cleanUpProcessorsFunc(&session.afterClosures)
+		} 
+		return err
 	}
 	return nil
 }
@@ -1333,6 +1410,17 @@ func (session *Session) innerInsertMulti(rowsSlicePtr interface{}) (int64, error
 	for i := 0; i < size; i++ {
 		elemValue := sliceValue.Index(i).Interface()
 		colPlaces := make([]string, 0)
+		
+		// handle BeforeInsertProcessor
+		for _, closure := range session.beforeClosures {
+			closure(elemValue)
+		}
+
+		if processor, ok := interface{}(elemValue).(BeforeInsertProcessor); ok {
+			processor.BeforeInsert()
+		}
+		// --
+		
 
 		if i == 0 {
 			for _, col := range table.Columns {
@@ -1409,6 +1497,28 @@ func (session *Session) innerInsertMulti(rowsSlicePtr interface{}) (int64, error
 	if table.Cacher != nil && session.Statement.UseCache {
 		session.cacheInsert(session.Statement.TableName())
 	}
+	
+	hasAfterClosures := len(session.afterClosures) > 0 
+	for i := 0; i < size; i++ {
+		elemValue := sliceValue.Index(i).Interface()
+		// handle AfterInsertProcessor
+		if session.IsAutoCommit {
+			for _, closure := range session.afterClosures {
+				closure(elemValue)
+			}
+			if processor, ok := interface{}(elemValue).(AfterInsertProcessor); ok {
+				processor.AfterInsert()
+			}
+		} else {
+			if hasAfterClosures {
+				session.afterInsertBeans = append(session.afterInsertBeans, elemValue)				
+			} else {
+				if _, ok := interface{}(elemValue).(AfterInsertProcessor); ok {
+					session.afterInsertBeans = append(session.afterInsertBeans, elemValue)	
+				}
+			}
+		}
+	}
 
 	return res.RowsAffected()
 }
@@ -1682,6 +1792,18 @@ func (session *Session) innerInsert(bean interface{}) (int64, error) {
 	table := session.Engine.autoMap(bean)
 	session.Statement.RefTable = table
 
+	// handle BeforeInsertProcessor
+	for _, closure := range session.beforeClosures {
+		closure(bean)
+	}
+	if processor, ok := interface{}(bean).(BeforeInsertProcessor); ok {
+		session.Engine.LogDebug(session.Statement.TableName(), " has before insert processor")
+		processor.BeforeInsert()
+	} else {
+		session.Engine.LogDebug(session.Statement.TableName(), " has no before insert processor")
+	}
+	// --		
+
 	colNames, args, err := table.genCols(session, bean, false, false)
 	if err != nil {
 		return 0, err
@@ -1699,12 +1821,36 @@ func (session *Session) innerInsert(bean interface{}) (int64, error) {
 		session.Engine.QuoteStr(),
 		colPlaces)
 
+
+	handleAfterInsertProcessorFunc := func(bean interface{}) {
+
+		if session.IsAutoCommit {
+			for _, closure := range session.afterClosures {
+				closure(bean)
+			}
+ 			if processor, ok := interface{}(bean).(AfterInsertProcessor); ok {
+		    	session.Engine.LogDebug(session.Statement.TableName(), " has after insert processor")
+				processor.AfterInsert()
+			}
+		} else {
+			if len(session.afterClosures) > 0 {
+				session.afterInsertBeans = append(session.afterInsertBeans, bean)				
+			} else {
+ 				if _, ok := interface{}(bean).(AfterInsertProcessor); ok {
+					session.afterInsertBeans = append(session.afterInsertBeans, bean)				
+				}
+			}
+		}
+	}	
+
 	// for postgres, many of them didn't implement lastInsertId, so we should
 	// implemented it ourself.
 	if session.Engine.DriverName != POSTGRES || table.PrimaryKey == "" {
 		res, err := session.exec(sql, args...)
 		if err != nil {
 			return 0, err
+		} else {
+			handleAfterInsertProcessorFunc(bean)
 		}
 
 		if table.Cacher != nil && session.Statement.UseCache {
@@ -1748,6 +1894,8 @@ func (session *Session) innerInsert(bean interface{}) (int64, error) {
 		res, err := session.query(sql, args...)
 		if err != nil {
 			return 0, err
+		} else {
+			handleAfterInsertProcessorFunc(bean)
 		}
 
 		if table.Cacher != nil && session.Statement.UseCache {
@@ -1979,6 +2127,17 @@ func (session *Session) Update(bean interface{}, condiBean ...interface{}) (int6
 	var args []interface{}
 	var table *Table
 
+
+	// handle before update processors
+	for _, closure := range session.beforeClosures {
+		closure(bean)
+	}
+
+	if processor, ok := interface{}(bean).(BeforeUpdateProcessor); ok {
+		processor.BeforeUpdate()
+	}
+	// --
+
 	if t.Kind() == reflect.Struct {
 		table = session.Engine.autoMap(bean)
 		session.Statement.RefTable = table
@@ -2077,6 +2236,26 @@ func (session *Session) Update(bean interface{}, condiBean ...interface{}) (int6
 		table.Cacher.ClearBeans(session.Statement.TableName())
 	}
 
+	// handle after update processors
+	if session.IsAutoCommit {
+		for _, closure := range session.afterClosures {
+			closure(bean)
+		}
+		if processor, ok := interface{}(bean).(AfterUpdateProcessor); ok {
+			session.Engine.LogDebug(session.Statement.TableName(), " has after update processor")
+			processor.AfterUpdate()
+		}
+	} else {
+		if len(session.afterClosures) > 0 {
+			session.afterUpdateBeans = append(session.afterUpdateBeans, bean)                               
+		} else {
+			if _, ok := interface{}(bean).(AfterUpdateProcessor); ok {
+				session.afterUpdateBeans = append(session.afterUpdateBeans, bean)                               
+			}
+		}
+ 	}
+	// --
+
 	return res.RowsAffected()
 }
 
@@ -2142,6 +2321,17 @@ func (session *Session) Delete(bean interface{}) (int64, error) {
 		defer session.Close()
 	}
 
+
+	// handle before delete processors
+	for _, closure := range session.beforeClosures {
+		closure(bean)
+	}
+
+	if processor, ok := interface{}(bean).(BeforeDeleteProcessor); ok {
+		processor.BeforeDelete()
+	}
+	// --
+
 	table := session.Engine.autoMap(bean)
 	session.Statement.RefTable = table
 	colNames, args := buildConditions(session.Engine, table, bean, true,
@@ -2174,5 +2364,25 @@ func (session *Session) Delete(bean interface{}) (int64, error) {
 		return 0, err
 	}
 
+	// handle after delete processors
+	if session.IsAutoCommit {
+		for _, closure := range session.afterClosures {
+			closure(bean)
+		}
+		if processor, ok := interface{}(bean).(AfterDeleteProcessor); ok {
+			session.Engine.LogDebug(session.Statement.TableName(), " has after update processor")
+			processor.AfterDelete()
+		}
+	} else {
+		if len(session.afterClosures) > 0 {
+			session.afterDeleteBeans = append(session.afterDeleteBeans, bean)                               
+		} else {
+			if _, ok := interface{}(bean).(AfterDeleteProcessor); ok {
+				session.afterDeleteBeans = append(session.afterDeleteBeans, bean)                               
+			}
+		}
+ 	}
+	// --
+
 	return res.RowsAffected()
 }