WIP

Author: Pior Bastida

component.go

@@ -15,6 +15,9 @@ type component struct {
 	services  *group
 }
 
+// Component manages two groups of Runnable, processes and services.
+// Services are stopped after processes.
+// The intended purposes is to manage the simplest form of runnable dependencies.
 func Component(processes ...Runnable) *component {
 	return &component{
 		processes: &group{timeout: 30 * time.Second, runnables: processes},
@@ -31,31 +34,31 @@ func (c *component) WithShutdownTimeout(process, service time.Duration) *compone
 
 // Add registers runnables as process. Processes will be shutdown before services.
 func (c *component) Add(runner ...Runnable) {
-	c.processes.runnables = append(c.processes.runnables, runner...)
+	c.processes.add(runner...)
 }
 
 // Add registers runnables as services. Services will be shutdown after processes.
 func (c *component) AddService(service ...Runnable) {
-	c.services.runnables = append(c.services.runnables, service...)
+	c.services.add(service...)
 }
 
 func (c *component) Run(ctx context.Context) error {
-	ctxValues := ContextValues(ctx)
+	ctxNoCancel := context.WithoutCancel(ctx)
 
 	// Starting
 
 	Log(c, "starting services")
-	c.services.start(ctxValues)
+	c.services.Start(ctxNoCancel)
 
 	Log(c, "starting processes")
-	c.processes.start(ctxValues)
+	c.processes.Start(ctxNoCancel)
 
 	// Waiting for shutdown
 
 	select {
-	case <-c.processes.errors:
-		Log(c, "a runner stopped")
-	case <-c.services.errors:
+	case <-c.processes.WaitForShutdown():
+		Log(c, "a process stopped")
+	case <-c.services.WaitForShutdown():
 		Log(c, "a service stopped")
 	case <-ctx.Done():
 		Log(c, "context cancelled")
@@ -64,10 +67,10 @@ func (c *component) Run(ctx context.Context) error {
 	// shutdown
 
 	Log(c, "shutting down processes")
-	c.processes.stop()
+	c.processes.Stop()
 
 	Log(c, "shutting down services")
-	c.services.stop()
+	c.services.Stop()
 
 	Log(c, "shutdown complete")
 	return ctx.Err()

component_test.go

@@ -1,23 +1,48 @@
 package runnable
 
-func ExampleComponent() {
+import (
+	"context"
+	stdlog "log"
+	"os"
+)
+
+func ExampleComponent_cancelled() {
 	ctx, cancel := initializeForExample()
 	defer cancel()
 
 	c := Component()
 	c.Add(newDummyRunnable())
 
-	c.Run(ctx)
+	_ = c.Run(ctx)
 
 	// Output:
 	// component: starting services
 	// component: starting processes
-	// group: dummyRunnable started
 	// dummyRunnable: started
 	// component: context cancelled
 	// component: shutting down processes
 	// dummyRunnable: stopped
-	// group: dummyRunnable stopped
+	// component: shutting down services
+	// component: shutdown complete
+}
+
+func ExampleComponent_failing() {
+	ctx := context.Background()
+
+	SetLogger(stdlog.New(os.Stdout, "", 0))
+
+	c := Component()
+	c.Add(newDyingRunnable())
+
+	_ = c.Run(ctx)
+
+	// Output:
+	// component: starting services
+	// component: starting processes
+	// dyingRunnable: started
+	// dyingRunnable: stopped with error: dying
+	// component: a process stopped
+	// component: shutting down processes
 	// component: shutting down services
 	// component: shutdown complete
 }

group.go

@@ -2,69 +2,76 @@ package runnable
 
 import (
 	"context"
-	"errors"
-	"sync/atomic"
 	"time"
 )
 
 type group struct {
 	runnables []Runnable
-	running   uint32
-	errors    chan error
-	cancel    func()
 	timeout   time.Duration
+
+	contextCancel func()
+	started       []*StartedRunnable
+	shutdownCh    chan struct{}
+	stoppedCh     chan *StartedRunnable
 }
 
-func (g *group) start(ctx context.Context) {
-	newCtx, cancel := context.WithCancel(ctx)
+func (g *group) add(rs ...Runnable) {
+	g.runnables = append(g.runnables, rs...)
+}
 
-	g.cancel = cancel
-	g.errors = make(chan error)
+// func (g *group) Run(ctx context.Context) error {
+// 	g.Start(ctx)
 
-	for _, r := range g.runnables {
-		atomic.AddUint32(&g.running, 1)
-		go g.run(newCtx, r)
-	}
+// 	select {
+// 	case <-ctx.Done():
+// 	case <-g.shutdownCh:
+// 	}
+
+// 	g.waitCh <- struct{}{}
+
+// 	errs := append([]error{ctx.Err()}, g.Stop()...)
+// 	return errors.Join(errs...)
+// }
+
+func (g *group) WaitForShutdown() chan struct{} {
+	return g.shutdownCh
 }
 
-func (g *group) stop() {
-	g.cancel()
+func (g *group) Start(ctx context.Context) {
+	ctx, g.contextCancel = context.WithCancel(ctx)
+
+	g.started = make([]*StartedRunnable, 0, len(g.runnables))
+	g.shutdownCh = make(chan struct{}, len(g.runnables))
+	g.stoppedCh = make(chan *StartedRunnable, len(g.runnables))
 
-	if g.running == 0 {
-		return
+	for _, r := range g.runnables {
+		g.started = append(g.started, StartRunnable(ctx, r, g.shutdownCh, g.stoppedCh))
 	}
 
-	ticker := time.NewTicker(5 * time.Second)
-	defer ticker.Stop()
+}
+
+func (g *group) Stop() (groupErrors []error) {
+	g.contextCancel() // stop all other runnings
 
-	after := time.After(g.timeout)
+	stopTimeout := time.After(g.timeout)
+	logTicker := time.NewTicker(3 * time.Second)
+	defer logTicker.Stop()
 
-	for {
+	running := len(g.started)
+	for running > 0 {
 		select {
-		case <-ticker.C:
-			Log(g, "%d still running", g.running)
-		case <-g.errors:
-			if g.running == 0 {
-				return
+		case stopped := <-g.stoppedCh:
+			running--
+			if stopped.err != nil {
+				groupErrors = append(groupErrors, stopped.err)
 			}
-		case <-after:
-			Log(g, "waited %s, %d still running", g.timeout, g.running)
+		case <-logTicker.C:
+			Log(g, "%d still running", running)
+		case <-stopTimeout:
+			Log(g, "waited %s, %d still running", g.timeout, running)
 			return
 		}
 	}
-}
-
-func (g *group) run(ctx context.Context, runnable Runnable) {
-	name := findName(runnable)
-
-	log.Printf("group: %s started", name)
-	err := Recover(runnable).Run(ctx)
-	if err == nil || errors.Is(err, context.Canceled) {
-		log.Printf("group: %s stopped", name)
-	} else {
-		log.Printf("group: %s stopped with error: %+v", name, err)
-	}
 
-	atomic.AddUint32(&g.running, ^uint32(0))
-	g.errors <- err
+	return
 }

start.go

@@ -0,0 +1,59 @@
+package runnable
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+type StartedRunnable struct {
+	baseWrapper
+
+	runnable   Runnable
+	shutdownCh chan struct{}
+	stoppedCh  chan *StartedRunnable
+
+	mu      sync.Mutex
+	running bool
+	err     error
+}
+
+func StartRunnable(ctx context.Context, r Runnable, shutdownCh chan struct{}, stoppedCh chan *StartedRunnable) *StartedRunnable {
+	s := &StartedRunnable{
+		baseWrapper: baseWrapper{"", r},
+		runnable:    r,
+		shutdownCh:  shutdownCh,
+		stoppedCh:   stoppedCh,
+	}
+
+	go func() {
+		Log(r, "started")
+		err := Recover(r).Run(ctx)
+		if err == nil || errors.Is(err, context.Canceled) {
+			Log(r, "stopped")
+		} else {
+			Log(r, "stopped with error: %+v", err)
+		}
+
+		s.mu.Lock()
+		defer s.mu.Unlock()
+
+		s.running = false
+		s.err = err
+
+		if s.shutdownCh != nil {
+			s.shutdownCh <- struct{}{}
+		}
+		if s.stoppedCh != nil {
+			s.stoppedCh <- s
+		}
+	}()
+
+	return s
+}
+
+func (s *StartedRunnable) State() (running bool, err error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.running, s.err
+}

testing_test.go

@@ -20,9 +20,9 @@ func newDummyRunnable() *dummyRunnable {
 type dummyRunnable struct{}
 
 func (r *dummyRunnable) Run(ctx context.Context) error {
-	Log(r, "started")
+	// Log(r, "DEBUG:started")
 	<-ctx.Done()
-	Log(r, "stopped")
+	// Log(r, "DEBUG:stopped")
 	return ctx.Err()
 }