-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
9e50d07
commit 756e4cd
Showing
9 changed files
with
152 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package bunnify | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| amqp "github.com/rabbitmq/amqp091-go" | ||
| "go.opentelemetry.io/otel" | ||
| "go.opentelemetry.io/otel/propagation" | ||
| ) | ||
|
|
||
| // inject the span context to amqp table | ||
| func injectToHeaders(ctx context.Context) amqp.Table { | ||
| carrier := propagation.MapCarrier{} | ||
| otel.GetTextMapPropagator().Inject(ctx, carrier) | ||
|
|
||
| header := amqp.Table{} | ||
| for k, v := range carrier { | ||
| header[k] = v | ||
| } | ||
| return header | ||
| } | ||
|
|
||
| // extract the amqp table to a span context | ||
| func extractToContext(headers amqp.Table) context.Context { | ||
| carrier := propagation.MapCarrier{} | ||
| for k, v := range headers { | ||
| value, ok := v.(string) | ||
| if ok { | ||
| carrier[k] = value | ||
| } | ||
| } | ||
|
|
||
| return otel.GetTextMapPropagator().Extract(context.TODO(), carrier) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package tests | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/google/uuid" | ||
| "github.com/pmorelli92/bunnify/bunnify" | ||
| "go.opentelemetry.io/otel" | ||
| "go.opentelemetry.io/otel/propagation" | ||
| tracesdk "go.opentelemetry.io/otel/sdk/trace" | ||
| "go.opentelemetry.io/otel/trace" | ||
| "go.uber.org/goleak" | ||
| ) | ||
|
|
||
| func TestConsumerPublisherTracing(t *testing.T) { | ||
| // Setup tracing | ||
| otel.SetTracerProvider(tracesdk.NewTracerProvider()) | ||
| otel.SetTextMapPropagator(propagation.TraceContext{}) | ||
|
|
||
| // Setup amqp | ||
| queueName := uuid.NewString() | ||
| exchangeName := uuid.NewString() | ||
| routingKey := uuid.NewString() | ||
|
|
||
| connection := bunnify.NewConnection() | ||
| connection.Start() | ||
|
|
||
| // Exercise consuming | ||
| var actualTraceID trace.TraceID | ||
| eventHandler := func(ctx context.Context, _ bunnify.ConsumableEvent[any]) error { | ||
| actualTraceID = trace.SpanFromContext(ctx).SpanContext().TraceID() | ||
| return nil | ||
| } | ||
|
|
||
| consumer := connection.NewConsumer( | ||
| queueName, | ||
| bunnify.WithBindingToExchange(exchangeName), | ||
| bunnify.WithHandler(routingKey, eventHandler)) | ||
| if err := consumer.Consume(); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| // Exercise publishing | ||
| publisher := connection.NewPublisher() | ||
| publishingContext, _ := otel.Tracer("amqp").Start(context.Background(), "publish-test") | ||
|
|
||
| err := publisher.Publish( | ||
| publishingContext, | ||
| exchangeName, | ||
| routingKey, | ||
| bunnify.NewPublishableEvent(struct{}{})) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| time.Sleep(50 * time.Millisecond) | ||
|
|
||
| if err := connection.Close(); err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| // Assert | ||
| publishingTraceID := trace.SpanFromContext(publishingContext).SpanContext().TraceID() | ||
| if actualTraceID != publishingTraceID { | ||
| t.Fatalf("expected traceID %s, got %s", publishingTraceID, actualTraceID) | ||
| } | ||
|
|
||
| goleak.VerifyNone(t) | ||
| } |