Adiciona timeout para envio

Inclui parâmetro de tempo limite e utiliza contexto com timeout na publicação.
Ajusta configuração e tratamento de atributos para garantir resposta rápida.

Author: jpfaria

wrapper/publisher/driver/contrib/cloud.google.com/pubsub/v1/config.go

@@ -14,6 +14,7 @@ const (
 	logRoot                   = ".log"
 	orderingKey               = ".orderingKey"
 	level                     = logRoot + ".level"
+	publishTimeout            = ".publishTimeout"
 	settings                  = ".settings"
 	delayThreshold            = settings + ".delayThreshold"
 	countThreshold            = settings + ".countThreshold"
@@ -60,4 +61,5 @@ func ConfigAdd(path string) {
 	config.Add(path+limitExceededBehavior, 1, "behavior when flow control limits are exceeded: Block or Ignore")
 	config.Add(path+enableCompression, false, "whether to compress messages before sending")
 	config.Add(path+compressionBytesThreshold, 240, "the minimum size a message must be to be compressed before sending")
+	config.Add(path+publishTimeout, 60*time.Second, "the maximum duration to wait for a publish to complete")
 }

wrapper/publisher/driver/contrib/cloud.google.com/pubsub/v1/options.go

@@ -1,6 +1,8 @@
 package pubsub
 
 import (
+	"time"
+
 	"cloud.google.com/go/pubsub"
 	"github.com/xgodev/boost/wrapper/config"
 )
@@ -10,8 +12,10 @@ type Options struct {
 	Log struct {
 		Level string
 	}
-	OrderingKey bool
-	Settings    pubsub.PublishSettings
+	OrderingKey    bool
+	Settings       pubsub.PublishSettings
+	Timeout        time.Duration
+	PublishTimeout time.Duration
 }
 
 // NewOptions returns options from config file or environment vars.

wrapper/publisher/driver/contrib/cloud.google.com/pubsub/v1/pubsub.go

@@ -74,7 +74,9 @@ func (p *client) send(ctx context.Context, events []*v2.Event) ([]publisher.Publ
 		go func(ev *v2.Event) {
 			defer wg.Done()
 
-			ctx := context.Background()
+			base := context.WithoutCancel(ctx)
+			pubCtx, cancel := context.WithTimeout(base, p.options.PublishTimeout)
+			defer cancel()
 
 			logger := log.WithField("subject", ev.Subject()).
 				WithField("id", ev.ID())
@@ -99,12 +101,17 @@ func (p *client) send(ctx context.Context, events []*v2.Event) ([]publisher.Publ
 				"ce_id":          ev.ID(),
 				"ce_source":      ev.Source(),
 				"ce_type":        ev.Type(),
-				"content-type":   ev.DataContentType(),
-				"ce_time":        ev.Time().String(),
+				"ce_time":        ev.Time().UTC().Format(time.RFC3339),
 				"ce_path":        "/",
 				"ce_subject":     ev.Subject(),
 			}
 
+			if ct := ev.DataContentType(); ct != "" {
+				attrs["content-type"] = ct
+			} else {
+				attrs["content-type"] = "application/json"
+			}
+
 			msg := &pubsub.Message{ID: ev.ID(), Data: raw, Attributes: attrs, PublishTime: time.Now()}
 			if p.options.OrderingKey {
 				if pk, err := p.getPartitionKey(ev); err == nil {
@@ -115,8 +122,8 @@ func (p *client) send(ctx context.Context, events []*v2.Event) ([]publisher.Publ
 			topic := p.getTopic(ev.Subject())
 			err = try.Do(func(attempt int) (bool, error) {
 				logger.Tracef("publishing to topic %s, attempt %d", ev.Subject(), attempt)
-				r := topic.Publish(ctx, msg)
-				if _, err := r.Get(ctx); err != nil {
+				r := topic.Publish(pubCtx, msg)
+				if _, err := r.Get(pubCtx); err != nil {
 					log.Error(err)
 					return attempt < 5, errors.NewInternal(err, "Pub/Sub publish failed")
 				}