[#91] Added Depenedncy Mapping of components to manager (#92)

Author: nandagopalan

lifecycle/component.go

@@ -27,6 +27,8 @@ var ErrCompAlreadyStopped = errors.New("component already stopped")
 
 var ErrInvalidComponentState = errors.New("invalid component state")
 
+var ErrCyclicDependency = errors.New("cyclic dependency")
+
 // Component is the interface that wraps the basic Start and Stop methods.
 type Component interface {
 	// Id is the unique identifier for the component.
@@ -44,6 +46,8 @@ type Component interface {
 
 // ComponentManager is the interface that manages multiple components.
 type ComponentManager interface {
+	// AddDependency will add a dependency between the two components.
+	AddDependency(id, dependsOn string) error
 	// GetState will return the current state of the LifeCycle for the component with the given id.
 	GetState(id string) ComponentState
 	//List will return a list of all the Components.

lifecycle/simple_component.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 	"syscall"
 
+	"oss.nandlabs.io/golly/collections"
 	"oss.nandlabs.io/golly/errutils"
 )
 
@@ -117,6 +118,31 @@ type SimpleComponentManager struct {
 	componentIds []string
 	cMutex       *sync.RWMutex
 	waitChan     chan struct{}
+	dependencies map[string]collections.List[string]
+}
+
+// AddDependency will add a dependency between the two components.
+func (scm *SimpleComponentManager) AddDependency(id, dependsOn string) (err error) {
+	scm.cMutex.Lock()
+	defer scm.cMutex.Unlock()
+	if _, exists := scm.components[id]; !exists {
+		return ErrCompNotFound
+	}
+	if _, exists := scm.components[dependsOn]; !exists {
+		return ErrCompNotFound
+	}
+
+	//detect cyclic dependencies
+	if v, ok := scm.dependencies[dependsOn]; ok && v.Contains(id) {
+		return ErrCyclicDependency
+	}
+
+	if _, exists := scm.dependencies[id]; !exists {
+		scm.dependencies[id] = collections.NewArrayList[string]()
+	}
+	scm.dependencies[id].Add(dependsOn)
+	logger.InfoF("Added dependency %s depends on %s:", id, dependsOn)
+	return
 }
 
 // GetState will return the current state of the LifeCycle for the component with the given id.
@@ -171,21 +197,54 @@ func (scm *SimpleComponentManager) Start(id string) (err error) {
 	scm.cMutex.Lock()
 	defer scm.cMutex.Unlock()
 	component, exists := scm.components[id]
-	if exists {
-		if component.State() != Running {
-			var err error = nil
-			go func(c Component, scm *SimpleComponentManager) {
-				err = component.Start()
-				if err != nil {
-					logger.ErrorF("Error starting component: %v", err)
-				}
-			}(component, scm)
-			return err
+	if !exists {
+		return ErrCompNotFound
+	}
+	if component.State() == Running {
+		return
+	}
+	// Start the dependencies first
+	if v, ok := scm.dependencies[id]; ok {
+		logger.DebugF("Component %s has dependencies. Starting dependencies", id)
+		dependecyWait := sync.WaitGroup{}
+		var multiError *errutils.MultiError = errutils.NewMultiErr(nil)
+		for ite := v.Iterator(); ite.HasNext(); {
+			dependentComp := scm.components[ite.Next()]
+			if dependentComp.State() != Running {
+				dependecyWait.Add(1)
+				go func(c Component, scm *SimpleComponentManager) {
+					logger.DebugF("Starting dependent component %s", dependentComp.Id())
+					err = dependentComp.Start()
+					if err != nil {
+						multiError.Add(err)
+						logger.ErrorF("Error starting component: %v", err)
+					} else {
+						logger.DebugF("Started dependent component %s", dependentComp.Id())
+					}
+					dependecyWait.Done()
+				}(dependentComp, scm)
+			} else {
+				logger.DebugF("Dependent component %s already running", dependentComp.Id())
+			}
+		}
+		dependecyWait.Wait()
+
+		if multiError.HasErrors() {
+			return multiError
 		} else {
-			return ErrCompAlreadyStarted
+			logger.Info("All dependencies started")
 		}
+
+	}
+	logger.DebugF("Starting component %s", id)
+	err = component.Start()
+	if err != nil {
+		logger.ErrorF("Error starting component: %v", err)
+	} else {
+		logger.DebugF("Started component %s", id)
 	}
-	return ErrCompNotFound
+
+	return
 }
 
 // StartAll will start all the Components. Returns the number of components started
@@ -211,32 +270,79 @@ func (scm *SimpleComponentManager) StartAndWait() {
 
 }
 
+// Stop will stop the LifeCycle for the component with the given id. It returns if the component was stopped.
+func (scm *SimpleComponentManager) Stop(id string) (err error) {
+
+	component, exists := scm.components[id]
+	if !exists {
+		return ErrCompNotFound
+	}
+	if component.State() == Stopped {
+		return
+	}
+	// check if the component has dependencies
+	if v, ok := scm.dependencies[id]; ok {
+		logger.DebugF("Component %s has dependencies", id)
+		dependecyWait := sync.WaitGroup{}
+		var multiError *errutils.MultiError = errutils.NewMultiErr(nil)
+		for ite := v.Iterator(); ite.HasNext(); {
+			dependentComp := scm.components[ite.Next()]
+			logger.DebugF("Checking dependent component %s", dependentComp.Id())
+			if dependentComp.State() != Stopped {
+				dependecyWait.Add(1)
+				go func(c Component, scm *SimpleComponentManager) {
+					logger.InfoF("Stopping dependent component %s", c.Id())
+					err = dependentComp.Stop()
+					if err != nil {
+						multiError.Add(err)
+						logger.ErrorF("Error stopping component: %v", err)
+					} else {
+						logger.DebugF("Stopped dependent component %s", c.Id())
+					}
+					dependecyWait.Done()
+
+				}(dependentComp, scm)
+			} else {
+				logger.InfoF("Dependent component %s already stopped", dependentComp.Id())
+			}
+		}
+		dependecyWait.Wait()
+		if multiError.HasErrors() {
+			return multiError
+		} else {
+			logger.DebugF("All dependencies stopped proceeding to stop component %s", id)
+		}
+	}
+	scm.cMutex.Lock()
+	defer scm.cMutex.Unlock()
+	if component.State() == Running {
+		logger.Debug("Stopping component ", id)
+		err := component.Stop()
+
+		if err != nil {
+			logger.ErrorF("Error stopping component: %v", err)
+		} else {
+			logger.InfoF("Stopped component %s", id)
+		}
+
+	}
+	return
+}
+
 // StopAll will stop all the Components.
 func (scm *SimpleComponentManager) StopAll() error {
 	logger.InfoF("Stopping all components")
 	err := errutils.NewMultiErr(nil)
-	scm.cMutex.Lock()
-	defer scm.cMutex.Unlock()
-	wg := &sync.WaitGroup{}
 	for i := len(scm.componentIds) - 1; i >= 0; i-- {
-		component := scm.components[scm.componentIds[i]]
-		if component.State() == Running {
-			wg.Add(1)
-			go func(c Component, wg *sync.WaitGroup) {
-				e := component.Stop()
-				if e != nil {
-					logger.ErrorF("Error stopping component: %v", err)
-					err.Add(e)
-				}
-
-				wg.Done()
-			}(component, wg)
+		e := scm.Stop(scm.componentIds[i])
+		if e != nil {
+			logger.ErrorF("Error stopping component: %v", err)
+			err.Add(e)
 		}
 	}
-	wg.Wait()
+	logger.Info("All components stopped")
 	select {
 	case <-scm.waitChan:
-		logger.Info("All components stopped")
 	default:
 		close(scm.waitChan)
 	}
@@ -247,28 +353,6 @@ func (scm *SimpleComponentManager) StopAll() error {
 	}
 }
 
-// Stop will stop the LifeCycle for the component with the given id. It returns if the component was stopped.
-func (scm *SimpleComponentManager) Stop(id string) error {
-	scm.cMutex.Lock()
-	defer scm.cMutex.Unlock()
-	component, exists := scm.components[id]
-	if exists {
-		if component.State() == Running {
-			err := component.Stop()
-			if err != nil {
-				logger.ErrorF("Error stopping component: %v", err)
-			}
-			return err
-		} else if component.State() == Stopped {
-			return ErrCompAlreadyStopped
-		} else {
-			return ErrInvalidComponentState
-		}
-
-	}
-	return ErrCompNotFound
-}
-
 // Unregister will unregister a Component.
 func (scm *SimpleComponentManager) Unregister(id string) {
 	scm.cMutex.Lock()
@@ -304,9 +388,10 @@ func (scm *SimpleComponentManager) Wait() {
 // NewSimpleComponentManager will return a new SimpleComponentManager.
 func NewSimpleComponentManager() ComponentManager {
 	manager := &SimpleComponentManager{
-		components: make(map[string]Component),
-		cMutex:     &sync.RWMutex{},
-		waitChan:   make(chan struct{}),
+		components:   make(map[string]Component),
+		cMutex:       &sync.RWMutex{},
+		waitChan:     make(chan struct{}),
+		dependencies: make(map[string]collections.List[string]),
 	}
 	sigs := make(chan os.Signal, 1)
 

lifecycle/simple_component_test.go

@@ -166,24 +166,89 @@ func TestSimpleComponentManager_StartAll(t *testing.T) {
 // It checks if the StartAndWait method waits for the components to finish.
 func TestSimpleComponentManager_StopAll(t *testing.T) {
 	manager := NewSimpleComponentManager()
-	component := &SimpleComponent{
-		CompId: "test",
+	componentA := &SimpleComponent{
+		CompId: "comp-a",
 		StopFunc: func() error {
+			logger.InfoF("Stopping component %s", "comp-a")
 			return nil
 		},
 		StartFunc: func() error {
+			logger.InfoF("Starting component %s", "comp-a")
 			return nil
 		},
 	}
-	manager.Register(component)
+	componentB := &SimpleComponent{
+		CompId: "comp-b",
+		StopFunc: func() error {
+			logger.InfoF("Stopping component %s", "comp-b")
+			return nil
+		},
+		StartFunc: func() error {
+			logger.InfoF("Starting component %s", "comp-b")
+			return nil
+		},
+	}
+	manager.Register(componentA)
+	manager.Register(componentB)
 	manager.StartAll()
 	time.Sleep(500 * time.Millisecond)
 
 	manager.StopAll()
 	time.Sleep(500 * time.Millisecond)
-	if component.State() != Stopped {
-		t.Errorf("StopAll() state = %v, want %v", component.CompState, Stopped)
+	if componentA.State() != Stopped {
+		t.Errorf("StopAll() state = %v, want %v", componentA.CompState, Stopped)
+	}
+
+	if componentB.State() != Stopped {
+		t.Errorf("StopAll() state = %v, want %v", componentB.CompState, Stopped)
+	}
+
+}
+
+// TestSimpleComponentManager_StartAndWait tests the StartAndWait method of the SimpleComponentManager struct.
+// It verifies the behavior of the StartAndWait method by checking if the components are started correctly.
+// The test case includes starting a component with a StartFunc that returns no error.
+// It checks if the StartAndWait method waits for the components to finish.
+func TestSimpleComponentManager_DependencyCheck(t *testing.T) {
+	manager := NewSimpleComponentManager()
+	componentA := &SimpleComponent{
+		CompId: "comp-a",
+		StopFunc: func() error {
+			logger.InfoF("Stopping component %s", "comp-a")
+			return nil
+		},
+		StartFunc: func() error {
+			logger.InfoF("Starting component %s", "comp-a")
+			return nil
+		},
+	}
+	componentB := &SimpleComponent{
+		CompId: "comp-b",
+		StopFunc: func() error {
+			logger.InfoF("Stopping component %s", "comp-b")
+			return nil
+		},
+		StartFunc: func() error {
+			logger.InfoF("Starting component %s", "comp-b")
+			return nil
+		},
 	}
+	manager.Register(componentA)
+	manager.Register(componentB)
+	manager.AddDependency("comp-a", "comp-b")
+	manager.StartAll()
+	time.Sleep(500 * time.Millisecond)
+
+	manager.StopAll()
+	time.Sleep(500 * time.Millisecond)
+	if componentA.State() != Stopped {
+		t.Errorf("StopAll() state = %v, want %v", componentA.CompState, Stopped)
+	}
+
+	if componentB.State() != Stopped {
+		t.Errorf("StopAll() state = %v, want %v", componentB.CompState, Stopped)
+	}
+
 }
 
 // TestSimpleComponentManager_Unregister tests the Unregister method of the SimpleComponentManager struct.