Add delayq message format

Author: aclisp

delayq/message.go

@@ -0,0 +1,53 @@
+package delayq
+
+import (
+	"bytes"
+	"context"
+	"encoding/gob"
+
+	ztrace "github.com/zeromicro/go-zero/core/trace"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/propagation"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type (
+	Action  string
+	Message struct {
+		Carrier propagation.MapCarrier
+		Action  Action
+		Body    []byte
+	}
+)
+
+func (m *Message) Encode() ([]byte, error) {
+	var buf bytes.Buffer
+	enc := gob.NewEncoder(&buf)
+	if err := enc.Encode(m); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (m *Message) Inject(ctx context.Context) (context.Context, trace.Span) {
+	tracer := otel.Tracer(ztrace.TraceName)
+	spanCtx, span := tracer.Start(ctx, string(m.Action), trace.WithSpanKind(trace.SpanKindProducer))
+	m.Carrier = propagation.MapCarrier{}
+	propagator := otel.GetTextMapPropagator()
+	propagator.Inject(spanCtx, m.Carrier)
+	return spanCtx, span
+}
+
+func (m *Message) Decode(data []byte) error {
+	buf := bytes.NewBuffer(data)
+	dec := gob.NewDecoder(buf)
+	return dec.Decode(m)
+}
+
+func (m *Message) Extract(ctx context.Context) (context.Context, trace.Span) {
+	propagator := otel.GetTextMapPropagator()
+	ctx = propagator.Extract(ctx, m.Carrier)
+	tracer := otel.Tracer(ztrace.TraceName)
+	spanCtx, span := tracer.Start(ctx, string(m.Action), trace.WithSpanKind(trace.SpanKindConsumer))
+	return spanCtx, span
+}

delayq/message_test.go

@@ -0,0 +1,26 @@
+package delayq
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMessageEncodeDecode(t *testing.T) {
+	msg := Message{
+		Carrier: map[string]string{
+			"key1": "value1",
+			"key2": "value2",
+		},
+		Action: "action1",
+		Body:   []byte("body1"),
+	}
+	data, err := msg.Encode()
+	assert.NoError(t, err)
+
+	newMsg := Message{}
+	err = newMsg.Decode(data)
+	assert.NoError(t, err)
+
+	assert.Exactly(t, msg, newMsg)
+}

go.mod

@@ -8,6 +8,8 @@ require (
 	github.com/gorilla/sessions v1.3.0
 	github.com/stretchr/testify v1.9.0
 	github.com/zeromicro/go-zero v1.7.0
+	go.opentelemetry.io/otel v1.24.0
+	go.opentelemetry.io/otel/trace v1.24.0
 	xorm.io/xorm v1.3.9
 )
 
@@ -38,7 +40,6 @@ require (
 	github.com/robfig/cron/v3 v3.0.1 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/syndtr/goleveldb v1.0.0 // indirect
-	go.opentelemetry.io/otel v1.24.0 // indirect
 	go.opentelemetry.io/otel/exporters/jaeger v1.17.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.24.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.24.0 // indirect
@@ -47,7 +48,6 @@ require (
 	go.opentelemetry.io/otel/exporters/zipkin v1.24.0 // indirect
 	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.24.0 // indirect
-	go.opentelemetry.io/otel/trace v1.24.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/automaxprocs v1.5.3 // indirect
 	golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 // indirect