func OpenTelemetryMiddleware(h message.HandlerFunc) message.HandlerFunc {
	// Setup tracing
	tr := otel.Tracer("watermill")

	// Setup metrics
	meter := global.Meter("watermill")
	offsetTracker := metric.Must(meter).NewInt64UpDownCounter("watermill_kafka_subscriber_offset")
	subLagTracker := metric.Must(meter).NewInt64Histogram("watermill_kafka_subscriber_lag_ms")
	publishTracker := metric.Must(meter).NewInt64Histogram("watermill_kafka_publish_ms")
	processingTracker := metric.Must(meter).NewInt64Histogram("watermill_processing_ms")

	return func(msg *message.Message) ([]*message.Message, error) {
		ctx := msg.Context()
		proAttr, subAttr, pubAttr := buildAttrs(ctx)

		// Measure offsets and lag from kafka write time
		o, ok := kafka.MessagePartitionOffsetFromCtx(ctx)
		if ok {
			meter.RecordBatch(
				ctx,
				subAttr,
				offsetTracker.Measurement(o),
			)
		}

		t, ok := kafka.MessageTimestampFromCtx(ctx)
		if ok {
			meter.RecordBatch(
				ctx,
				subAttr,
				subLagTracker.Measurement(time.Since(t).Milliseconds()),
			)
		}

		// Extract open telemetry data form metadata, create new span and propagate it
		ctx = otel.GetTextMapPropagator().Extract(ctx, metadataCarrier(msg.Metadata))
		ctx, span := tr.Start(ctx, "process message", trace.WithSpanKind(trace.SpanKindConsumer))
		span.SetAttributes(subAttr...)
		defer span.End()
		msg.SetContext(ctx)

		// Call next handler and measure execution time
		processingStart := time.Now()
		producedMessages, err := h(msg)
		meter.RecordBatch(
			ctx,
			proAttr,
			processingTracker.Measurement(time.Since(processingStart).Milliseconds()),
		)

		// Inject open telemetry data to metadata
		for _, m := range producedMessages {
			ctx, span := tr.Start(m.Context(), "publish messages", trace.WithSpanKind(trace.SpanKindProducer))
			span.SetAttributes(pubAttr...)

			// Track processing complete
			publishStart := time.Now()
			go func(msg *message.Message, publishStart time.Time) {
				select {
				case <-msg.Acked():
				case <-msg.Nacked():
					span.SetStatus(codes.Error, "nacked")
					span.RecordError(fmt.Errorf("message nacked"))
				}

				span.End()
				meter.RecordBatch(
					ctx,
					pubAttr,
					publishTracker.Measurement(time.Since(publishStart).Milliseconds()),
				)
			}(m, publishStart)

			otel.GetTextMapPropagator().Inject(ctx, metadataCarrier(m.Metadata))
		}

		if err != nil {
			// Mark span as error
			span.SetStatus(codes.Error, "handler failed to process message")
			span.RecordError(err)
		}

		return producedMessages, err
	}
}
// OpenTelemetryMiddleware collects metrics and injects tracing data into context
// If trace propagation is configured will extract or inject tracing data into metadata for watermill to pass it on the wire,
// If metrics are enabled collects them
OpenTelemetryMiddleware collects metrics and injects tracing data into context
If trace propagation is configured will extract or inject tracing data into metadata for watermill to pass it on the wire,
If metrics are enabled collects them
// OpenTelemetryMiddleware collects metrics and injects tracing data into context // If trace propagation is configured will extract or inject tracing data into metadata for watermill to pass it on the wire // If metrics are enabled collects them
func OpenTelemetryMiddleware(h message.HandlerFunc) message.HandlerFunc {
	tr := otel.Tracer("watermill")
	meter := global.Meter("watermill")
	offsetTracker := metric.Must(meter).NewInt64UpDownCounter("watermill_kafka_subscriber_offset")
	subLagTracker := metric.Must(meter).NewInt64Histogram("watermill_kafka_subscriber_lag_ms")
	publishTracker := metric.Must(meter).NewInt64Histogram("watermill_kafka_publish_ms")
	processingTracker := metric.Must(meter).NewInt64Histogram("watermill_processing_ms")
	return func(msg *message.Message) ([]*message.Message, error) {
		ctx := msg.Context()
		proAttr, subAttr, pubAttr := buildAttrs(ctx)
		o, ok := kafka.MessagePartitionOffsetFromCtx(ctx)
		if ok {
			meter.RecordBatch(
				ctx,
				subAttr,
				offsetTracker.Measurement(o),
			)
		}
		t, ok := kafka.MessageTimestampFromCtx(ctx)
		if ok {
			meter.RecordBatch(
				ctx,
				subAttr,
				subLagTracker.Measurement(time.Since(t).Milliseconds()),
			)
		}
		ctx = otel.GetTextMapPropagator().Extract(ctx, metadataCarrier(msg.Metadata))
		ctx, span := tr.Start(ctx, "process message", trace.WithSpanKind(trace.SpanKindConsumer))
		span.SetAttributes(subAttr...)
		defer span.End()
		msg.SetContext(ctx)
		processingStart := time.Now()
		producedMessages, err := h(msg)
		meter.RecordBatch(
			ctx,
			proAttr,
			processingTracker.Measurement(time.Since(processingStart).Milliseconds()),
		)
		for _, m := range producedMessages {
			ctx, span := tr.Start(m.Context(), "publish messages", trace.WithSpanKind(trace.SpanKindProducer))
			span.SetAttributes(pubAttr...)
			publishStart := time.Now()
			go func(msg *message.Message, publishStart time.Time) {
				select {
				case <-msg.Acked():
				case <-msg.Nacked():
					span.SetStatus(codes.Error, "nacked")
					span.RecordError(fmt.Errorf("message nacked"))
				}
				span.End()
				meter.RecordBatch(
					ctx,
					pubAttr,
					publishTracker.Measurement(time.Since(publishStart).Milliseconds()),
				)
			}(m, publishStart)
			otel.GetTextMapPropagator().Inject(ctx, metadataCarrier(m.Metadata))
		}
		if err != nil {
			span.SetStatus(codes.Error, "handler failed to process message")
			span.RecordError(err)
		}
		return producedMessages, err
	}
}
func OpenTelemetryMiddleware ( h message . HandlerFunc ) message . HandlerFunc { tr := otel . Tracer ( " " ) meter := global . Meter ( " " ) offsetTracker := metric . Must ( meter ) . NewInt64UpDownCounter ( " " ) subLagTracker := metric . Must ( meter ) . NewInt64Histogram ( " " ) publishTracker := metric . Must ( meter ) . NewInt64Histogram ( " " ) processingTracker := metric . Must ( meter ) . NewInt64Histogram ( " " ) return func ( msg * message . Message ) ( [ ] * message . Message , error ) { ctx := msg . Context ( ) proAttr , subAttr , pubAttr := buildAttrs ( ctx ) o , ok := kafka . MessagePartitionOffsetFromCtx ( ctx ) if ok { meter . RecordBatch ( ctx , subAttr , offsetTracker . Measurement ( o ) , ) } t , ok := kafka . MessageTimestampFromCtx ( ctx ) if ok { meter . RecordBatch ( ctx , subAttr , subLagTracker . Measurement ( time . Since ( t ) . Milliseconds ( ) ) , ) } ctx = otel . GetTextMapPropagator ( ) . Extract ( ctx , metadataCarrier ( msg . Metadata ) ) ctx , span := tr . Start ( ctx , " " , trace . WithSpanKind ( trace . SpanKindConsumer ) ) span . SetAttributes ( subAttr ... ) defer span . End ( ) msg . SetContext ( ctx ) processingStart := time . Now ( ) producedMessages , err := h ( msg ) meter . RecordBatch ( ctx , proAttr , processingTracker . Measurement ( time . Since ( processingStart ) . Milliseconds ( ) ) , ) for _ , m := range producedMessages { ctx , span := tr . Start ( m . Context ( ) , " " , trace . WithSpanKind ( trace . SpanKindProducer ) ) span . SetAttributes ( pubAttr ... ) publishStart := time . Now ( ) go func ( msg * message . Message , publishStart time . Time ) { select { case <- msg . Acked ( ) : case <- msg . Nacked ( ) : span . SetStatus ( codes . Error , " " ) span . RecordError ( fmt . Errorf ( " " ) ) } span . End ( ) meter . RecordBatch ( ctx , pubAttr , publishTracker . Measurement ( time . Since ( publishStart ) . Milliseconds ( ) ) , ) } ( m , publishStart ) otel . GetTextMapPropagator ( ) . Inject ( ctx , metadataCarrier ( m . Metadata ) ) } if err != nil { span . SetStatus ( codes . Error , " " ) span . RecordError ( err ) } return producedMessages , err } }
OpenTelemetryMiddleware collects metrics and injects tracing data into context
If trace propagation is configured will extract or inject tracing data into metadata for watermill to pass it on the wire,
If metrics are enabled collects them
OpenTelemetryMiddleware collects metrics and injects tracing data into context If trace propagation is configured will extract or inject tracing data into metadata for watermill to pass it on the wire If metrics are enabled collects them
// Setup tracing // Setup metrics // Measure offsets and lag from kafka write time // Extract open telemetry data form metadata, create new span and propagate it // Call next handler and measure execution time // Inject open telemetry data to metadata // Track processing complete // Mark span as error