examples/http: Add HTTP example to demonstrate RunShutdownable

actor.go

@@ -88,3 +88,22 @@ type Resettable interface {
 	// invalid and any process which depends on it should be shut down.
 	Reset(ctx context.Context) error
 }
+
+// RunShutdownable is any object that needs a context in order to shut down.
+// Actors which implement this interface cannot implement the Runnable
+// interface. This can be used to run an http.Server as an actor.
+type RunShutdownable interface {
+	// Run is similar to [Runnable.Run], but takes no context and blocks
+	// until Shutdown has run and exited. It may also return when something
+	// goes wrong and it cannot continue, in which case it must return a
+	// non-nil error explaining why. Shutdown does not need to be called in
+	// the latter case.
+	Run() error
+	// Shutdown shuts down the actor. It must unblock Run in all cases even
+	// on failure, context expiration, etc. Shutdown must return when or
+	// before the context expires, and must return ctx.Err if there is no
+	// other error to be returned. If Shutdown returns any error, the object
+	// must be treated as invalid and any other process which depends on it
+	// should be shut down.
+	Shutdown(ctx context.Context) error
+}

actorset.go

@@ -16,6 +16,7 @@ type actorSets struct {
 	configurable    actorSet[Configurable]
 	initializable   actorSet[Initializable]
 	runnable        actorSet[Runnable]
+	runShutdownable actorSet[RunShutdownable]
 	trimmable       actorSet[Trimmable]
 }
 
@@ -52,6 +53,7 @@ func (sets *actorSets) All() iter.Seq[actorSetIface] {
 		yield(&sets.configurable)
 		yield(&sets.initializable)
 		yield(&sets.runnable)
+		yield(&sets.runShutdownable)
 		yield(&sets.trimmable)
 	}
 }
@@ -66,7 +68,9 @@ func (this *actorSets) add(ctx context.Context, actor Actor) {
 		done:  done,
 		order: this.nextOrder,
 	}
-	if _, ok := actor.(Runnable); ok {
+	_, isRunnable        := actor.(Runnable)
+	_, isRunShutdownable := actor.(RunShutdownable)
+	if isRunnable || isRunShutdownable {
 		info.stopped = make(chan struct { })
 	}
 	this.inf[actor] = info

environment.go

@@ -122,7 +122,9 @@ func (this *environment) Add(ctx context.Context, actors ...Actor) error {
 		}
 	}
 	for _, actor := range actors {
-		if actor, ok := actor.(Runnable); ok {
+		_, isRunnable        := actor.(Runnable)
+		_, isRunShutdownable := actor.(RunShutdownable)
+		if isRunnable || isRunShutdownable {
 			this.start(actor)
 		}
 	}
@@ -213,7 +215,7 @@ func (this *environment) info(actor Actor) actorInfo {
 // start increments the wait group by one and starts the given actor in the
 // background, restarting it if it fails. this function will exit immediately.
 // see the documentation for run for details.
-func (this *environment) start(actor Runnable) {
+func (this *environment) start(actor Actor) {
 	this.group.Add(1)
 	go this.run(actor)
 }
@@ -223,14 +225,12 @@ func (this *environment) start(actor Runnable) {
 // environment once this function exits, and the environment's wait group
 // counter will be decremented. note that this function will never increment the
 // wait group counter, so start should usually be used instead.
-func (this *environment) run(actor Runnable) {
+func (this *environment) run(actor Actor) {
 	// clean up when done
 	defer this.group.Done()
 
 	// logging
-	acto, ok := actor.(Actor)
-	if !ok { return }
-	typ := acto.Type()
+	typ := actor.Type()
 	if this.Verb() { log.Printf("(i) [%s] running", typ) }
 	var stopErr error
 	var exited  bool
@@ -247,10 +247,26 @@ func (this *environment) run(actor Runnable) {
 	}()
 
 	// contains context information
-	info := this.info(acto)
+	info := this.info(actor)
 	ctx := info.ctx
 	defer close(info.stopped)
+	
+	switch actor := actor.(type) {
+	case Runnable:
+		stopErr, exited = this.runRunnable(ctx, actor)
+	case RunShutdownable:
+		stopErr, exited = this.runRunnable(ctx, &runShutdownableShim {
+			shutdownTimeout: defaul(this.timing.shutdownTimeout.Load(), defaultShutdownTimeout),
+			underlying: actor,
+		})
+	default:
+		panic("actor was neither Runnable or RunShutdownable")
+	}
+}
 
+// runRunnable runs an actor implementing [Runnable]. this should only be called
+// from within [environment.run].
+func (this *environment) runRunnable(ctx context.Context, actor Runnable) (stopErr error, exited bool) {
 	// timing
 	restartThreshold         := defaul(this.timing.restartThreshold.Load(), defaultRestartThreshold)
 	restartInitialInterval   := defaul(this.timing.restartInitialInterval.Load(), defaultRestartInitialInterval)
@@ -259,11 +275,14 @@ func (this *environment) run(actor Runnable) {
 	resetTimeout             := defaul(this.timing.resetTimeout.Load(), defaultResetTimeout)
 	restartInterval          := restartInitialInterval
 
-	// main loop
+	acto, ok := actor.(Actor)
+	if !ok { return }
+	typ := acto.Type()
+	
 	for {
 		// run actor
 		lastStart := time.Now()
-		err := panicWrap(ctx, actor.Run)
+		err := panicWrapCtx(ctx, actor.Run)
 
 		// detect context cancellation
 		if ctxErr := ctx.Err(); ctxErr != nil {
@@ -387,3 +406,24 @@ func (this *environment) applyConfig() error {
 	if err != nil { return err }
 	return nil
 }
+
+type runShutdownableShim struct {
+	underlying RunShutdownable
+	shutdownTimeout time.Duration
+}
+
+func (this *runShutdownableShim) Type() string {
+	return this.underlying.(Actor).Type()
+}
+
+func (this *runShutdownableShim) Run(ctx context.Context) error {
+	ctx, done := context.WithCancel(ctx)
+	defer done()
+	go func() {
+		<- ctx.Done()
+		shutdownCtx, done := context.WithTimeout(context.Background(), this.shutdownTimeout)
+		defer done()
+		this.underlying.Shutdown(shutdownCtx)
+	}()
+	return this.underlying.Run()
+}

examples/http/main.go

@@ -0,0 +1,90 @@
+// Example http demonstrates the usage of [camfish.RunShutdowner] to run an http
+// server.
+package main
+
+import "fmt"
+import "log"
+import "iter"
+import "errors"
+import "context"
+import "net/http"
+import "git.tebibyte.media/sashakoshka/camfish"
+import "git.tebibyte.media/sashakoshka/go-util/sync"
+
+func main() {
+	camfish.Run("http",
+		"Example http demonstrates the usage of " +
+		"camfish.RunShutdowner to run an http server",
+		new(httpServer),
+		new(database))
+}
+
+// httpServer serves data over http.
+type httpServer struct {
+	server *http.Server
+	database *database
+}
+var _ camfish.RunShutdownable = new(httpServer)
+var _ camfish.Initializable = new(httpServer)
+func (this *httpServer) Type() string { return "http-server" }
+
+func (this *httpServer) Init(ctx context.Context) error {
+	this.server = &http.Server {
+		Addr: "localhost:8080",
+		Handler: this,
+	}
+	if actor, ok := camfish.Find("database").(*database); ok {
+		this.database = actor
+	} else {
+		return errors.New("could not locate database")
+	}
+	return ctx.Err()
+}
+
+func (this *httpServer) Run() error {
+	log.Printf("[http] listening on %s", this.server.Addr)
+	err := this.server.ListenAndServe()
+	if errors.Is(err, http.ErrServerClosed) { return nil }
+	return err
+}
+
+func (this *httpServer) Shutdown(ctx context.Context) error {
+	return this.server.Shutdown(ctx)
+}
+
+func (this *httpServer) ServeHTTP(res http.ResponseWriter, req *http.Request) {
+	fmt.Fprintf(res, "<!DOCTYPE html><html><head><title>inventory</title></head><body>")
+	fmt.Fprintf(res, "<table><tr><th>Item</th><th>Count</th></tr>")
+	for item, count := range this.database.Inventory() {
+		fmt.Fprintf(res, "<tr><td>%s</td><td>%d</td></tr>", item, count)
+	}
+	fmt.Fprintf(res, "</table>")
+	fmt.Fprintf(res, "</body></html>")
+}
+
+// database provides data that can be served.
+type database struct {
+	inventory usync.RWMonitor[map[string] int]
+}
+func (this *database) Type() string { return "database" }
+
+func (this *database) Init(ctx context.Context) error {
+	this.inventory.Set(map[string] int {
+		"screws": 34,
+		"blood": 90,
+		"paperclips": 5230,
+		"wood": 3,
+		"grains of rice": 238409,
+	})
+	return ctx.Err()
+}
+
+func (this *database) Inventory() iter.Seq2[string, int] {
+	return func(yield func(string, int) bool) {
+		inventory, done := this.inventory.RBorrow()
+		defer done()
+		for item, amount := range inventory {
+			yield(item, amount)
+		}
+	}
+}

mock/config_test.go

@@ -8,16 +8,16 @@ func TestConfig(test *testing.T) {
 		"multiple": []string { "item0", "item1" },
 		"empty": []string { },
 	}
-	if correct, got := config.Get("single"), "aslkdjasd"; correct != got {
+	if correct, got := "aslkdjasd", config.Get("single"); correct != got {
 		test.Fatal("not equal:", got)
 	}
-	if correct, got := config.Get("multiple"), "item0"; correct != got {
+	if correct, got := "item0", config.Get("multiple"); correct != got {
 		test.Fatal("not equal:", got)
 	}
-	if correct, got := config.Get("empty"), ""; correct != got {
+	if correct, got := "", config.Get("empty"); correct != got {
 		test.Fatal("not equal:", got)
 	}
-	if correct, got := config.Get("non-existent"), ""; correct != got {
+	if correct, got := "", config.Get("non-existent"); correct != got {
 		test.Fatal("not equal:", got)
 	}
 	for index, value := range config.GetAll("single") {

phases.go

@@ -199,7 +199,7 @@ func (this *environment) phase60Initialization() bool {
 		initializable = actors.initializable.all()
 	}()
 	if err := this.initializeActors(this.ctx, initializable...); err != nil {
-		log.Println(".// (60) failed to initialize:", err)
+		log.Println("XXX (60) failed to initialize:", err)
 		return false
 	}
 	if this.Verb() { log.Println(".// (60) initialized") }
@@ -215,7 +215,10 @@ func (this *environment) phase70Running() bool {
 		actors, done := this.actors.RBorrow()
 		defer done()
 		for _, actor := range actors.runnable.all() {
-			this.start(actor)
+			this.start(actor.(Actor))
+		}
+		for _, actor := range actors.runShutdownable.all() {
+			this.start(actor.(Actor))
 		}
 		
 	}()

run.go

@@ -9,7 +9,7 @@ var env environment
 // when all running actors have stopped. Error and log messages will be printed.
 // The correct way to use this function is to have it be the only thing in main:
 // 
-//   func main () {
+//   func main() {
 //           camfish.Run("name", "what it does", new(SomeActor), new(AnotherActor))
 //   }
 //
@@ -50,21 +50,22 @@ var env environment
 //       is configurable, but by default it is 8 minutes. The vast majority of
 //       actors should initialize in under 100 milliseconds.
 //   
-//   70. Running: Actors which implement [Runnable] are run, each in their own
-//       goroutine. The environment is able to restart actors which have failed,
-//       which entails resetting the actor if it implements [Resettable], and
-//       running the actor again within the same goroutine. If an actor does not
-//       run for a meaningful amount of time after resetting/initialization
-//      before failing, it is considered erratic and further attempts to restart
-//      it will be spaced by a limited, constantly increasing time interval. The
-//      timing is configurable, but by default the threshold for a meaningful
-//      amount of runtime is 16 seconds, the initial delay interval is 8
-//      seconds, the interval increase per attempt is 8 seconds, and the maximum
-//      interval is one hour. Additionally, programs which implement [Trimmable]
-//      will be trimmed regularly whenever they are running. The trimming
-//      interval is also configurable, but by default it is once every minute.
-//      When an actor which implements [Resettable] is reset, it is given a
-//      configurable timeout, which is 8 minutes by default.
+//   70. Running: Actors which implement [Runnable] or [RunShutdownable] are
+//       run, each in their own goroutine. The environment is able to restart
+//       actors which have failed, which entails resetting the actor if it
+//       implements [Resettable], and running the actor again within the same
+//       goroutine. If an actor does not run for a meaningful amount of time
+//       after resetting/initialization before failing, it is considered erratic
+//       and further attempts to restart it will be spaced by a limited,
+//       constantly increasing time interval. The timing is configurable, but by
+//       default the threshold for a meaningful amount of runtime is 16 seconds,
+//       the initial delay interval is 8 seconds, the interval increase per
+//       attempt is 8 seconds, and the maximum interval is one hour.
+//       Additionally, programs which implement [Trimmable] will be trimmed
+//       regularly whenever they are running. The trimming interval is also
+//       configurable, but by default it is once every minute. When an actor
+//       which implements [Resettable] is reset, it is given a configurable
+//       timeout, which is 8 minutes by default.
 //   
 //   80. Shutdown: This can be triggered by all actors being removed from the
 //       environment, a catastrophic error, [Done] being called, or the program

util.go

@@ -19,7 +19,22 @@ func defaul[T comparable](value, def T) T {
 	return value
 }
 
-func panicWrap(ctx context.Context, f func (context.Context) error) (err error) {
+func panicWrap(f func() error) (err error) {
+	defer func () {
+		if pan := recover(); pan != nil {
+			if panErr, ok := pan.(error); ok {
+				err = panErr
+			} else {
+				err = errors.New(fmt.Sprint(pan))
+			}
+		}
+	} ()
+	
+	err = f()
+	return
+}
+
+func panicWrapCtx(ctx context.Context, f func(context.Context) error) (err error) {
 	defer func () {
 		if pan := recover(); pan != nil {
 			if panErr, ok := pan.(error); ok {

util_test.go

@@ -17,6 +17,24 @@ func TestDefaul(test *testing.T) {
 }
 
 func TestPanicWrap(test *testing.T) {
+	err := panicWrap(func (ctx context.Context) error {
+		return errors.New("test case 0")
+	})
+	test.Log(err)
+	if err.Error() != "test case 0" { test.Fatal("not equal") }
+	err = panicWrap(func (ctx context.Context) error {
+		panic(errors.New("test case 1"))
+	})
+	test.Log(err)
+	if err.Error() != "test case 1" { test.Fatal("not equal") }
+	err = panicWrap( func (ctx context.Context) error {
+		return nil
+	})
+	test.Log(err)
+	if err != nil { test.Fatal("not equal") }
+}
+
+func TestPanicWrapCtx(test *testing.T) {
 	err := panicWrap(context.Background(), func (ctx context.Context) error {
 		return errors.New("test case 0")
 	})