事件驱动系列—watermill

type Publisher interface {
	Publish(topic string, messages ...*Message) error
	Close() error
}

func NewRedisPublisher(rdb *redis.Client, watermillLogger watermill.LoggerAdapter) message.Publisher {
	var pub message.Publisher
	pub, err := redisstream.NewPublisher(redisstream.PublisherConfig{Client: rdb}, watermillLogger)
	if err != nil {
		panic(err)
	}
	pub = observability.TracingPublisherDecorator{pub}
	return pub
}

type TracingPublisherDecorator struct {
	message.Publisher
}

func (p TracingPublisherDecorator) Publish(topic string, messages ...*message.Message) error {
	for i := range messages {
		otel.GetTextMapPropagator().Inject(messages[i].Context(), propagation.MapCarrier(messages[i].Metadata))
	}
	return p.Publisher.Publish(topic, messages...)
}

pub, err := redisstream.NewPublisher(redisstream.PublisherConfig{Client: rdb}, watermillLogger)
if err != nil {
  panic(err)
}
pub = observability.TracingPublisherDecorator{pub}

type PublisherConfig struct {
	Client     redis.UniversalClient
	Marshaller Marshaller
	Maxlens    map[string]int64
}

  logger := watermill.NewStdLogger(false, false)

  router, err := message.NewRouter(message.RouterConfig{}, logger)
  if err != nil {
    panic(err)
  }
  rdb := redis.NewClient(&redis.Options{
    Addr: os.Getenv("REDIS_ADDR"),
  })
  sub, err := redisstream.NewSubscriber(redisstream.SubscriberConfig{
    Client: rdb,
  }, logger)
  if err != nil {
    panic(err)
  }

  pub, err := redisstream.NewPublisher(redisstream.PublisherConfig{
    Client: rdb,
  }, logger)
  if err != nil {
    panic(err)
  }

  router.AddHandler(
    "handler for test",
    "subscriber_topic",
    sub,
    "publisher_topic",
    pub,
    func(msg *message.Message) ([]*message.Message, error) {
      // 处理逻辑
      return []*message.Message{msg}, nil
    },
  )
	err = router.Run(context.Background())
	if err != nil {
		panic(err)
	}

router.AddNoPublisherHandler(
		"handler for test",
		"subscriber_topic",
		sub,
		func(msg *message.Message) error {
			// 处理逻辑
			return nil
		},
	)

  router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc {
		return func(msg *message.Message) ([]*message.Message, error) {
			logger.Info("log", watermill.LogFields{"payload":string(msg.Payload)})
			return h(msg)
		}
	})

router.AddMiddleware(middleware.Recoverer)

var (
	messageProcessTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "messages",
			Name:      "processed_total",
			Help:      "The total processed messages",
		},
		[]string{"topic", "handler"},
	)
	messageProcessFailedTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "messages",
			Name:      "processed_failed_total",
			Help:      "The total processed messages",
		},
		[]string{"topic", "handler"},
	)
	messagesProcessingDuration = promauto.NewSummaryVec(
		prometheus.SummaryOpts{
			Namespace:  "messages",
			Name:       "processing_duration_seconds",
			Help:       "The total time spent processing messages",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"topic", "handler"},
	)
)

router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc {
		return func(msg *message.Message) ([]*message.Message, error) {
			topic := message.SubscribeTopicFromCtx(msg.Context())
			handler := message.HandlerNameFromCtx(msg.Context())
			labels := prometheus.Labels{"topic": topic, "handler": handler}
			messageProcessTotal.With(labels).Inc()
			now := time.Now()
			msgs, err := h(msg)
			if err != nil {
				messageProcessFailedTotal.With(labels).Inc()
			}

			messagesProcessingDuration.With(labels).Observe(time.Now().Sub(now).Seconds())
			return msgs, err
		}
	})

router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc {
		return func(msg *message.Message) (events []*message.Message, err error) {
			topic := message.SubscribeTopicFromCtx(msg.Context())
			handler := message.HandlerNameFromCtx(msg.Context())

			ctx := otel.GetTextMapPropagator().Extract(msg.Context(), propagation.MapCarrier(msg.Metadata))
			ctx, span := otel.Tracer("").Start(
				ctx,
				fmt.Sprintf("topic: %s, handler: %s", topic, handler),
				trace.WithAttributes(
					attribute.String("topic", topic),
					attribute.String("handler", handler),
				),
			)
			defer span.End()
			msg.SetContext(ctx)

			msgs, err := h(msg)
			if err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
			}
			return msgs, err
		}
	})

const correlationIDMessageMetadataKey = "correlation_id"
const correlationIDKey = "correlation_key"

type CorrelationPublisherDecorator struct {
	message.Publisher
}

func (c CorrelationPublisherDecorator) Publish(topic string, messages ...*message.Message) error {
	for i := range messages {
		// if correlation_id is already set, let's not override
		if messages[i].Metadata.Get(correlationIDMessageMetadataKey) != "" {
			continue
		}

		// correlation_id as const
		messages[i].Metadata.Set(correlationIDMessageMetadataKey, CorrelationIDFromContext(messages[i].Context()))
	}

	return c.Publisher.Publish(topic, messages...)
}

func CorrelationIDFromContext(ctx context.Context) string {
	v, ok := ctx.Value(correlationIDKey).(string)
	if ok {
		return v
	}

	// add "gen_" prefix to distinguish generated correlation IDs from correlation IDs passed by the client
	// it's useful to detect if correlation ID was not passed properly
	return "gen_" + shortuuid.New()
}

func main() {
  // ....
  router.AddPublisherDecorators(func(pub message.Publisher) (message.Publisher, error) {
		return CorrelationPublisherDecorator{pub}, nil
	})
  // ....
}

router.AddSubscriberDecorators(func(sub message.Subscriber) (message.Subscriber, error) {
		return CorrelationSubscriberDecorator{pub}, nil
	})

router.AddPlugin(func(router *message.Router) error {
		<-router.Running()
		fmt.Println("started")
		return nil
	})

func NewBus(publisher message.Publisher, logger watermill.LoggerAdapter) *cqrs.EventBus {
	eventBus, err := cqrs.NewEventBusWithConfig(
		publisher,
		cqrs.EventBusConfig{
			GeneratePublishTopic: func(params cqrs.GenerateEventPublishTopicParams) (string, error) {
				event, ok := params.Event.(entities.Event)
				if !ok {
					return "", fmt.Errorf("invalid event type: %T doesn't implement entities.Event", params.Event)
				}
				return params.EventName, nil
			},
			Marshaler: cqrs.JSONMarshaler{
				GenerateName: cqrs.StructName,
			},
			Logger: logger,
		})
	if err != nil {
		panic(err)
	}
	return eventBus
}

type EventBusConfig struct {
	// GeneratePublishTopic is used to generate topic name for publishing event.
	GeneratePublishTopic GenerateEventPublishTopicFn

	// OnPublish is called before sending the event.
	// The *message.Message can be modified.
	//
	// This option is not required.
	OnPublish OnEventSendFn

	// Marshaler is used to marshal and unmarshal events.
	// It is required.
	Marshaler CommandEventMarshaler

	// Logger instance used to log.
	// If not provided, watermill.NopLogger is used.
	Logger watermill.LoggerAdapter
}

func (c EventBus) Publish(ctx context.Context, event any) error {}

func NewEventProcessorWithConfig(router *message.Router, config EventProcessorConfig) (*EventProcessor, error) {
	config.setDefaults()

	if err := config.Validate(); err != nil {
		return nil, errors.Wrap(err, "invalid config EventProcessor")
	}
	if router == nil && !config.disableRouterAutoAddHandlers {
		return nil, errors.New("missing router")
	}

	return &EventProcessor{
		router: router,
		config: config,
	}, nil
}

type EventProcessorConfig struct {
	// GenerateSubscribeTopic is used to generate topic for subscribing to events.
	// If event processor is using handler groups, GenerateSubscribeTopic is used instead.
	GenerateSubscribeTopic EventProcessorGenerateSubscribeTopicFn

	// SubscriberConstructor is used to create subscriber for EventHandler.
	//
	// This function is called for every EventHandler instance.
	// If you want to re-use one subscriber for multiple handlers, use GroupEventProcessor instead.
	SubscriberConstructor EventProcessorSubscriberConstructorFn

	// OnHandle is called before handling event.
	// OnHandle works in a similar way to middlewares: you can inject additional logic before and after handling a event.
	//
	// Because of that, you need to explicitly call params.Handler.Handle() to handle the event.
	//
	//  func(params EventProcessorOnHandleParams) (err error) {
	//      // logic before handle
	//      //  (...)
	//
	//      err := params.Handler.Handle(params.Message.Context(), params.Event)
	//
	//      // logic after handle
	//      //  (...)
	//
	//      return err
	//  }
	//
	// This option is not required.
	OnHandle EventProcessorOnHandleFn

	// AckOnUnknownEvent is used to decide if message should be acked if event has no handler defined.
	AckOnUnknownEvent bool

	// Marshaler is used to marshal and unmarshal events.
	// It is required.
	Marshaler CommandEventMarshaler

	// Logger instance used to log.
	// If not provided, watermill.NopLogger is used.
	Logger watermill.LoggerAdapter

	// disableRouterAutoAddHandlers is used to keep backwards compatibility.
	// it is set when EventProcessor is created by NewEventProcessor.
	// Deprecated: please migrate to NewEventProcessorWithConfig.
	disableRouterAutoAddHandlers bool
}

func NewEventHandler[T any](handlerName string, handleFunc func(ctx context.Context, event *T) error) EventHandler {
	return &genericEventHandler[T]{
		handleFunc:  handleFunc,
		handlerName: handlerName,
	}
}

type EventHandler interface {
	// HandlerName is the name used in message.Router while creating handler.
	//
	// It will be also passed to EventsSubscriberConstructor.
	// May be useful, for example, to create a consumer group per each handler.
	//
	// WARNING: If HandlerName was changed and is used for generating consumer groups,
	// it may result with **reconsuming all messages** !!!
	HandlerName() string

	NewEvent() any

	Handle(ctx context.Context, event any) error
}

func NewBus(redisPublisher message.Publisher, watermillLogger *log.WatermillLogrusAdapter) *cqrs.CommandBus {
	commandBus, err := cqrs.NewCommandBusWithConfig(
		redisPublisher,
		cqrs.CommandBusConfig{
			GeneratePublishTopic: func(params cqrs.CommandBusGeneratePublishTopicParams) (string, error) {
				return "commands." + params.CommandName, nil
			},
			Marshaler: cqrs.JSONMarshaler{
				GenerateName: cqrs.StructName,
			},
			Logger: watermillLogger,
		})
	if err != nil {
		panic(err)
	}
	return commandBus
}

// 使用示例
func Send(/*params*/) {
 bus := NewBus(params)
 bus.Send(ctx, Command)
}

type SendNotification struct {
	NotificationID string
	Email          string
	Message        string
}

type Sender interface {
	SendNotification(ctx context.Context, notificationID, email, message string) error
}

func NewProcessor(router *message.Router, sender Sender, sub message.Subscriber, watermillLogger watermill.LoggerAdapter) *cqrs.CommandProcessor {
	eventProcessor, err := cqrs.NewCommandProcessorWithConfig(
		router,
		cqrs.CommandProcessorConfig{
			GenerateSubscribeTopic: func(params cqrs.CommandProcessorGenerateSubscribeTopicParams) (string, error) {
				return "commands." + params.CommandName, nil
			},
			SubscriberConstructor: func(params cqrs.CommandProcessorSubscriberConstructorParams) (message.Subscriber, error) {
				return sub, nil
			},
			Marshaler: cqrs.JSONMarshaler{
				GenerateName: cqrs.StructName,
			},
			Logger: watermillLogger,
		},
	)
	if err != nil {
		panic(err)
	}

	err = eventProcessor.AddHandlers(cqrs.NewCommandHandler(
		"send_notification",
		func(ctx context.Context, event *SendNotification) error {
			return sender.SendNotification(ctx, event.NotificationID, event.Email, event.Message)
		},
	))
	if err != nil {
		panic(err)
	}

	return eventProcessor
}

err := commandProcessor.AddHandlers(
    requestreply.NewCommandHandler(
        "hotel_room_booking",
        ts.RequestReplyBackend,
        func(ctx context.Context, cmd *BookHotelRoom) error {
            return fmt.Errorf("some error")
        },
    ),
)

reply, err := requestreply.SendWithReply[requestreply.NoResult](
    context.Background(),
    ts.CommandBus,
    ts.RequestReplyBackend,
    &BookHotelRoom{ID: "1"},
)

err := commandProcessor.AddHandlers(
    requestreply.NewCommandHandlerWithResult[PayForRoom, PayForRoomResult](
        "pay_for_room",
        ts.RequestReplyBackend,
        func(ctx context.Context, cmd *PayForRoom) (PayForRoomResult, error) {
            return PayForRoomResult{PaymentReference: "1234"}, nil
        },
    ),
)
// ...
reply, err := requestreply.SendWithReply[requestreply.NoResult](
    context.Background(),
    ts.CommandBus,
    ts.RequestReplyBackend,
    &TestCommand{ID: "1"},
)

// ...

fmt.Println(reply.Result.PaymentReference) // it's equal to "1234"
fmt.Println(reply.Error) // it's nil

func PublishInTx(
	message *message.Message,
	tx *sql.Tx,
	logger watermill.LoggerAdapter,
) error {
	publisher, err := watermillSQL.NewPublisher(
		tx,
		watermillSQL.PublisherConfig{
			SchemaAdapter: watermillSQL.DefaultPostgreSQLSchema{},
      AutoInitializeSchema: true,
		},
		logger,
	)
	if err != nil {
		return err
	}
	return publisher.Publish("ItemAddedToCart", message)
}

func SubscribeForMessages(db *sqlx.DB, topic string, logger watermill.LoggerAdapter) (<-chan *message.Message, error) {
	subscriber, err := sql.NewSubscriber(
		db,
		sql.SubscriberConfig{
			SchemaAdapter:    sql.DefaultPostgreSQLSchema{},
			OffsetsAdapter:   sql.DefaultPostgreSQLOffsetsAdapter{},
			InitializeSchema: true,
		},
		logger,
	)
	if err != nil {
		return nil, err
	}
	messages, err := subscriber.Subscribe(context.Background(), topic)
	if err != nil {
		panic(err)
	}
	return messages, nil
}

CREATE TABLE IF NOT EXISTS [table name] (
    consumer_group VARCHAR(255) NOT NULL,
    offset_acked BIGINT,
    offset_consumed BIGINT NOT NULL,
    PRIMARY KEY(consumer_group)
;

var outboxTopic = "events_to_forward"

func PublishInTx(
	msg *message.Message,
	tx *sql.Tx,
	logger watermill.LoggerAdapter,
) error {
	publisher, err := watermillSQL.NewPublisher(
		tx,
		watermillSQL.PublisherConfig{
			SchemaAdapter: watermillSQL.DefaultPostgreSQLSchema{},
		},
		logger,
	)
	if err != nil {
		return err
	}

	fp := forwarder.NewPublisher(publisher, forwarder.PublisherConfig{
		ForwarderTopic: outboxTopic,
	})
	defer fp.Close()
	return fp.Publish("TopicName", msg)
}

func RunForwarder(
	db *sqlx.DB,
	rdb *redis.Client,
	outboxTopic string,
	logger watermill.LoggerAdapter,
) error {
	publisher, err := redisstream.NewPublisher(
		redisstream.PublisherConfig{
			Client: rdb,
		},
		logger,
	)
	if err != nil {
		return err
	}
	subscriber, err := sql.NewSubscriber(
		db,
		sql.SubscriberConfig{
			SchemaAdapter:    sql.DefaultPostgreSQLSchema{},
			OffsetsAdapter:   sql.DefaultPostgreSQLOffsetsAdapter{},
			InitializeSchema: true,
		}, logger,
	)
	if err != nil {
		return err
	}
	fd, err := forwarder.NewForwarder(subscriber, publisher, logger, forwarder.Config{
		ForwarderTopic: outboxTopic,
	})
	if err != nil {
		return err
	}
	go func() {
		err = fd.Run(context.Background())
		if err != nil {
			panic(err)
		}
	}()
	<-fd.Running()
	return nil
}

	eventProcessor, err := cqrs.NewEventGroupProcessorWithConfig(
		router,
		cqrs.EventGroupProcessorConfig{
			GenerateSubscribeTopic: func(params cqrs.EventGroupProcessorGenerateSubscribeTopicParams) (string, error) {
				return "events", nil
			},
			SubscriberConstructor: func(params cqrs.EventGroupProcessorSubscriberConstructorParams) (message.Subscriber, error) {
        // use ConsumerGroup
				sub, err := kafka.NewSubscriber(kafka.SubscriberConfig{
					Brokers:               []string{kafkaAddr},
					Unmarshaler:           kafka.DefaultMarshaler{},
					OverwriteSaramaConfig: newConfig(),
					ConsumerGroup:         params.EventGroupName,
				}, logger)
				if err != nil {
					return nil, err
				}
				return sub, nil
			},
			AckOnUnknownEvent: true,
			Marshaler:         cqrs.JSONMarshaler{},
			Logger:            logger,
		},
	)
	if err != nil {
		panic(err)
	}

	pub, _ := kafka.NewPublisher(kafka.PublisherConfig{
		Brokers:   []string{kafkaAddr},
		Marshaler: kafka.DefaultMarshaler{},
	}, logger)

	eventProcessor.AddHandlersGroup(
		"employee-related",
		cqrs.NewGroupEventHandler(HandleEmployeePositionTransfered),
		cqrs.NewGroupEventHandler(HandleEmployeeLeft),
	)

// newConfig
func newConfig() *sarama.Config {
	cfg := sarama.NewConfig()
	cfg.Consumer.Offsets.Initial = sarama.OffsetOldest
	return cfg
}

pub, err := kafka.NewPublisher(kafka.PublisherConfig{
		Brokers: []string{kafkaAddr},
		Marshaler: kafka.NewWithPartitioningMarshaler(func(topic string, msg *message.Message) (string, error) {
			return msg.Metadata.Get("employee_id"), nil
		}),
	}, logger)

cqrs.NewEventHandler("OnEmployeeLeft", func(ctx context.Context, event *EmployeeLeft) error {
			employee := getEmployee(event.EmployeeID)

			if event.Version-1 != employee.Version {
				return fmt.Errorf("version not match")
			}

			employee.IsLeft = true
			employee.Version = event.Version

			return nil
})

var (
	messageProcessTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "messages",
			Name:      "processed_total",
			Help:      "The total processed messages",
		},
		[]string{"topic", "handler"},
	)
	messageProcessFailedTotal = promauto.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: "messages",
			Name:      "processed_failed_total",
			Help:      "The total processed messages",
		},
		[]string{"topic", "handler"},
	)
	messagesProcessingDuration = promauto.NewSummaryVec(
		prometheus.SummaryOpts{
			Namespace:  "messages",
			Name:       "processing_duration_seconds",
			Help:       "The total time spent processing messages",
			Objectives: map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"topic", "handler"},
	)
)

func useMiddlewares(router *message.Router, watermillLogger watermill.LoggerAdapter) {
	router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc {
		return func(msg *message.Message) ([]*message.Message, error) {
			topic := message.SubscribeTopicFromCtx(msg.Context())
			handler := message.HandlerNameFromCtx(msg.Context())
			labels := prometheus.Labels{"topic": topic, "handler": handler}
			messageProcessTotal.With(labels).Inc()
			now := time.Now()
			msgs, err := h(msg)
			if err != nil {
				messageProcessFailedTotal.With(labels).Inc()
			}

			messagesProcessingDuration.With(labels).Observe(time.Now().Sub(now).Seconds())
			return msgs, err
		}
	})
}

// echo
import (
  libHttp "github.com/ThreeDotsLabs/go-event-driven/common/http"
  "github.com/prometheus/client_golang/prometheus/promhttp"
  "github.com/labstack/echo/v4"
)

func NewHttpRouter(){
  e := libHttp.NewEcho()
	e.GET("/metrics", echo.WrapHandler(promhttp.Handler()))
}

// iris
import (
	"github.com/kataras/iris/v12"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

func NewHttpRouter(){
	app := iris.New()
	app.Get("/metrics", iris.FromStd(promhttp.Handler()))
}

type TracingPublisherDecorator struct {
	message.Publisher
}

func (p TracingPublisherDecorator) Publish(topic string, messages ...*message.Message) error {
	for i := range messages {
		otel.GetTextMapPropagator().Inject(messages[i].Context(), propagation.MapCarrier(messages[i].Metadata))
	}
	return p.Publisher.Publish(topic, messages...)
}

var pub message.Publisher
pub, _ = redisstream.NewPublisher(...)
pub = PublishDecorator{pub}

func useMiddlewares(router *message.Router, watermillLogger watermill.LoggerAdapter) {
	router.AddMiddleware(func(h message.HandlerFunc) message.HandlerFunc {
		return func(msg *message.Message) (events []*message.Message, err error) {
			topic := message.SubscribeTopicFromCtx(msg.Context())
			handler := message.HandlerNameFromCtx(msg.Context())

			ctx := otel.GetTextMapPropagator().Extract(msg.Context(), propagation.MapCarrier(msg.Metadata))
			ctx, span := otel.Tracer("").Start(
				ctx,
				fmt.Sprintf("topic: %s, handler: %s", topic, handler),
				trace.WithAttributes(
					attribute.String("topic", topic),
					attribute.String("handler", handler),
				),
			)
			defer span.End()
			msg.SetContext(ctx)

			msgs, err := h(msg)
			if err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
			}
			return msgs, err
		}
	})
}

import (
	"github.com/jmoiron/sqlx"
	_ "github.com/lib/pq"
	"github.com/uptrace/opentelemetry-go-extra/otelsql"
	"go.opentelemetry.io/otel/semconv/v1.4.0"
)

func getDB() *sqlx.DB {
	traceDB, err := otelsql.Open(
		"postgres",
		os.Getenv("POSTGRES_URL"),
		otelsql.WithAttributes(semconv.DBSystemPostgreSQL),
		otelsql.WithDBName("db"),
	)
	if err != nil {
		panic(err)
	}
	dbConn := sqlx.NewDb(traceDB, "postgres")
	return dbConn
}

import (
	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
  "net/http"
)

func getHttpClient() *http.Client {
	traceHttpClient := &http.Client{
		Transport: otelhttp.NewTransport(
			http.DefaultTransport,
			otelhttp.WithSpanNameFormatter(func(operation string, r *http.Request) string {
				return fmt.Sprintf("HTTP %s %s %s", r.Method, r.URL.String(), operation)
			})),
	}
}

package observability

import (
	"fmt"
	"github.com/ThreeDotsLabs/watermill/message"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/jaeger"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/sdk/resource"
	"go.opentelemetry.io/otel/sdk/trace"
	"go.opentelemetry.io/otel/semconv/v1.21.0"
	"os"
)

func ConfigureTraceProvider() *trace.TracerProvider {
	jaegerEndpoint := os.Getenv("JAEGER_ENDPOINT")
	if jaegerEndpoint == "" {
		jaegerEndpoint = fmt.Sprintf("%s/jaeger-api/api/traces", os.Getenv("GATEWAY_ADDR"))
	}

	exp, err := jaeger.New(
		jaeger.WithCollectorEndpoint(
			jaeger.WithEndpoint(jaegerEndpoint),
		),
	)
	if err != nil {
		panic(err)
	}

	tp := trace.NewTracerProvider(
		// WARNING: `tracesdk.WithSyncer` should be not used in production.
		// For production, you should use `tracesdk.WithBatcher`.
		trace.WithSyncer(exp),
		trace.WithResource(resource.NewWithAttributes(
			semconv.SchemaURL,
			semconv.ServiceName("tickets"),
		)),
	)

	otel.SetTracerProvider(tp)

	// Don't forget this line! Omitting it will cause the trace to not be propagated via messages.
	otel.SetTextMapPropagator(propagation.TraceContext{})

	return tp
}

func PublishInTx(
	msg *message.Message,
	tx *sql.Tx,
	logger watermill.LoggerAdapter,
) error {
	var publisher message.Publisher
	var err error

	publisher, err = watermillSQL.NewPublisher(
		tx,
		watermillSQL.PublisherConfig{
			SchemaAdapter: watermillSQL.DefaultPostgreSQLSchema{},
		},
		logger,
	)
	if err != nil {
		return fmt.Errorf("failed to create outbox publisher: %w", err)
	}

	publisher = TracingPublisherDecorator{publisher}

	publisher = forwarder.NewPublisher(publisher, forwarder.PublisherConfig{
		ForwarderTopic: outboxTopic,
	})

	publisher = TracingPublisherDecorator{publisher}

	return publisher.Publish("ItemAddedToCart", msg)
}

router, _ := message.NewRouter(message.RouterConfig{}, logger)
router.AddMiddleware(middleware.NewThrottle(10, time.Second).Middleware) // 允许一秒最多处理10个消息

import (
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/sony/gobreaker"
  "github.com/ThreeDotsLabs/watermill/message"
)


router, err := message.NewRouter(message.RouterConfig{}, logger)
if err != nil {
  return err
}
router.AddMiddleware(middleware.NewCircuitBreaker(gobreaker.Settings{
  Name:        "breaker",
  MaxRequests: 1,
  Timeout:     time.Second,
}).Middleware)

router.AddMiddleware(middleware.PoisonQueue(publisher, "poison_queue"))

pq, err := middleware.PoisonQueueWithFilter(pub, "PoisonQueue", func(err error) bool {
	var permErr PermanentError
	if errors.As(err, &permErr) && permErr.IsPermanent() {
		return true
	}

	return false
})
router.AddMiddleware(pq)

router.AddMiddleware(
	middleware.PoisonQueue(publisher, "poison_queue"),
	middleware.Retry{
		// Config
	}.Middleware,
)