Resolve "Internal Changes representation is not working as expected"

Description

Related Issue

Motivation and Context

How Has This Been Tested?

Screenshots (if appropriate):

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist:

• My code follows the code style of this project.
• My change requires a change to the documentation.
• I have updated the documentation accordingly.
• I have read the CONTRIBUTING document.
• I have added tests to cover my changes.
• All new and existing tests passed.

Closes #152

nucleus/change.go

 package nucleus
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"sync"
@@ -43,10 +44,12 @@ func NewChange(device uuid.UUID, currentState ygot.GoStruct, change ygot.GoStruc
 		intendedState: change,
 		callback:      callback,
 	}
-	stateIn, stateOut, errChan := stateManager(c, changeTimeout)
+	stateManagerCtx, stateManagerCancel := context.WithCancel(context.Background())
+	stateIn, stateOut, errChan := stateManager(c, stateManagerCtx, changeTimeout)
 	c.stateIn = stateIn
 	c.stateOut = stateOut
 	c.errChan = errChan
+	c.stateManagerCancel = stateManagerCancel
 	return c
 }
 
@@ -55,17 +58,18 @@ func NewChange(device uuid.UUID, currentState ygot.GoStruct, change ygot.GoStruc
 // state. It keeps track if the state is committed and confirmed. A callback
 // exists to acess the proper transport for the changed OND
 type Change struct {
-	cuid          uuid.UUID
-	duid          uuid.UUID
-	state         ppb.Change_State
-	timestamp     time.Time
-	previousState ygot.GoStruct
-	intendedState ygot.GoStruct
-	callback      func(ygot.GoStruct, ygot.GoStruct) error
-	stateMu       sync.RWMutex
-	errChan       <-chan error
-	stateIn       chan<- ppb.Change_State
-	stateOut      <-chan ppb.Change_State
+	cuid               uuid.UUID
+	duid               uuid.UUID
+	state              ppb.Change_State
+	timestamp          time.Time
+	previousState      ygot.GoStruct
+	intendedState      ygot.GoStruct
+	callback           func(ygot.GoStruct, ygot.GoStruct) error
+	stateMu            sync.RWMutex
+	errChan            <-chan error
+	stateIn            chan<- ppb.Change_State
+	stateOut           <-chan ppb.Change_State
+	stateManagerCancel context.CancelFunc
 }
 
 // ID returns the Change's UUID
@@ -77,6 +81,7 @@ func (c *Change) ID() uuid.UUID {
 // and starts the timeout-timer for the Change. If the timer expires
 // the change is rolled back.
 func (c *Change) Commit() error {
+	//TODO: check if already commited
 	c.stateIn <- ppb.Change_COMMITTED
 	select {
 	case err := <-c.errChan:
@@ -88,6 +93,7 @@ func (c *Change) Commit() error {
 
 // Confirm confirms a committed Change and stops the rollback timer.
 func (c *Change) Confirm() error {
+	//TODO: check if already confirmed
 	c.stateIn <- ppb.Change_CONFIRMED
 	select {
 	case err := <-c.errChan:
@@ -110,7 +116,7 @@ func (c *Change) State() ppb.Change_State {
 	return state
 }
 
-func stateManager(ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <-chan ppb.Change_State, <-chan error) {
+func stateManager(ch *Change, ctx context.Context, timeout time.Duration) (chan<- ppb.Change_State, <-chan ppb.Change_State, <-chan error) {
 	stateIn := make(chan ppb.Change_State)
 	stateOut := make(chan ppb.Change_State)
 	// A Goroutine, which is created while a new Change is initialized acts as
@@ -122,11 +128,13 @@ func stateManager(ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <
 	ticker.Stop()
 
 	go func() {
+	running:
 		for {
 			//TODO: priority select? should ticker have priority?
 			select {
 			case <-ticker.C:
-				if ch.state == ppb.Change_CONFIRMED {
+				state := ch.State()
+				if state == ppb.Change_CONFIRMED {
 					continue
 				}
 				err := ch.callback(ch.intendedState, ch.previousState)
@@ -148,8 +156,9 @@ func stateManager(ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <
 			case s := <-stateIn:
 				switch s {
 				case ppb.Change_COMMITTED:
-					if ch.state == ppb.Change_COMMITTED || ch.state == ppb.Change_CONFIRMED {
-						errChan <- fmt.Errorf("change %v already %s", ch.cuid, ch.state.String())
+					state := ch.State()
+					if state == ppb.Change_COMMITTED || state == ppb.Change_CONFIRMED {
+						errChan <- fmt.Errorf("change %v already %s", ch.cuid, state.String())
 						continue
 					}
 					// reset ticker to enable activate the change timeout
@@ -165,22 +174,27 @@ func stateManager(ch *Change, timeout time.Duration) (chan<- ppb.Change_State, <
 					ch.stateMu.Lock()
 					ch.state = ppb.Change_COMMITTED
 					ch.stateMu.Unlock()
-					stateOut <- ch.state
+					stateOut <- state
 				case ppb.Change_CONFIRMED:
-					if ch.state != ppb.Change_COMMITTED {
+					state := ch.State()
+					if state != ppb.Change_COMMITTED {
 						errChan <- fmt.Errorf("cannot confirm uncommitted change %v", ch.cuid)
 						continue
 					}
 					// The change has been confirmed and the timer is stopped,
 					// since a rollback is not necessary anymore.
-					ticker.Stop()
 					ch.stateMu.Lock()
 					ch.state = ppb.Change_CONFIRMED
 					ch.stateMu.Unlock()
-					stateOut <- ch.state
+					stateOut <- state
+					ch.stateManagerCancel()
 				}
+			case <-ctx.Done():
+				ticker.Stop()
+				break running
 			}
 		}
+		log.Info("statemanager routine done for: ", ch.cuid)
 	}()
 	return stateIn, stateOut, errChan
 }