Add a Component runner with support for dependencies

Author: Pior Bastida

component.go

@@ -0,0 +1,74 @@
+package runnable
+
+import (
+	"context"
+	"time"
+)
+
+type C interface {
+	Add(runner Runnable)
+	AddService(service Runnable)
+}
+
+type component struct {
+	processes *group
+	services  *group
+}
+
+func Component(processes ...Runnable) *component {
+	return &component{
+		processes: &group{timeout: 30 * time.Second, runnables: processes},
+		services:  &group{timeout: 30 * time.Second},
+	}
+}
+
+// WithShutdownTimeout changes the default timeout (30s).
+func (c *component) WithShutdownTimeout(process, service time.Duration) *component {
+	c.processes.timeout = process
+	c.services.timeout = service
+	return c
+}
+
+// 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...)
+}
+
+// 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...)
+}
+
+func (c *component) Run(ctx context.Context) error {
+	ctxValues := ContextValues(ctx)
+
+	// Starting
+
+	log.Printf("group: starting services")
+	c.services.start(ctxValues)
+
+	log.Printf("group: starting processes")
+	c.processes.start(ctxValues)
+
+	// Waiting for shutdown
+
+	select {
+	case <-c.processes.errors:
+		log.Printf("group: a runner stopped")
+	case <-c.services.errors:
+		log.Printf("group: a service stopped")
+	case <-ctx.Done():
+		log.Printf("group: context cancelled")
+	}
+
+	// shutdown
+
+	log.Printf("group: shutting down processes")
+	c.processes.stop()
+
+	log.Printf("group: shutting down services")
+	c.services.stop()
+
+	log.Printf("group: shutdown complete")
+	return ctx.Err()
+}

examples/component/jobqueue.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+type StupidJobQueue struct {
+	queue    chan string
+	executed int64
+}
+
+func NewStupidJobQueue() *StupidJobQueue {
+	return &StupidJobQueue{queue: make(chan string, 100)}
+}
+
+func (s *StupidJobQueue) Enqueue(url string) {
+	s.queue <- url
+}
+
+func (s *StupidJobQueue) Waiting() int {
+	return len(s.queue)
+}
+
+func (s *StupidJobQueue) Executed() int64 {
+	return atomic.LoadInt64(&s.executed)
+}
+
+func (s *StupidJobQueue) Run(ctx context.Context) error {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		for url := range s.queue {
+			fmt.Printf("Performing job: %s\n", url)
+			atomic.AddInt64(&s.executed, 1)
+			time.Sleep(time.Second)
+		}
+		wg.Done()
+	}()
+
+	<-ctx.Done()
+	fmt.Printf("Draining job queue\n")
+	close(s.queue)
+	wg.Wait()
+
+	return ctx.Err()
+}

examples/component/main.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/pior/runnable"
+)
+
+func monitor(jobs *StupidJobQueue) runnable.RunnableFunc {
+	return func(ctx context.Context) error {
+		fmt.Printf("Task executed: %d\t\ttasks waiting: %d\n", jobs.Executed(), jobs.Waiting())
+		return nil
+	}
+}
+
+func main() {
+	jobs := NewStupidJobQueue()
+
+	// curl http://localhost:8000/?count=3
+	server := &http.Server{
+		Addr: "localhost:8000",
+		Handler: http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+			r.ParseForm()
+			count := r.Form.Get("count")
+			if count == "" {
+				count = "1"
+			}
+			countInt, _ := strconv.Atoi(count)
+
+			for i := 0; i < countInt; i++ {
+				jobs.Enqueue(r.URL.Path)
+				fmt.Fprintln(rw, "Job enqueued!")
+			}
+		}),
+	}
+	serverRunner := runnable.HTTPServer(server)
+
+	monitor := runnable.Every(runnable.Func(monitor(jobs)), 2*time.Second)
+
+	c := runnable.Component().WithShutdownTimeout(5*time.Second, 3*time.Second)
+	c.Add(serverRunner)
+	c.AddService(jobs)
+	c.Add(monitor)
+
+	fmt.Println("Enqueue jobs with: curl http://localhost:8000/?count=3")
+
+	runnable.Run(c)
+}

group.go

@@ -0,0 +1,66 @@
+package runnable
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"time"
+)
+
+type group struct {
+	runnables []Runnable
+	running   uint32
+	errors    chan error
+	cancel    func()
+	timeout   time.Duration
+}
+
+func (g *group) start(ctx context.Context) {
+	newCtx, cancel := context.WithCancel(ctx)
+
+	g.cancel = cancel
+	g.errors = make(chan error)
+
+	for _, r := range g.runnables {
+		atomic.AddUint32(&g.running, 1)
+		go g.run(newCtx, r)
+	}
+}
+
+func (g *group) stop() {
+	g.cancel()
+
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	after := time.After(g.timeout)
+
+	for {
+		select {
+		case <-ticker.C:
+			log.Printf("group: %d still running", g.running)
+		case <-g.errors:
+			if g.running == 0 {
+				return
+			}
+		case <-after:
+			log.Printf("group: waited %s, %d still running", g.timeout, g.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
+}