release: 1.6.2

README.md

@@ -869,7 +869,7 @@ same or get better, never worse.
 
  require (
 -	github.com/xgodev/boost v1.1.1
-+	github.com/xgodev/boost v1.6.1
++	github.com/xgodev/boost v1.6.2
  )
 ```
 
@@ -880,7 +880,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.6.1`**
+Releases: 2024-07 → `v1.1.0` → `v1.1.1` → **`v1.6.2`**
 ([all releases](https://github.com/xgodev/boost/releases)).
 
 ## There's a factory for that.

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v1/cmd.go

@@ -6,17 +6,16 @@ import (
 	"github.com/xgodev/boost/bootstrap/function"
 )
 
-// New returns CmdFunc for cloudevents command.
+// New returns the CmdFunc of the gcp_pubsub command: it handles the messages
+// of the configured subscriptions until the run context is done.
 func New[T any](client *pubsub.Client) function.CmdFunc[T] {
 	return func(fn function.Handler[T]) *co.Command {
 		return &co.Command{
 			Use:   "gcp_pubsub",
 			Short: "gcp_pubsub",
 			Long:  "",
 			RunE: func(cmd *co.Command, args []string) error {
-				helper := NewHelper[T](client, fn)
-				helper.Start()
-				return nil
+				return NewHelper[T](client, fn).Run(cmd.Context())
 			},
 		}
 	}

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v1/helper.go

@@ -33,9 +33,28 @@ func NewHelper[T any](client *pubsub.Client, handler function.Handler[T]) *Helpe
 	return NewHelperWithOptions(client, handler, opt)
 }
 
-// Start subscribes to the topics and processes messages concurrently.
+// Start runs the helper with no way to stop it, and logs the error Run
+// returns.
+//
+// Deprecated: use Run with the run context, so the app stops on SIGTERM and
+// fails when a subscription cannot be received.
 func (h *Helper[T]) Start() {
-	logger := log.WithTypeOf(*h)
+	if err := h.Run(context.Background()); err != nil {
+		log.WithTypeOf(*h).Error(err.Error())
+	}
+}
+
+// Run receives the messages of every subscription concurrently until ctx is
+// done. Then it pulls no new message, waits for the handlers to finish the
+// ones they hold, and returns. When a subscription cannot be received, for
+// instance because it does not exist, Run stops the others and returns that
+// error, so the app fails instead of running without it.
+func (h *Helper[T]) Run(ctx context.Context) error {
+	logger := log.FromContext(ctx).WithTypeOf(*h)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	errs := make(chan error, len(h.options.Subscriptions))
 	var wg sync.WaitGroup
 
 	// Subscribe to each subscription in a goroutine
@@ -45,17 +64,16 @@ func (h *Helper[T]) Start() {
 		go func(subscription string) {
 			defer wg.Done()
 
-			subscriber := NewSubscriber[T](h.client, h.handler, subscription, h.options)
-
-			// Subscribe to the subscription
-			if err := subscriber.Subscribe(context.Background()); err != nil {
-				logger.Errorf("Failed to subscribe to subscription %s: %v", subscription, err)
-			} else {
-				logger.Infof("Successfully subscribed to subscription %s", subscription)
+			logger.Infof("pubsub: receiving on %s", subscription)
+			if err := NewSubscriber[T](h.client, h.handler, subscription, h.options).Subscribe(ctx); err != nil {
+				errs <- err
+				cancel()
 			}
 		}(subscription)
 	}
 
-	// Wait for all subscriptions to complete
+	// Wait for all subscriptions to stop
 	wg.Wait()
+	close(errs)
+	return <-errs
 }

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v1/run_test.go

@@ -0,0 +1,180 @@
+package pubsub
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"cloud.google.com/go/pubsub"
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	"github.com/stretchr/testify/require"
+	"github.com/xgodev/boost"
+)
+
+// blocking is a handler that signals when it takes a message, holds it
+// until release is closed, and records whether its context was cancelled.
+type blocking struct {
+	started   chan struct{}
+	release   chan struct{}
+	handled   atomic.Int32
+	cancelled atomic.Bool
+	fail      bool
+}
+
+func newBlocking() *blocking {
+	return &blocking{started: make(chan struct{}, 10), release: make(chan struct{})}
+}
+
+func (b *blocking) handle(ctx context.Context, _ cloudevents.Event) (cloudevents.Event, error) {
+	b.started <- struct{}{}
+	<-b.release
+	b.cancelled.Store(ctx.Err() != nil)
+	b.handled.Add(1)
+	if b.fail {
+		return cloudevents.Event{}, errors.New("handler failed")
+	}
+	return cloudevents.Event{}, nil
+}
+
+// subscription creates topic and subscription name on the pstest server of
+// client.
+func subscription(t *testing.T, ctx context.Context, client *pubsub.Client, name string) *pubsub.Topic {
+	t.Helper()
+	topic, err := client.CreateTopic(ctx, name)
+	require.NoError(t, err)
+	_, err = client.CreateSubscription(ctx, name, pubsub.SubscriptionConfig{Topic: topic, AckDeadline: 10 * time.Second})
+	require.NoError(t, err)
+	return topic
+}
+
+func publish(t *testing.T, ctx context.Context, topic *pubsub.Topic, data string) {
+	t.Helper()
+	_, err := topic.Publish(ctx, &pubsub.Message{Data: []byte(data)}).Get(ctx)
+	require.NoError(t, err)
+}
+
+func options(subscription string) *Options {
+	return &Options{Subscriptions: []string{subscription}, Concurrency: 1, RetryLimit: -1, BackoffBase: time.Millisecond, MaxBackoff: time.Millisecond, ProcessTimeout: 10 * time.Second}
+}
+
+// TestHelperRun_StopsOnTheRunContext cancels the run context while the
+// handler holds a message (#204): the handler finishes with a context the
+// stop did not cancel, Run returns, and no new message is taken.
+func TestHelperRun_StopsOnTheRunContext(t *testing.T) {
+	boost.Start()
+	client, ctx := setup(t)
+	topic := subscription(t, ctx, client, "orders")
+	h := newBlocking()
+	helper := NewHelperWithOptions[cloudevents.Event](client, h.handle, options("orders"))
+
+	runCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	ran := make(chan error, 1)
+	go func() { ran <- helper.Run(runCtx) }()
+
+	publish(t, ctx, topic, `{"item":"book"}`)
+	select {
+	case <-h.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the handler never took the message")
+	}
+
+	cancel()
+	select {
+	case err := <-ran:
+		t.Fatalf("Run returned (%v) while the handler held a message", err)
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	close(h.release)
+	select {
+	case err := <-ran:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run did not return after the handler finished")
+	}
+	require.EqualValues(t, 1, h.handled.Load())
+	require.False(t, h.cancelled.Load(), "the stop cancelled the context of the handler")
+
+	publish(t, ctx, topic, `{"item":"pen"}`)
+	time.Sleep(300 * time.Millisecond)
+	require.EqualValues(t, 1, h.handled.Load(), "a message published after Run returned was taken")
+}
+
+// TestHelperRun_StopsRetryingOnTheRunContext: a handler that fails while the
+// app stops is not retried; the message goes back to Pub/Sub.
+func TestHelperRun_StopsRetryingOnTheRunContext(t *testing.T) {
+	boost.Start()
+	client, ctx := setup(t)
+	topic := subscription(t, ctx, client, "orders")
+	h := newBlocking()
+	h.fail = true
+	helper := NewHelperWithOptions[cloudevents.Event](client, h.handle, options("orders"))
+
+	runCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	ran := make(chan error, 1)
+	go func() { ran <- helper.Run(runCtx) }()
+
+	publish(t, ctx, topic, `{"item":"book"}`)
+	select {
+	case <-h.started:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the handler never took the message")
+	}
+	cancel()
+	close(h.release)
+	select {
+	case err := <-ran:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run kept retrying after the stop")
+	}
+	require.EqualValues(t, 1, h.handled.Load())
+}
+
+// TestHelperRun_FailsWhenASubscriptionIsMissing: one of the subscriptions
+// does not exist (#205). Run stops the others and returns an error naming
+// it, so the app fails its boot instead of running without it.
+func TestHelperRun_FailsWhenASubscriptionIsMissing(t *testing.T) {
+	boost.Start()
+	client, ctx := setup(t)
+	subscription(t, ctx, client, "orders")
+	opts := options("orders")
+	opts.Subscriptions = append(opts.Subscriptions, "missing")
+	helper := NewHelperWithOptions[cloudevents.Event](client, newBlocking().handle, opts)
+
+	ran := make(chan error, 1)
+	go func() { ran <- helper.Run(ctx) }()
+	select {
+	case err := <-ran:
+		require.ErrorContains(t, err, "pubsub: could not receive on missing")
+	case <-time.After(5 * time.Second):
+		t.Fatal("Run kept running without subscription missing")
+	}
+}
+
+// TestNew_RunsUntilTheCommandContextIsDone: the gcp_pubsub command runs the
+// helper, with the subscriptions of config, until the context cobra
+// executes it with is done.
+func TestNew_RunsUntilTheCommandContextIsDone(t *testing.T) {
+	boost.Start()
+	client, ctx := setup(t)
+	subscription(t, ctx, client, "changeme")
+	cmd := New[cloudevents.Event](client)(newBlocking().handle)
+	require.Equal(t, "gcp_pubsub", cmd.Name())
+
+	runCtx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
+	defer cancel()
+	cmd.SetArgs([]string{})
+	done := make(chan error, 1)
+	go func() { done <- cmd.ExecuteContext(runCtx) }()
+	select {
+	case err := <-done:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("the command did not stop with its context")
+	}
+}

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v1/subscriber.go

@@ -32,7 +32,9 @@ func NewSubscriber[T any](client *pubsub.Client, handler function.Handler[T], su
 	}
 }
 
-// Subscribe subscribes and consumes messages from multiple Pub/Sub topics concurrently
+// Subscribe receives the messages of the subscription until ctx is done, and
+// returns once the handlers finished the ones they hold. It returns an error
+// naming the subscription when it cannot be received.
 func (l *Subscriber[T]) Subscribe(ctx context.Context) error {
 	logger := log.FromContext(ctx).WithTypeOf(*l)
 
@@ -53,7 +55,7 @@ func (l *Subscriber[T]) Subscribe(ctx context.Context) error {
 	})
 
 	if err != nil {
-		logger.Fatalf("Failed to start subscription %s: %v", l.subscription, err)
+		return errors.Wrap(err, errors.Internalf("pubsub: could not receive on %s", l.subscription))
 	}
 
 	return nil
@@ -72,8 +74,9 @@ func (l *Subscriber[T]) processMessage(ctx context.Context, msg *pubsub.Message)
 	}
 
 	for {
-		// Timeout por tentativa
-		msgCtx, cancel := context.WithTimeout(ctx, l.options.ProcessTimeout)
+		// Timeout por tentativa. The stop of the app cancels ctx: the handler
+		// finishes the message it holds instead of failing on it.
+		msgCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), l.options.ProcessTimeout)
 
 		// Processes the event via handler
 		if _, err := l.handler(msgCtx, in); err != nil {
@@ -87,6 +90,11 @@ func (l *Subscriber[T]) processMessage(ctx context.Context, msg *pubsub.Message)
 				return errors.Wrap(err, errors.Internalf("max retry limit reached"))
 			}
 
+			// The app is stopping: give the message back instead of retrying
+			if ctx.Err() != nil {
+				return errors.Wrap(err, errors.Internalf("stopped before a retry"))
+			}
+
 			// Apply backoff if enabled
 			if l.options.Backoff {
 				l.applyBackoff(retryCount)

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v2/cmd.go

@@ -14,9 +14,7 @@ func New[T any](client *pubsub.Client) function.CmdFunc[T] {
 			Short: "gcp_pubsub",
 			Long:  "",
 			RunE: func(cmd *co.Command, args []string) error {
-				helper := NewHelper[T](client, fn)
-				helper.Start()
-				return nil
+				return NewHelper[T](client, fn).Run(cmd.Context())
 			},
 		}
 	}

bootstrap/function/adapter/contrib/cloud.google.com/pubsub/v2/helper.go

@@ -2,6 +2,7 @@ package pubsub
 
 import (
 	"context"
+	"sync"
 
 	"cloud.google.com/go/pubsub/v2"
 	"github.com/xgodev/boost/bootstrap/function"
@@ -33,17 +34,47 @@ func NewHelper[T any](client *pubsub.Client, handler function.Handler[T]) *Helpe
 	return NewHelperWithOptions(client, handler, opt)
 }
 
-// Start subscribes to the topics and processes messages concurrently.
+// Start runs the helper with no way to stop it, and logs the error Run
+// returns.
+//
+// Deprecated: use Run with the run context, so the app stops on SIGTERM and
+// fails when a subscription cannot be received.
 func (h *Helper[T]) Start() {
+	if err := h.Run(context.Background()); err != nil {
+		log.WithTypeOf(*h).Error(err.Error())
+	}
+}
+
+// Run receives the messages of every subscription concurrently until ctx is
+// done. Then it pulls no new message, waits for the handlers to finish the
+// ones they hold, and returns. When a subscription cannot be received, for
+// instance because it does not exist, Run stops the others and returns that
+// error, so the app fails instead of running without it.
+func (h *Helper[T]) Run(ctx context.Context) error {
+	logger := log.FromContext(ctx).WithTypeOf(*h)
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	errs := make(chan error, len(h.options.Subscriptions))
+	var wg sync.WaitGroup
 
 	// Subscribe to each subscription in a goroutine
 	for _, subscription := range h.options.Subscriptions {
+		wg.Add(1)
+
 		go func(subscription string) {
-			// Subscribe to the subscription
-			if err := NewSubscriber[T](h.client, h.handler, subscription, h.options).Subscribe(context.Background()); err != nil {
-				logger := log.WithTypeOf(*h)
-				logger.Errorf("Failed to subscribe to subscription %s: %v", subscription, err)
+			defer wg.Done()
+
+			logger.Infof("pubsub: receiving on %s", subscription)
+			if err := NewSubscriber[T](h.client, h.handler, subscription, h.options).Subscribe(ctx); err != nil {
+				errs <- err
+				cancel()
 			}
 		}(subscription)
 	}
+
+	// Wait for all subscriptions to stop
+	wg.Wait()
+	close(errs)
+	return <-errs
 }