Add AMQP transport (#746) (#746)

amqp transport publisher

amqp transport tests

lint fixes for amqp transport

fixed formatting and punctuation

zzz

default ContentType is null, increased max length of correlationId to 255

refractored subscriber EncodeResponseFunc into encode func, and send func

zzz

transport/amqp/doc.go

@@ -0,0 +1,2 @@
+// Package amqp implements an AMQP transport.
+package amqp

transport/amqp/encode-decode.go

@@ -0,0 +1,22 @@
+package amqp
+
+import (
+	"context"
+	"github.com/streadway/amqp"
+)
+
+// DecodeRequestFunc extracts a user-domain request object from
+// an AMQP Delivery object. It is designed to be used in AMQP Subscribers.
+type DecodeRequestFunc func(context.Context, *amqp.Delivery) (request interface{}, err error)
+
+// EncodeRequestFunc encodes the passed request object into
+// an AMQP Publishing object. It is designed to be used in AMQP Publishers.
+type EncodeRequestFunc func(context.Context, *amqp.Publishing, interface{}) error
+
+// EncodeResponseFunc encodes the passed reponse object to
+// an AMQP Publishing object. It is designed to be used in AMQP Subscribers.
+type EncodeResponseFunc func(context.Context, *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.
+type DecodeResponseFunc func(context.Context, *amqp.Delivery) (response interface{}, err error)

transport/amqp/publisher.go

@@ -0,0 +1,152 @@
+package amqp
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-kit/kit/endpoint"
+	"github.com/streadway/amqp"
+)
+
+// The golang AMQP implementation requires the []byte representation of
+// correlation id strings to have a maximum length of 255 bytes.
+const maxCorrelationIdLength = 255
+
+// Publisher wraps an AMQP channel and queue, and provides a method that
+// implements endpoint.Endpoint.
+type Publisher struct {
+	ch      Channel
+	q       *amqp.Queue
+	enc     EncodeRequestFunc
+	dec     DecodeResponseFunc
+	before  []RequestFunc
+	after   []PublisherResponseFunc
+	timeout time.Duration
+}
+
+// NewPublisher constructs a usable Publisher for a single remote method.
+func NewPublisher(
+	ch Channel,
+	q *amqp.Queue,
+	enc EncodeRequestFunc,
+	dec DecodeResponseFunc,
+	options ...PublisherOption,
+) *Publisher {
+	p := &Publisher{
+		ch:      ch,
+		q:       q,
+		enc:     enc,
+		dec:     dec,
+		timeout: 10 * time.Second,
+	}
+	for _, option := range options {
+		option(p)
+	}
+	return p
+}
+
+// PublisherOption sets an optional parameter for clients.
+type PublisherOption func(*Publisher)
+
+// PublisherBefore sets the RequestFuncs that are applied to the outgoing AMQP
+// request before it's invoked.
+func PublisherBefore(before ...RequestFunc) PublisherOption {
+	return func(p *Publisher) { p.before = append(p.before, before...) }
+}
+
+// PublisherAfter sets the ClientResponseFuncs applied to the incoming AMQP
+// request prior to it being decoded. This is useful for obtaining anything off
+// of the response and adding onto the context prior to decoding.
+func PublisherAfter(after ...PublisherResponseFunc) PublisherOption {
+	return func(p *Publisher) { p.after = append(p.after, after...) }
+}
+
+// PublisherTimeout sets the available timeout for an AMQP request.
+func PublisherTimeout(timeout time.Duration) PublisherOption {
+	return func(p *Publisher) { p.timeout = timeout }
+}
+
+// Endpoint returns a usable endpoint that invokes the remote endpoint.
+func (p Publisher) Endpoint() endpoint.Endpoint {
+	return func(ctx context.Context, request interface{}) (interface{}, error) {
+		ctx, cancel := context.WithTimeout(ctx, p.timeout)
+		defer cancel()
+
+		pub := amqp.Publishing{
+			ReplyTo:       p.q.Name,
+			CorrelationId: randomString(randInt(5, maxCorrelationIdLength)),
+		}
+
+		if err := p.enc(ctx, &pub, request); err != nil {
+			return nil, err
+		}
+
+		for _, f := range p.before {
+			ctx = f(ctx, &pub)
+		}
+
+		deliv, err := p.publishAndConsumeFirstMatchingResponse(ctx, &pub)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, f := range p.after {
+			ctx = f(ctx, deliv)
+		}
+		response, err := p.dec(ctx, deliv)
+		if err != nil {
+			return nil, err
+		}
+
+		return response, nil
+	}
+}
+
+// publishAndConsumeFirstMatchingResponse publishes the specified Publishing
+// and returns the first Delivery object with the matching correlationId.
+// If the context times out while waiting for a reply, an error will be returned.
+func (p Publisher) publishAndConsumeFirstMatchingResponse(
+	ctx context.Context,
+	pub *amqp.Publishing,
+) (*amqp.Delivery, error) {
+	err := p.ch.Publish(
+		getPublishExchange(ctx),
+		getPublishKey(ctx),
+		false, //mandatory
+		false, //immediate
+		*pub,
+	)
+	if err != nil {
+		return nil, err
+	}
+	autoAck := getConsumeAutoAck(ctx)
+
+	msg, err := p.ch.Consume(
+		p.q.Name,
+		"", //consumer
+		autoAck,
+		false, //exclusive
+		false, //noLocal
+		false, //noWait
+		getConsumeArgs(ctx),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	for {
+		select {
+		case d := <-msg:
+			if d.CorrelationId == pub.CorrelationId {
+				if !autoAck {
+					d.Ack(false) //multiple
+				}
+				return &d, nil
+			}
+
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+
+}