lint fixes for amqp transport

Author: mattfung

transport/amqp/doc.go

@@ -1,2 +1,2 @@
-// Package AMQP provides AMQP transport
+// Package amqp implements a go-kit transport for AMQP
 package amqp

transport/amqp/encode-decode.go

@@ -16,7 +16,7 @@ type EncodeRequestFunc func(context.Context, *amqp.Publishing, interface{}) erro
 // EncodeResponseFunc encodes the passed reponse object to
 // an AMQP channel for publishing. It is designed to be used in AMQP Subscribers
 type EncodeResponseFunc func(context.Context,
-	*amqp.Delivery, AMQPChannel, *amqp.Publishing, interface{}) error
+	*amqp.Delivery, Channel, *amqp.Publishing, interface{}) error
 
 // DecodeResponseFunc extracts a user-domain response object from
 // an AMQP Delivery object. It is designed to be used in AMQP Publishers

transport/amqp/publisher.go

@@ -11,7 +11,7 @@ import (
 // Publisher wraps AMQP channel and queue and provides a method that
 // implements endpoint.Endpoint
 type Publisher struct {
-	ch      AMQPChannel
+	ch      Channel
 	q       *amqp.Queue
 	enc     EncodeRequestFunc
 	dec     DecodeResponseFunc
@@ -22,7 +22,7 @@ type Publisher struct {
 
 // NewPublisher constructs a usable Publisher for a single remote method.
 func NewPublisher(
-	ch AMQPChannel,
+	ch Channel,
 	q *amqp.Queue,
 	enc EncodeRequestFunc,
 	dec DecodeResponseFunc,

transport/amqp/publisher_test.go

@@ -11,8 +11,6 @@ import (
 	"github.com/streadway/amqp"
 )
 
-import "log"
-
 // TestBadEncode tests if encode errors are handled properly
 func TestBadEncode(t *testing.T) {
 	ch := &mockChannel{f: nullFunc}
@@ -23,15 +21,15 @@ func TestBadEncode(t *testing.T) {
 		func(context.Context, *amqp.Publishing, interface{}) error { return errors.New("err!") },
 		func(context.Context, *amqp.Delivery) (response interface{}, err error) { return struct{}{}, nil },
 	)
-	err_chan := make(chan error, 1)
+	errChan := make(chan error, 1)
 	var err error
 	go func() {
 		_, err := pub.Endpoint()(context.Background(), struct{}{})
-		err_chan <- err
+		errChan <- err
 
 	}()
 	select {
-	case err = <-err_chan:
+	case err = <-errChan:
 		break
 	case <-time.After(100 * time.Millisecond):
 		t.Fatal("Timed out waiting for result")
@@ -66,20 +64,20 @@ func TestBadDecode(t *testing.T) {
 			return struct{}{}, errors.New("err!")
 		},
 		amqptransport.PublisherBefore(
-			amqptransport.SetCorrelationId(cid),
+			amqptransport.SetCorrelationID(cid),
 		),
 	)
 
 	var err error
-	err_chan := make(chan error, 1)
+	errChan := make(chan error, 1)
 	go func() {
 		_, err := pub.Endpoint()(context.Background(), struct{}{})
-		err_chan <- err
+		errChan <- err
 
 	}()
 
 	select {
-	case err = <-err_chan:
+	case err = <-errChan:
 		break
 	case <-time.After(100 * time.Millisecond):
 		t.Fatal("Timed out waiting for result")
@@ -113,15 +111,15 @@ func TestPublisherTimeout(t *testing.T) {
 	)
 
 	var err error
-	err_chan := make(chan error, 1)
+	errChan := make(chan error, 1)
 	go func() {
 		_, err := pub.Endpoint()(context.Background(), struct{}{})
-		err_chan <- err
+		errChan <- err
 
 	}()
 
 	select {
-	case err = <-err_chan:
+	case err = <-errChan:
 		break
 	case <-time.After(100 * time.Millisecond):
 		t.Fatal("timed out waiting for result")
@@ -165,31 +163,31 @@ func TestSuccessfulPublisher(t *testing.T) {
 		testReqEncoder,
 		testResDeliveryDecoder,
 		amqptransport.PublisherBefore(
-			amqptransport.SetCorrelationId(cid),
+			amqptransport.SetCorrelationID(cid),
 		),
 	)
 
 	var res testRes
 	var ok bool
-	res_chan := make(chan interface{}, 1)
-	err_chan := make(chan error, 1)
+	resChan := make(chan interface{}, 1)
+	errChan := make(chan error, 1)
 	go func() {
 		res, err := pub.Endpoint()(context.Background(), mockReq)
 		if err != nil {
-			err_chan <- err
+			errChan <- err
 		} else {
-			res_chan <- res
+			resChan <- res
 		}
 	}()
 
 	select {
-	case response := <-res_chan:
+	case response := <-resChan:
 		res, ok = response.(testRes)
 		if !ok {
 			t.Error("failed to assert endpoint response type")
 		}
 		break
-	case err = <-err_chan:
+	case err = <-errChan:
 		break
 	case <-time.After(100 * time.Millisecond):
 		t.Fatal("timed out waiting for result")

transport/amqp/request_response_func.go

@@ -17,7 +17,7 @@ type RequestFunc func(context.Context, *amqp.Publishing) context.Context
 // subscribers, after invoking the endpoint but prior to publishing a reply.
 type SubscriberResponseFunc func(context.Context,
 	*amqp.Delivery,
-	AMQPChannel,
+	Channel,
 	*amqp.Publishing,
 ) context.Context
 
@@ -44,7 +44,7 @@ func SetPublishKey(publishKey string) RequestFunc {
 	}
 }
 
-// SetPersistentReply sets the delivery mode of a Publishing
+// SetPublishDeliveryMode sets the delivery mode of a Publishing
 // please refer to AMQP delivery mode constants in the AMQP package
 func SetPublishDeliveryMode(dmode uint8) RequestFunc {
 	return func(ctx context.Context, pub *amqp.Publishing,
@@ -98,9 +98,9 @@ func SetContentType(contentType string) RequestFunc {
 	}
 }
 
-// SetCorrelationId returns a RequestFunc that sets the CorrelationId field
+// SetCorrelationID returns a RequestFunc that sets the CorrelationId field
 // of an AMQP Publishing
-func SetCorrelationId(cid string) RequestFunc {
+func SetCorrelationID(cid string) RequestFunc {
 	return func(ctx context.Context, pub *amqp.Publishing,
 	) context.Context {
 		pub.CorrelationId = cid
@@ -115,7 +115,7 @@ func SetCorrelationId(cid string) RequestFunc {
 func SetAckAfterEndpoint(multiple bool) SubscriberResponseFunc {
 	return func(ctx context.Context,
 		deliv *amqp.Delivery,
-		ch AMQPChannel,
+		ch Channel,
 		pub *amqp.Publishing,
 	) context.Context {
 		deliv.Ack(multiple)
@@ -126,62 +126,56 @@ func SetAckAfterEndpoint(multiple bool) SubscriberResponseFunc {
 func getPublishExchange(ctx context.Context) string {
 	if exchange := ctx.Value(ContextKeyExchange); exchange != nil {
 		return exchange.(string)
-	} else {
-		return ""
 	}
+	return ""
 }
 
 func getPublishKey(ctx context.Context) string {
 	if publishKey := ctx.Value(ContextKeyPublishKey); publishKey != nil {
 		return publishKey.(string)
-	} else {
-		return ""
 	}
+	return ""
 }
 
 func getNackSleepDuration(ctx context.Context) time.Duration {
 	if duration := ctx.Value(ContextKeyNackSleepDuration); duration != nil {
 		return duration.(time.Duration)
-	} else {
-		return 0
 	}
+	return 0
 }
 
 func getConsumeAutoAck(ctx context.Context) bool {
 	if autoAck := ctx.Value(ContextKeyAutoAck); autoAck != nil {
 		return autoAck.(bool)
-	} else {
-		return false
 	}
+	return false
 }
 
 func getConsumeArgs(ctx context.Context) amqp.Table {
 	if args := ctx.Value(ContextKeyConsumeArgs); args != nil {
 		return args.(amqp.Table)
-	} else {
-		return nil
 	}
+	return nil
 }
 
 type contextKey int
 
 const (
-	// ContextKeyExchange
-	// It is the value of the reply Exchange in amqp.Publish
+	// ContextKeyExchange is the value of the reply Exchange in
+	// amqp.Publish
 	ContextKeyExchange contextKey = iota
-	// ContextKeyPublishKey
-	// It is the value of the ReplyTo field in amqp.Publish
+	// ContextKeyPublishKey is the value of the ReplyTo field in
+	// amqp.Publish
 	ContextKeyPublishKey
-	// ContextKeyNackSleepDuration
-	// It is the duration to sleep for if the service
-	// Nack and requeues a message.
+	// ContextKeyNackSleepDuration is the duration to sleep for if the
+	// service Nack and requeues a message.
 	// This is to prevent sporadic send-resending of message
 	// when a message is constantly Nack'd and requeued
 	ContextKeyNackSleepDuration
-	// ContextKeyAutoAck
-	// It is the value of autoAck field when calling amqp.Channel.Consume
+	// ContextKeyAutoAck is the value of autoAck field when calling
+	// amqp.Channel.Consume
 	ContextKeyAutoAck
-	// ContextKeyConsumeArgs
-	// It is the value of consumeArgs field when calling amqp.Channel.Consume
+	// ContextKeyConsumeArgs is the value of consumeArgs field when calling
+	// amqp.Channel.Consume
 	ContextKeyConsumeArgs
 )

transport/amqp/subscriber.go

@@ -75,8 +75,8 @@ func SubscriberErrorLogger(logger log.Logger) SubscriberOption {
 
 // ServeDelivery handles AMQP Delivery messages
 // It is strongly recommended to use *amqp.Channel as the
-// AMQPChannel interface implementation
-func (s Subscriber) ServeDelivery(ch AMQPChannel) func(deliv *amqp.Delivery) {
+// Channel interface implementation
+func (s Subscriber) ServeDelivery(ch Channel) func(deliv *amqp.Delivery) {
 	return func(deliv *amqp.Delivery) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -122,7 +122,7 @@ func (s Subscriber) ServeDelivery(ch AMQPChannel) func(deliv *amqp.Delivery) {
 func EncodeJSONResponse(
 	ctx context.Context,
 	deliv *amqp.Delivery,
-	ch AMQPChannel,
+	ch Channel,
 	pub *amqp.Publishing,
 	response interface{},
 ) error {
@@ -155,7 +155,7 @@ func EncodeJSONResponse(
 func EncodeNopResponse(
 	ctx context.Context,
 	deliv *amqp.Delivery,
-	ch AMQPChannel,
+	ch Channel,
 	pub *amqp.Publishing,
 	response interface{},
 ) error {
@@ -167,32 +167,30 @@ func EncodeNopResponse(
 // their replies, and will likely want to pass and check for their own error
 // types.
 type ErrorEncoder func(ctx context.Context,
-	err error, deliv *amqp.Delivery, ch AMQPChannel, pub *amqp.Publishing)
+	err error, deliv *amqp.Delivery, ch Channel, pub *amqp.Publishing)
 
 // DefaultErrorEncoder simply ignores the message. It does not reply
 // nor Ack/Nack the message.
 func DefaultErrorEncoder(ctx context.Context,
-	err error, deliv *amqp.Delivery, ch AMQPChannel, pub *amqp.Publishing) {
-	return
+	err error, deliv *amqp.Delivery, ch Channel, pub *amqp.Publishing) {
 }
 
-// NackErrorEncoder issues a Nack to the delivery with multiple flag set as false
+// SingleNackRequeueErrorEncoder issues a Nack to the delivery with multiple flag set as false
 // and requeue flag set as true. It does not reply the message.
 func SingleNackRequeueErrorEncoder(ctx context.Context,
-	err error, deliv *amqp.Delivery, ch AMQPChannel, pub *amqp.Publishing) {
+	err error, deliv *amqp.Delivery, ch Channel, pub *amqp.Publishing) {
 	deliv.Nack(
 		false, //multiple
 		true,  //requeue
 	)
 	duration := getNackSleepDuration(ctx)
 	time.Sleep(duration)
-	return
 }
 
 // ReplyErrorEncoder serializes the error message as a DefaultErrorResponse
 // JSON and sends the message to the ReplyTo address
 func ReplyErrorEncoder(ctx context.Context,
-	err error, deliv *amqp.Delivery, ch AMQPChannel, pub *amqp.Publishing) {
+	err error, deliv *amqp.Delivery, ch Channel, pub *amqp.Publishing) {
 
 	if pub.CorrelationId == "" {
 		pub.CorrelationId = deliv.CorrelationId
@@ -221,21 +219,23 @@ func ReplyErrorEncoder(ctx context.Context,
 	)
 }
 
-// ReplyErrorEncoder serializes the error message as a DefaultErrorResponse
+// ReplyAndAckErrorEncoder serializes the error message as a DefaultErrorResponse
 // JSON and sends the message to the ReplyTo address then Acks the original
 // message
-func ReplyAndAckErrorEncoder(ctx context.Context, err error, deliv *amqp.Delivery, ch AMQPChannel, pub *amqp.Publishing) {
+func ReplyAndAckErrorEncoder(ctx context.Context, err error, deliv *amqp.Delivery, ch Channel, pub *amqp.Publishing) {
 	ReplyErrorEncoder(ctx, err, deliv, ch, pub)
 	deliv.Ack(false)
 }
 
+// DefaultErrorResponse is the default structure of responses in the event
+// of an error
 type DefaultErrorResponse struct {
 	Error string `json:"err"`
 }
 
-// Channel interface to make testing possible
+// Channel is a channel interface to make testing possible
 // It is highly recommended to use *amqp.Channel as the interface implementation
-type AMQPChannel interface {
+type Channel interface {
 	Publish(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error
 	Consume(queue, consumer string, autoAck, exclusive, noLocal, noWail bool, args amqp.Table) (<-chan amqp.Delivery, error)
 }