AMQP Transport #746

mattfung · 2018-08-07T17:40:27Z

Hello,

I noticed that here had been interest in AMQP transport similiar to NATS (#681) as well as talks of pubsub support (#295) so I thought I might like to share an implementation of a simple AMQP client & server transport.

Comments/Reviews would be appreciated.

peterbourgon · 2018-08-08T07:37:29Z

Cool, this looks reasonable at a first glance. As a next step, please run go vet, golint, and staticcheck and fix the relevant errors. A couple of problems that stood out to me: underscores in variable names, and superfluous newlines in function declarations.

mattfung · 2018-08-10T18:42:11Z

May bad with the underscores.. bad python habits.
The problems should be fixed now, lemme know if anything can be improved

jstoja · 2018-10-16T21:56:37Z

Hey @peterbourgon @mattfung any chance that this moves forward? This would be a great addition along NATS!

jstoja · 2018-10-17T10:34:47Z

@mattfung I had some issues with running the tests on my end. Rebasing solves all the tests, would you be able to do that?

jstoja · 2018-10-17T18:27:21Z

I tried the subscriber par only and it worked great! I seems to me that it could be enhanced when looking at Nats.
With the Nats transporter, you can directly pass a Handler when subscribing, like this

cSub, err := nc.QueueSubscribe("stringsvc.count", "stringsvc", countHandler.ServeMsg(nc))

Which doesn't seem possible without additional code for this implementation since the AMQP client library returns a channel of Deliveries:

func (ch *Channel) Consume(queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args Table) (<-chan Delivery, error)
func (s Subscriber) ServeDelivery(ch Channel) func(deliv *amqp.Delivery)

and the ServeDelivery process message per message.

So you have to create a helper function to listen to the channel and map the ServiceDelivery return to each message. Have you any idea on how to improve this? Or maybe I'm missing something in the implementation...

mattfung · 2018-10-17T18:57:45Z

make a PR to https://github.com/streadway/amqp with channel listener functionality :)

jokes aside, Id be interested on improvements

jstoja · 2018-10-17T19:30:21Z

Haha yes indeed! Let’s think about that in another issue ;) It think this is definitely mergeable! Great job!

…

On 17 Oct 2018, at 20:58, Matthew Fung ***@***.***> wrote: make a PR to https://github.com/streadway/amqp with channel listener functionality :) jokes aside, Id be interested on improvements — You are receiving this because you commented. Reply to this email directly, view it on GitHub, or mute the thread.

jstoja · 2018-11-04T17:16:54Z

@peterbourgon Do you have any feedback on this for merging it? I've used it for a professional project as a demonstration (so with small usage), and it ran several days without any issues.

peterbourgon · 2018-11-04T18:40:43Z

That’s great feedback, thanks. I just need to block off an hour to review. I’ll do that in the next 2 days. Sorry for the delay; I have a lot on my plate these days :/

…

On Sun, Nov 4, 2018 at 09:17 Julien ***@***.***> wrote: @peterbourgon <https://github.com/peterbourgon> Do you have any feedback on this for merging it? I've used it for a professional project as a demonstration (so with small usage), and it ran several days without any issues. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#746 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAG_TUbL68zxAf9rcX0WyAwXrFBm52DYks5uryEIgaJpZM4Vyl8X> .

jstoja · 2018-11-04T18:43:29Z

No worries :) just trying to know if help was needed on that!

…

Sent from my iPhone

On 4 Nov 2018, at 19:40, Peter Bourgon ***@***.***> wrote: That’s great feedback, thanks. I just need to block off an hour to review. I’ll do that in the next 2 days. Sorry for the delay; I have a lot on my plate these days :/ On Sun, Nov 4, 2018 at 09:17 Julien ***@***.***> wrote: > @peterbourgon <https://github.com/peterbourgon> Do you have any feedback > on this for merging it? I've used it for a professional project as a > demonstration (so with small usage), and it ran several days without any > issues. > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <#746 (comment)>, or mute > the thread > <https://github.com/notifications/unsubscribe-auth/AAG_TUbL68zxAf9rcX0WyAwXrFBm52DYks5uryEIgaJpZM4Vyl8X> > . > — You are receiving this because you commented. Reply to this email directly, view it on GitHub, or mute the thread.

peterbourgon

A couple of relatively minor design questions, and lots of style nits, most notably that I want proper punctuation for all block comments :) But in general this looks good, with those changes and some explanation, I'm see no problem to merge. Thank you for the great contribution!

peterbourgon · 2018-11-05T22:46:01Z

transport/amqp/doc.go

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


Suggested change

// Package amqp implements a go-kit transport for AMQP

// Package amqp implements an AMQP transport.

peterbourgon · 2018-11-05T22:46:27Z

transport/amqp/encode-decode.go

+type DecodeRequestFunc func(context.Context, *amqp.Delivery) (request interface{}, err error)
+
+// EncodeRequestFunc encodes the passed request object into
+// an AMQP channel for publishing. It is designed to be used in AMQP Publishers


Suggested change

// an AMQP channel for publishing. It is designed to be used in AMQP Publishers

// an AMQP channel for publishing. It is designed to be used in AMQP Publishers.

peterbourgon · 2018-11-05T22:46:34Z

transport/amqp/encode-decode.go

+type EncodeRequestFunc func(context.Context, *amqp.Publishing, interface{}) error
+
+// EncodeResponseFunc encodes the passed reponse object to
+// an AMQP channel for publishing. It is designed to be used in AMQP Subscribers


Suggested change

// an AMQP channel for publishing. It is designed to be used in AMQP Subscribers

// an AMQP channel for publishing. It is designed to be used in AMQP Subscribers.

peterbourgon · 2018-11-05T22:47:00Z

transport/amqp/encode-decode.go

+	*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


Suggested change

// an AMQP Delivery object. It is designed to be used in AMQP Publishers

// an AMQP Delivery object. It is designed to be used in AMQP Publishers.

peterbourgon · 2018-11-05T22:47:24Z

transport/amqp/publisher.go

+	"github.com/streadway/amqp"
+)
+
+// Publisher wraps AMQP channel and queue and provides a method that


Suggested change

// Publisher wraps AMQP channel and queue and provides a method that

// Publisher wraps an AMQP channel and queue, and provides a method that

peterbourgon · 2018-11-05T22:51:48Z

transport/amqp/request_response_func.go

+type PublisherResponseFunc func(context.Context, *amqp.Delivery) context.Context
+
+// SetPublishExchange returns a RequestFunc that sets the Exchange field
+// of AMQP Publish call


Please add ending punctuation (.) to this and all other header comment blocks in the PR.

peterbourgon · 2018-11-05T22:53:36Z

transport/amqp/subscriber.go

+// 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 Channel, pub *amqp.Publishing) {


Please either put the function definition on a single line, or use one parameter per line, e.g.

func Foo( ctx context.Context, err error, deliv *amqp.Delivery, ) {

peterbourgon · 2018-11-05T22:56:18Z

transport/amqp/publisher.go

+		}
+
+		deliv, err := p.publishAndConsumeFirstMatchingResponse(ctx, &pub)
+


Suggested change

peterbourgon · 2018-11-05T22:57:10Z

transport/amqp/subscriber.go

+		}
+
+		request, err := s.dec(ctx, deliv)
+


Suggested change

peterbourgon · 2018-11-05T22:58:39Z

transport/amqp/subscriber.go

+			ctx = f(ctx, deliv, ch, &pub)
+		}
+
+		if err := s.enc(ctx, deliv, ch, &pub, response); err != nil {


You're combining "encoding the response to the delivery" with "sending the delivery over AMQP" into a single operation, but my intuition is these should be separate steps, with separate failure modes. Is there a technical reason for that that I'm overlooking?

No good reason other than "cuz the nats transport did it"

I'll separate it to two functions

mattfung · 2018-11-06T00:17:14Z

Sounds good, I'll get them fixed within a day's time.
thx again for putting in the time to do this

mattfung · 2018-11-06T20:12:05Z

Committed the changes. I'm guessing the tests failed because I didn't rebase the code with the latest master branch?

Anyhow, let me know if anything needs to be changed.

jstoja · 2018-11-07T07:25:46Z

Tried without (and also with) the rebase locally and I didn't manage to make them fail... Maybe Peter will have an idea.

peterbourgon · 2018-11-07T18:58:49Z

Yes, this seems fine; can you rebase on master?

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

peterbourgon · 2018-11-08T02:13:33Z

Dang nice

tlogik · 2018-11-15T17:51:20Z

@mattfung I am new to gokit and struggling a bit getting AMQP up and working.
Besides the test do you have a sample app or something that shows how to use the AMQP setup.
I am starting prototyping moving some of out NET services to GO and we rely heavily on RabbitMQ.
Any help is very much appreciated.

mattfung · 2018-11-15T23:20:00Z

@tlogik for now, heres an implementation of stringsvc4 pub & sub
https://github.com/cloudpbx/kit/tree/amqp/examples/stringsvc4/amqp

Hope this helps.

jstoja · 2018-11-16T09:17:51Z

Awesome! Would be great to have it in this repository too, no?

mattfung · 2018-11-16T16:14:07Z

Thats the plan, my only reservation currently is that the commit makes some breaking changes with the existing stringsvc4 implementation. When I resolve that, I'll create a new merge request.

tlogik · 2018-11-17T06:03:18Z

Hope this helps.

It helped a lot. Much appreciated.

tlogik · 2018-11-18T19:48:10Z

Hi again @mattfung
I am using your example as a template, trying to get the behaviour i need.
Questions arise as i work with the implementation.
I find that in the Subscriber.go, func ServeDelivery, the channel for consuming is also used for publishing.
That is somewhat ok, except for at common requirement for subscribing from one Vhost and publishing to another Vhost.
AFAIK you need to create a new connection to each Vhost, create individual channels etc.

i do not seem to be able to setup the internal publisher, in the subscriber.go, for an alternate channel - can you confirm that is not possible in the current implementation?

mattfung · 2018-11-20T16:22:18Z

@tlogik correct, current implementation does not allow for a subscriber to subscribe to one channel (& consequently, Vhost) and publish to another. I am curious on how other frameworks handle this problem if you happen to know of any.

Currently, publishers and subscribers are (for simplicity's sake) tied to one channel. That can change if it is a significant limitation.

An alternative approach would be to have a subscriber for vhost A trigger a publisher for vhost B.

tlogik · 2018-11-20T18:36:43Z

@mattfung currently we use individual instances of subscriber and publisher with individual connstrings.
I was looking into using go-kit as the toolbox of choice to create a standard to metrics, health, service registration etc, but it is a little hard to extend the transport layer.
i am currently working on a service that needs to batch process messages consumed from MQ and then inserted into SOLR. It is a lot faster to insert in SOLR via batching than individual requests.
This will require a custom amqp implementation to support this.

jstoja · 2018-11-23T22:25:56Z

@tlogik @mattfung From steadway/amqp's package documentation:

It's advisable to use separate connections for Channel.Publish and Channel.Consume so not to have TCP pushback on publishing affect the ability to consume messages, so this parameter is here mostly for completeness.

source: https://godoc.org/github.com/streadway/amqp#Channel.Consume

So to me the implementation of @mattfung respect perfectly that I like that it's not possible to shoot myself in the foot using this implementation with go-kit.

Moreover, the vhosts are part of the connection string, you cannot have 1 channel listening on several vhost, since an amqp channel is some kind of wrapper adding multiplexing to the parent amqp connection which is itself tied to one vhost.

@tlogik I guess that if you want to implement that, you can easily consume messages from one channel following @mattfung examples and implement some kind of buffering inside your program in addition of using prefetching (https://godoc.org/github.com/streadway/amqp#Channel.Qos).

freemanoid · 2019-01-16T19:33:10Z

@mattfung thanks for the amqp implementation, I've adopted it and it works perfectly! I see that consuming function automatically publishes some "response". It's not needed for me and I don't see any possibility to turn it off. Can you explain why did you added it? Maybe I'm using amqp wrong.

peterbourgon · 2019-01-17T21:48:44Z

If you're not publishing responses, you're not doing RPC.

freemanoid · 2019-01-18T08:04:24Z

If you're not publishing responses, you're not doing RPC.

Do you mean this implementation is not about AMQP, but RPC over AMQP?

mattfung · 2019-01-18T16:16:36Z

@freemanoid Good point, one way AMQP communication is not uncommon. Currently the closest thing to a "no-op" is to use the NopEncoder such that an empty response is sent back.

There are two solutions to implement true "no-op" subscribers, one is to make the "encoder" both encode and send, and use EncodeNopResponse (though Im not sure if thats a good idea )

The other option is to add an options/ change the parameter to NewSubscriber

peterbourgon · 2019-01-18T16:33:33Z

@freemanoid

Do you mean this implementation is not about AMQP, but RPC over AMQP?

Yes. Go kit is concerned exclusively with RPC.

Non-goals

Supporting messaging patterns other than RPC (for now) — e.g. MPI, pub/sub, CQRS, etc.

@mattfung

The other option is to add an options/ change the parameter to NewSubscriber

For the record, if this is easy, I'd happily take a PR.

madshov · 2019-03-02T20:29:58Z

@mattfung Do you plan any work around adding an option/change parameter to NewSubscriber to no-op subscribers?

mattfung · 2019-03-19T01:42:44Z

@madshov sorry for the slow response, PR is posted #850

mattfung force-pushed the amqp branch from 53c0fb5 to 9cff7f5 Compare October 17, 2018 15:09

peterbourgon requested changes Nov 5, 2018

View reviewed changes

peterbourgon approved these changes Nov 7, 2018

View reviewed changes

mattfung force-pushed the amqp branch from 1e4d236 to d65c17b Compare November 7, 2018 19:59

mattfung force-pushed the amqp branch from d65c17b to 40a1cb2 Compare November 7, 2018 20:01

peterbourgon merged commit 12210fb into go-kit:master Nov 8, 2018

peterbourgon mentioned this pull request Feb 28, 2019

Async (non-RPC) AMQP transport #841

Closed

mattfung mentioned this pull request Mar 19, 2019

adds NopResponse to AMQP transport #850

Merged

kallydev mentioned this pull request Aug 19, 2020

Lack of AMQP documentation #920

Closed

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

	// Package amqp implements a go-kit transport for AMQP
	// Package amqp implements an AMQP transport.

	// an AMQP channel for publishing. It is designed to be used in AMQP Publishers
	// an AMQP channel for publishing. It is designed to be used in AMQP Publishers.

	// an AMQP Delivery object. It is designed to be used in AMQP Publishers
	// an AMQP Delivery object. It is designed to be used in AMQP Publishers.

	// Publisher wraps AMQP channel and queue and provides a method that
	// Publisher wraps an AMQP channel and queue, and provides a method that

		}

		deliv, err := p.publishAndConsumeFirstMatchingResponse(ctx, &pub)

AMQP Transport #746

AMQP Transport #746

Conversation

mattfung commented Aug 7, 2018

peterbourgon commented Aug 8, 2018

mattfung commented Aug 10, 2018

jstoja commented Oct 16, 2018

jstoja commented Oct 17, 2018 • edited Loading

jstoja commented Oct 17, 2018

mattfung commented Oct 17, 2018

jstoja commented Oct 17, 2018 via email

jstoja commented Nov 4, 2018

peterbourgon commented Nov 4, 2018 via email

jstoja commented Nov 4, 2018 via email

peterbourgon left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattfung commented Nov 6, 2018

mattfung commented Nov 6, 2018 • edited Loading

jstoja commented Nov 7, 2018

peterbourgon commented Nov 7, 2018

peterbourgon commented Nov 8, 2018

tlogik commented Nov 15, 2018

mattfung commented Nov 15, 2018

jstoja commented Nov 16, 2018

mattfung commented Nov 16, 2018 • edited Loading

tlogik commented Nov 17, 2018

tlogik commented Nov 18, 2018

mattfung commented Nov 20, 2018

tlogik commented Nov 20, 2018

jstoja commented Nov 23, 2018

freemanoid commented Jan 16, 2019

peterbourgon commented Jan 17, 2019

freemanoid commented Jan 18, 2019

mattfung commented Jan 18, 2019

peterbourgon commented Jan 18, 2019

madshov commented Mar 2, 2019

mattfung commented Mar 19, 2019

jstoja commented Oct 17, 2018 •

edited

Loading

mattfung commented Nov 6, 2018 •

edited

Loading

mattfung commented Nov 16, 2018 •

edited

Loading