release: 1.7.1

README.md

@@ -936,7 +936,7 @@ same or get better, never worse.
 
  require (
 -	github.com/xgodev/boost v1.1.1
-+	github.com/xgodev/boost v1.7.0
++	github.com/xgodev/boost v1.7.1
  )
 ```
 
@@ -947,7 +947,7 @@ ok  github.com/acme/shop/...
 
 1 line changed in go.mod, **0 lines changed in your code**.
 
-Releases: 2024-07 → `v1.1.0` → `v1.1.1` → **`v1.7.0`**
+Releases: 2024-07 → `v1.1.0` → `v1.1.1` → **`v1.7.1`**
 ([all releases](https://github.com/xgodev/boost/releases)).
 
 ## There's a factory for that.

bootstrap/function/adapter/contrib/cloudevents/sdk-go/v2/core/http/cmd.go

@@ -6,15 +6,16 @@ import (
 	"github.com/xgodev/boost/bootstrap/function"
 )
 
-// New returns CmdFunc for cloudevents command.
+// New returns the CmdFunc of the cloudevents_http command: it serves the
+// events posted to it until the run context is done.
 func New[T any](opts []client.Option, plugins ...Plugin) function.CmdFunc[T] {
 	return func(fn function.Handler[T]) *co.Command {
 		return &co.Command{
 			Use:   "cloudevents_http",
 			Short: "cloudevents_http",
 			Long:  "",
 			RunE: func(cmd *co.Command, args []string) error {
-				return Run[T](fn, opts, plugins...)
+				return RunContext[T](cmd.Context(), fn, opts, plugins...)
 			},
 		}
 	}

bootstrap/function/adapter/contrib/cloudevents/sdk-go/v2/core/http/run.go

@@ -17,10 +17,19 @@ import (
 // Plugin defines a function to process plugin.
 type Plugin func(context.Context, []cehttp.Option) []cehttp.Option
 
-func Run[T any](fn function.Handler[T], opts []client.Option, plugins ...Plugin) (err error) {
-
-	ctx := context.Background()
+// Run serves fn with a context that is never cancelled, so it never returns
+// on its own.
+//
+// Deprecated: use RunContext, which stops when its context is done.
+func Run[T any](fn function.Handler[T], opts []client.Option, plugins ...Plugin) error {
+	return RunContext[T](context.Background(), fn, opts, plugins...)
+}
 
+// RunContext serves fn over HTTP on Port() and Path() until ctx is done:
+// then it takes no new event, shuts the server down once the events in hand
+// are answered, and returns. A handler runs with a context the stop does not
+// cancel, so the event it holds finishes.
+func RunContext[T any](ctx context.Context, fn function.Handler[T], opts []client.Option, plugins ...Plugin) error {
 	logger := log.FromContext(ctx)
 
 	httpOpts := []cehttp.Option{ce.WithPort(Port()), ce.WithPath(Path())}
@@ -32,6 +41,7 @@ func Run[T any](fn function.Handler[T], opts []client.Option, plugins ...Plugin)
 	p, err := ce.NewHTTP(httpOpts...)
 	if err != nil {
 		logger.Errorf("failed to create protocol: %s", err.Error())
+		return err
 	}
 
 	c, err := ce.NewClient(p, opts...)
@@ -42,7 +52,10 @@ func Run[T any](fn function.Handler[T], opts []client.Option, plugins ...Plugin)
 
 	logger.Infof("listening on :%d%s", Port(), Path())
 
-	return c.StartReceiver(ctx, Wrapper(fn))
+	handle := Wrapper(fn)
+	return c.StartReceiver(ctx, func(ctx context.Context, event ce.Event) ce.Result {
+		return handle(context.WithoutCancel(ctx), event)
+	})
 }
 
 func Wrapper[T any](fn function.Handler[T]) func(context.Context, ce.Event) ce.Result {

bootstrap/function/adapter/contrib/cloudevents/sdk-go/v2/core/http/run_test.go

@@ -1,12 +1,30 @@
 package http
 
 import (
+	"context"
+	"net"
 	"net/http"
+	"os"
+	"strconv"
+	"strings"
 	"testing"
+	"time"
 
+	ce "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/client"
+	cehttp "github.com/cloudevents/sdk-go/v2/protocol/http"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/xgodev/boost"
 	"github.com/xgodev/boost/model/errors"
 )
 
+// TestMain boots boost: the adapter reads its port and path from config.
+func TestMain(m *testing.M) {
+	boost.Start()
+	os.Exit(m.Run())
+}
+
 type customForbidden struct{ msg string }
 
 func (c *customForbidden) Error() string { return c.msg }
@@ -28,3 +46,146 @@ func TestErrorStatusCode_BuiltinNotFound(t *testing.T) {
 		t.Fatal("builtin NotFound not mapped to 404")
 	}
 }
+
+// withPort is a Plugin that makes the adapter listen on port.
+func withPort(port int) Plugin {
+	return func(_ context.Context, opts []cehttp.Option) []cehttp.Option {
+		return append(opts, cehttp.WithPort(port))
+	}
+}
+
+// freePort returns a TCP port nothing listens on.
+func freePort(t *testing.T) int {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	port := lis.Addr().(*net.TCPAddr).Port
+	require.NoError(t, lis.Close())
+	return port
+}
+
+// post sends one order.created CloudEvent to port until the adapter
+// listens, and returns the status of its reply.
+func post(port int) (int, error) {
+	client := &http.Client{Timeout: 10 * time.Second}
+	tick := time.NewTicker(50 * time.Millisecond)
+	defer tick.Stop()
+	deadline := time.After(10 * time.Second)
+	for {
+		req, err := http.NewRequest(http.MethodPost, "http://127.0.0.1:"+strconv.Itoa(port)+"/", strings.NewReader(`{"sku":"tshirt"}`))
+		if err != nil {
+			return 0, err
+		}
+		req.Header.Set("Ce-Id", "evt-1")
+		req.Header.Set("Ce-Source", "test")
+		req.Header.Set("Ce-Type", "order.created")
+		req.Header.Set("Ce-Specversion", "1.0")
+		req.Header.Set("Content-Type", "application/json")
+		res, err := client.Do(req)
+		if err == nil {
+			_ = res.Body.Close()
+			return res.StatusCode, nil
+		}
+		select {
+		case <-tick.C:
+		case <-deadline:
+			return 0, err
+		}
+	}
+}
+
+// TestRunContext_LetsTheHandlerFinishAndReturns cancels the run context
+// while the handler holds an event (#262): the handler's context is not
+// cancelled by the stop, the sender gets the reply and RunContext returns
+// nil, so the app stops at once instead of waiting the fx stop timeout.
+func TestRunContext_LetsTheHandlerFinishAndReturns(t *testing.T) {
+	port := freePort(t)
+	held := make(chan context.Context, 1)
+	release := make(chan struct{})
+	fn := func(ctx context.Context, _ ce.Event) ([]*ce.Event, error) {
+		held <- ctx
+		<-release
+		return nil, nil
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ran := make(chan error, 1)
+	go func() { ran <- RunContext[[]*ce.Event](ctx, fn, nil, withPort(port)) }()
+
+	type reply struct {
+		status int
+		err    error
+	}
+	replied := make(chan reply, 1)
+	go func() {
+		status, err := post(port)
+		replied <- reply{status, err}
+	}()
+
+	var handlerCtx context.Context
+	select {
+	case handlerCtx = <-held:
+	case err := <-ran:
+		t.Fatalf("RunContext returned (%v) before the event", err)
+	case r := <-replied:
+		t.Fatalf("the event was answered (%d, %v) without the handler", r.status, r.err)
+	}
+
+	cancel()
+	assert.NoError(t, handlerCtx.Err(), "the stop cancelled the event in hand")
+	close(release)
+
+	r := <-replied
+	require.NoError(t, r.err)
+	assert.Equal(t, http.StatusOK, r.status, "the sender gets the reply of the event in hand")
+	select {
+	case err := <-ran:
+		assert.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("RunContext did not return after the run context was cancelled")
+	}
+}
+
+// TestRunContext_FailsOnABadProtocolOption returns the error of an option
+// the protocol rejects, instead of logging it and failing on a nil protocol.
+func TestRunContext_FailsOnABadProtocolOption(t *testing.T) {
+	fn := func(context.Context, ce.Event) ([]*ce.Event, error) { return nil, nil }
+
+	err := RunContext[[]*ce.Event](context.Background(), fn, nil, withPort(-1))
+	assert.ErrorContains(t, err, "invalid port: -1")
+
+	err = Run[[]*ce.Event](fn, nil, withPort(-1))
+	assert.ErrorContains(t, err, "invalid port: -1", "Run is RunContext with a context never cancelled")
+}
+
+// TestRunContext_FailsOnABadClientOption returns the error of a client
+// option that fails.
+func TestRunContext_FailsOnABadClientOption(t *testing.T) {
+	fn := func(context.Context, ce.Event) ([]*ce.Event, error) { return nil, nil }
+	bad := func(any) error { return errors.New("bad client option") }
+
+	err := RunContext[[]*ce.Event](context.Background(), fn, []client.Option{bad}, withPort(freePort(t)))
+	assert.ErrorContains(t, err, "bad client option")
+}
+
+// TestNew_RunsOnTheCommandContext executes the cloudevents_http command with
+// a context already done (#262): it runs on the command's context, so it
+// returns instead of serving forever.
+func TestNew_RunsOnTheCommandContext(t *testing.T) {
+	fn := func(context.Context, ce.Event) ([]*ce.Event, error) { return nil, nil }
+	cmd := New[[]*ce.Event](nil, withPort(freePort(t)))(fn)
+	assert.Equal(t, "cloudevents_http", cmd.Use)
+	cmd.SetArgs([]string{})
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	ran := make(chan error, 1)
+	go func() { ran <- cmd.ExecuteContext(ctx) }()
+	select {
+	case err := <-ran:
+		assert.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("the command did not return on a context already done")
+	}
+}

builder/README.md

@@ -307,6 +307,12 @@ package rest
 
 - A `@Provide` `[]Plugin` still builds a server (see [Servers](#servers)):
   that server gets no default set, and a `@Server` for it too fails.
+- A door whose server lists `otel` (or whose `@Function` runs the
+  `opentelemetry` middleware) also includes
+  [`otelfx.Module()`](../fx/modules/factory/contrib/go.opentelemetry.io/otel/v1):
+  it starts the OpenTelemetry providers on boot and shuts them down on stop,
+  so the spans and metrics reach `boost.factory.otel.endpoint` instead of a
+  noop provider. The module runs once per app, however many doors include it.
 
 ### App names
 
@@ -377,7 +383,8 @@ Factory clients too: `@Inject(index=0, name=quote, plugins=[otel])` on a
 godror or pgx) parameter makes
 inject build that client with those plugins and its own config under
 `boost.factory.resty.clients.quote` (see
-[`inject`](../inject#factory-clients)), so the app writes no factory module.
+[`inject`](../inject#factory-clients)), so the app writes no factory module;
+a client with the `otel` plugin starts the OpenTelemetry providers too.
 
 builder owns the main of every app: a `cmd/<app>/main.go` without that header
 fails the generation instead of being overwritten. Other files under `cmd/`

builder/annotations.go

@@ -74,6 +74,9 @@ type functionMiddleware struct {
 	// Needs are the unnamed values the middleware is built with, provided
 	// by the app.
 	Needs []typeRef
+	// OTel is set for a middleware that reports to the OpenTelemetry
+	// providers: the function door starts them.
+	OTel bool
 }
 
 var (
@@ -115,6 +118,7 @@ var (
 			Module: inject.ImportData{Alias: "opentelemetryfx", Path: fxMiddlewareRoot + "/opentelemetry"},
 			Pkg:    inject.ImportData{Alias: "opentelemetrymw", Path: middlewareRoot + "/go.opentelemetry.io/otel/v1"},
 			Type:   "OpenTelemetry",
+			OTel:   true,
 		},
 		"prometheus": {
 			Module: inject.ImportData{Alias: "prometheusfx", Path: fxMiddlewareRoot + "/prometheus"},