Clients v2 (#316)

cache/redis/redis.go

@@ -11,7 +11,7 @@ import (
 // Cache encapsulates a Redis-based caching mechanism,
 // driven by go-redis/redis/v7.
 type Cache struct {
-	rdb *redis.Client
+	rdb redis.Client
 	ctx context.Context
 }
 

client/amqp/amqp.go

@@ -1,4 +1,9 @@
 // Package amqp provides a client with included tracing capabilities.
+//
+// Deprecated: The AMQP client package is superseded by the `github.com/beatlabs/client/amqp/v2` package.
+// Please refer to the documents and the examples for the usage.
+//
+// This package is frozen and no new functionality will be added.
 package amqp
 
 import (
@@ -75,6 +80,11 @@ type TracedPublisher struct {
 // NewPublisher creates a new publisher with the following defaults
 // - exchange type: fanout
 // - notifications are not handled at this point TBD.
+//
+// Deprecated: The AMQP client package is superseded by the `github.com/beatlabs/client/amqp/v2` package.
+// Please refer to the documents and the examples for the usage.
+//
+// This package is frozen and no new functionality will be added.
 func NewPublisher(url, exc string, oo ...OptionFunc) (*TracedPublisher, error) {
 	if url == "" {
 		return nil, errors.New("url is required")

client/amqp/amqp_test.go

@@ -3,9 +3,9 @@ package amqp
 import (
 	"testing"
 
-	"github.com/golang/protobuf/proto"
-
+	"github.com/beatlabs/patron/examples"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 )
 
 func TestNewMessage(t *testing.T) {
@@ -24,15 +24,15 @@ func TestNewJSONMessage(t *testing.T) {
 }
 
 func TestNewProtobufMessage(t *testing.T) {
-	u := User{
+	u := examples.User{
 		Firstname: proto.String("John"),
 		Lastname:  proto.String("Doe"),
 	}
 	m, err := NewProtobufMessage(&u)
 	assert.NoError(t, err)
 	assert.Equal(t, "application/x-protobuf", m.contentType)
 	assert.Len(t, m.body, 11)
-	u = User{}
+	u = examples.User{}
 	_, err = NewProtobufMessage(&u)
 	assert.Error(t, err)
 }

client/amqp/v2/amqp.go

@@ -0,0 +1,100 @@
+// Package v2 provides a client with included tracing capabilities.
+package v2
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/beatlabs/patron/correlation"
+	patronerrors "github.com/beatlabs/patron/errors"
+	"github.com/beatlabs/patron/trace"
+	opentracing "github.com/opentracing/opentracing-go"
+	"github.com/opentracing/opentracing-go/ext"
+	"github.com/streadway/amqp"
+)
+
+const (
+	publisherComponent = "amqp-publisher"
+)
+
+// Publisher defines a RabbitMQ publisher with tracing instrumentation.
+type Publisher struct {
+	cfg        *amqp.Config
+	connection *amqp.Connection
+	channel    *amqp.Channel
+}
+
+// New constructor.
+func New(url string, oo ...OptionFunc) (*Publisher, error) {
+	if url == "" {
+		return nil, errors.New("url is required")
+	}
+
+	var err error
+	pub := &Publisher{}
+
+	for _, option := range oo {
+		err = option(pub)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	var conn *amqp.Connection
+
+	if pub.cfg == nil {
+		conn, err = amqp.Dial(url)
+	} else {
+		conn, err = amqp.DialConfig(url, *pub.cfg)
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to open connection: %w", err)
+	}
+
+	ch, err := conn.Channel()
+	if err != nil {
+		return nil, patronerrors.Aggregate(fmt.Errorf("failed to open channel: %w", err), conn.Close())
+	}
+
+	pub.connection = conn
+	pub.channel = ch
+	return pub, nil
+}
+
+// Publish a message to a exchange.
+func (tc *Publisher) Publish(ctx context.Context, exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error {
+	sp, _ := trace.ChildSpan(ctx, trace.ComponentOpName(publisherComponent, exchange),
+		publisherComponent, ext.SpanKindProducer, opentracing.Tag{Key: "exchange", Value: exchange})
+
+	if msg.Headers == nil {
+		msg.Headers = amqp.Table{}
+	}
+
+	c := amqpHeadersCarrier(msg.Headers)
+	err := sp.Tracer().Inject(sp.Context(), opentracing.TextMap, c)
+	if err != nil {
+		return fmt.Errorf("failed to inject tracing headers: %w", err)
+	}
+	msg.Headers[correlation.HeaderID] = correlation.IDFromContext(ctx)
+
+	err = tc.channel.Publish(exchange, key, mandatory, immediate, msg)
+	trace.SpanComplete(sp, err)
+	if err != nil {
+		return fmt.Errorf("failed to publish message: %w", err)
+	}
+
+	return nil
+}
+
+// Close the channel and connection.
+func (tc *Publisher) Close() error {
+	return patronerrors.Aggregate(tc.channel.Close(), tc.connection.Close())
+}
+
+type amqpHeadersCarrier map[string]interface{}
+
+// Set implements Set() of opentracing.TextMapWriter.
+func (c amqpHeadersCarrier) Set(key, val string) {
+	c[key] = val
+}

client/amqp/v2/amqp_test.go

@@ -0,0 +1,31 @@
+package v2
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestNew(t *testing.T) {
+	type args struct {
+		url string
+	}
+	tests := map[string]struct {
+		args        args
+		expectedErr string
+	}{
+		"fail, missing url": {args: args{}, expectedErr: "url is required"},
+	}
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			got, err := New(tt.args.url)
+			if tt.expectedErr != "" {
+				assert.EqualError(t, err, tt.expectedErr)
+				assert.Nil(t, got)
+			} else {
+				assert.NoError(t, err)
+				assert.NotNil(t, got)
+			}
+		})
+	}
+}

client/amqp/v2/option.go

@@ -0,0 +1,16 @@
+package v2
+
+import (
+	"github.com/streadway/amqp"
+)
+
+// OptionFunc definition for configuring the publisher in a functional way.
+type OptionFunc func(*Publisher) error
+
+// Config option for providing dial configuration.
+func Config(cfg amqp.Config) OptionFunc {
+	return func(p *Publisher) error {
+		p.cfg = &cfg
+		return nil
+	}
+}

client/amqp/v2/option_test.go

@@ -0,0 +1,18 @@
+package v2
+
+import (
+	"testing"
+
+	"github.com/streadway/amqp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestTimeout(t *testing.T) {
+	cfg := amqp.Config{
+		Locale: "123",
+	}
+
+	p := Publisher{}
+	assert.NoError(t, Config(cfg)(&p))
+	assert.Equal(t, cfg, *p.cfg)
+}

client/grpc/grpc.go

@@ -44,7 +44,6 @@ func (c *headersCarrier) Set(key, val string) {
 
 func unaryInterceptor(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn,
 	invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
-
 	span, ctx := trace.ChildSpan(ctx, trace.ComponentOpName(componentName, method), componentName, ext.SpanKindProducer,
 		ext.SpanKindProducer)