WIP

Author: Pior Bastida

component.go

@@ -5,6 +5,8 @@ import (
 	"time"
 )
 
+var DefaultComponentShutdownTimeout = 30 * time.Second
+
 type C interface {
 	Add(runner Runnable)
 	AddService(service Runnable)
@@ -15,59 +17,71 @@ 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},
-		services:  &group{timeout: 30 * time.Second},
+		processes: &group{timeout: DefaultComponentShutdownTimeout, runnables: processes},
+		services:  &group{timeout: DefaultComponentShutdownTimeout},
 	}
 }
 
-// WithShutdownTimeout changes the default timeout (30s).
-func (c *component) WithShutdownTimeout(process, service time.Duration) *component {
-	c.processes.timeout = process
-	c.services.timeout = service
+// WithProcessShutdownTimeout changes the shutdown timeout of the process runnables.
+// See also DefaultComponentShutdownTimeout.
+func (c *component) WithProcessShutdownTimeout(timeout time.Duration) *component {
+	c.processes.timeout = timeout
+	return c
+}
+
+// WithServiceShutdownTimeout changes the shutdown timeout of the service runnables.
+// See also DefaultComponentShutdownTimeout.
+func (c *component) WithServiceShutdownTimeout(timeout time.Duration) *component {
+	c.services.timeout = timeout
 	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...)
+// AddProcess registers runnables as process. Processes will be shutdown before services.
+func (c *component) AddProcess(runner ...Runnable) *component {
+	c.processes.Add(runner...)
+	return c
 }
 
 // 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) AddService(service ...Runnable) *component {
+	c.services.Add(service...)
+	return c
 }
 
 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:
-		Log(c, "a service stopped")
+	case r := <-c.processes.StoppedRunnables():
+		Log(c, "process stopped: %s", findName(r))
+	case r := <-c.services.StoppedRunnables():
+		Log(c, "service stopped: %s", findName(r))
 	case <-ctx.Done():
 		Log(c, "context cancelled")
 	}
 
 	// 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.AddProcess(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.AddProcess(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
 }

examples/component/jobqueue.go

@@ -4,47 +4,44 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"strconv"
 	"time"
 
 	"github.com/pior/runnable"
+	"github.com/pior/runnable/examples/jobqueue"
 )
 
-func monitor(jobs *StupidJobQueue) runnable.RunnableFunc {
+// Run it with:
+// go run ./examples/component/main.go
+
+// Test it with:
+// curl "http://localhost:8000/?count=3"
+
+func jobMonitor(jobs *jobqueue.JobQueue) 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()
+	jobs := jobqueue.New()
 
-	// 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!")
-			}
+			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)
+	c := runnable.Component().
+		WithProcessShutdownTimeout(5 * time.Second).
+		WithServiceShutdownTimeout(5 * time.Second).
+		AddProcess(runnable.HTTPServer(server)).
+		AddService(jobs).
+		AddProcess(
+			runnable.Every(runnable.Func(jobMonitor(jobs)), 2*time.Second),
+		)
 
 	fmt.Println("Enqueue jobs with: curl http://localhost:8000/?count=3")
 

examples/component/main.go

@@ -7,15 +7,16 @@ import (
 	"time"
 
 	"github.com/pior/runnable"
+	"github.com/pior/runnable/examples/jobqueue"
 )
 
 func main() {
-	jobs := NewStupidJobQueue()
+	jobs := jobqueue.New()
 
 	server := &http.Server{
 		Addr: "localhost:8000",
 		Handler: http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
-			jobs.Perform(r.URL.Path)
+			jobs.Enqueue(r.URL.Path)
 			fmt.Fprintln(rw, "Job enqueued!")
 		}),
 	}