Commit 53b763c5d945b8d57b43a549ec13e337e22c4287 - golang-github-go-kit-kit

Add amqp.Delivery to the request func in order to have the ability to change incoming message before decode function in case of zipkin context (#802) 0xdev authored 5 years ago Peter Bourgon committed 5 years ago

5 changed file(s) with 36 addition(s) and 35 deletion(s). Raw diff Collapse all Expand all

-23

~~transport/amqp/encode-decode.go~~ less more

0		package amqp
1
2		import (
3		"context"
4
5		"github.com/streadway/amqp"
6		)
7
8		// DecodeRequestFunc extracts a user-domain request object from
9		// an AMQP Delivery object. It is designed to be used in AMQP Subscribers.
10		type DecodeRequestFunc func(context.Context, *amqp.Delivery) (request interface{}, err error)
11
12		// EncodeRequestFunc encodes the passed request object into
13		// an AMQP Publishing object. It is designed to be used in AMQP Publishers.
14		type EncodeRequestFunc func(context.Context, *amqp.Publishing, interface{}) error
15
16		// EncodeResponseFunc encodes the passed response object to
17		// an AMQP Publishing object. It is designed to be used in AMQP Subscribers.
18		type EncodeResponseFunc func(context.Context, *amqp.Publishing, interface{}) error
19
20		// DecodeResponseFunc extracts a user-domain response object from
21		// an AMQP Delivery object. It is designed to be used in AMQP Publishers.
22		type DecodeResponseFunc func(context.Context, *amqp.Delivery) (response interface{}, err error)

+23

-0

transport/amqp/encode_decode.go less more

	0	package amqp
	1
	2	import (
	3	"context"
	4
	5	"github.com/streadway/amqp"
	6	)
	7
	8	// DecodeRequestFunc extracts a user-domain request object from
	9	// an AMQP Delivery object. It is designed to be used in AMQP Subscribers.
	10	type DecodeRequestFunc func(context.Context, *amqp.Delivery) (request interface{}, err error)
	11
	12	// EncodeRequestFunc encodes the passed request object into
	13	// an AMQP Publishing object. It is designed to be used in AMQP Publishers.
	14	type EncodeRequestFunc func(context.Context, *amqp.Publishing, interface{}) error
	15
	16	// EncodeResponseFunc encodes the passed response object to
	17	// an AMQP Publishing object. It is designed to be used in AMQP Subscribers.
	18	type EncodeResponseFunc func(context.Context, *amqp.Publishing, interface{}) error
	19
	20	// DecodeResponseFunc extracts a user-domain response object from
	21	// an AMQP Delivery object. It is designed to be used in AMQP Publishers.
	22	type DecodeResponseFunc func(context.Context, *amqp.Delivery) (response interface{}, err error)

-1

transport/amqp/publisher.go less more

81	81	}
82	82
83	83	for _, f := range p.before {
84		ctx = f(ctx, &pub)
	84	// Affect only amqp.Publishing
	85	ctx = f(ctx, &pub, nil)
85	86	}
86	87
87	88	deliv, err := p.publishAndConsumeFirstMatchingResponse(ctx, &pub)

+10

-10

transport/amqp/request_response_func.go less more

9	9	// RequestFunc may take information from a publisher request and put it into a
10	10	// request context. In Subscribers, RequestFuncs are executed prior to invoking
11	11	// the endpoint.
12		type RequestFunc func(context.Context, *amqp.Publishing) context.Context
	12	type RequestFunc func(context.Context, amqp.Publishing, amqp.Delivery) context.Context
13	13
14	14	// SubscriberResponseFunc may take information from a request context and use it to
15	15	// manipulate a Publisher. SubscriberResponseFuncs are only executed in

28	28	// SetPublishExchange returns a RequestFunc that sets the Exchange field
29	29	// of an AMQP Publish call.
30	30	func SetPublishExchange(publishExchange string) RequestFunc {
31		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	31	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
32	32	return context.WithValue(ctx, ContextKeyExchange, publishExchange)
33	33	}
34	34	}

36	36	// SetPublishKey returns a RequestFunc that sets the Key field
37	37	// of an AMQP Publish call.
38	38	func SetPublishKey(publishKey string) RequestFunc {
39		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	39	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
40	40	return context.WithValue(ctx, ContextKeyPublishKey, publishKey)
41	41	}
42	42	}

44	44	// SetPublishDeliveryMode sets the delivery mode of a Publishing.
45	45	// Please refer to AMQP delivery mode constants in the AMQP package.
46	46	func SetPublishDeliveryMode(dmode uint8) RequestFunc {
47		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	47	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
48	48	pub.DeliveryMode = dmode
49	49	return ctx
50	50	}

56	56	// One example is the SingleNackRequeueErrorEncoder.
57	57	// It is designed to be used by Subscribers.
58	58	func SetNackSleepDuration(duration time.Duration) RequestFunc {
59		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	59	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
60	60	return context.WithValue(ctx, ContextKeyNackSleepDuration, duration)
61	61	}
62	62	}

67	67	// a matching correlationId.
68	68	// It is designed to be used by Publishers.
69	69	func SetConsumeAutoAck(autoAck bool) RequestFunc {
70		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	70	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
71	71	return context.WithValue(ctx, ContextKeyAutoAck, autoAck)
72	72	}
73	73	}

76	76	// function.
77	77	// It is designed to be used by Publishers.
78	78	func SetConsumeArgs(args amqp.Table) RequestFunc {
79		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	79	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
80	80	return context.WithValue(ctx, ContextKeyConsumeArgs, args)
81	81	}
82	82	}

84	84	// SetContentType returns a RequestFunc that sets the ContentType field of
85	85	// an AMQP Publishing.
86	86	func SetContentType(contentType string) RequestFunc {
87		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	87	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
88	88	pub.ContentType = contentType
89	89	return ctx
90	90	}

93	93	// SetContentEncoding returns a RequestFunc that sets the ContentEncoding field
94	94	// of an AMQP Publishing.
95	95	func SetContentEncoding(contentEncoding string) RequestFunc {
96		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	96	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
97	97	pub.ContentEncoding = contentEncoding
98	98	return ctx
99	99	}

102	102	// SetCorrelationID returns a RequestFunc that sets the CorrelationId field
103	103	// of an AMQP Publishing.
104	104	func SetCorrelationID(cid string) RequestFunc {
105		return func(ctx context.Context, pub *amqp.Publishing) context.Context {
	105	return func(ctx context.Context, pub amqp.Publishing, _ amqp.Delivery) context.Context {
106	106	pub.CorrelationId = cid
107	107	return ctx
108	108	}

-1

transport/amqp/subscriber.go less more

83	83	pub := amqp.Publishing{}
84	84
85	85	for _, f := range s.before {
86		ctx = f(ctx, &pub)
	86	ctx = f(ctx, &pub, deliv)
87	87	}
88	88
89	89	request, err := s.dec(ctx, deliv)