NATS transport #680

kirooha · 2018-03-15T10:55:40Z

The PR contains transport for NATS (messaging system, https://nats.io)
I saw a request with the similar functional #623
Since the PR is still open, I want to suggest my PR.

peterbourgon · 2018-03-15T22:04:53Z

Oh, cool.. ! I would want to also see Publisher type (corresponding to Client), can that be added?

kirooha · 2018-03-15T22:22:09Z

yes, sure! I am adding

MrEhbr · 2018-03-16T07:28:21Z

examples/stringsvc4/main.go

+	"flag"
+
+	"github.com/go-kit/kit/endpoint"
+	natstransport "github.com/kirooha/kit/transport/nats"


github.com/kirooha/kit/transport/nats - maybe need to change to github.com/go-kit/kit/transport/nats

nussjustin · 2018-03-19T17:14:34Z

transport/nats/publisher.go

+		}
+
+		if !p.async {
+			resp, err := p.publisher.Request(p.subject, msg.Data, p.timeout)


Since you already have a context you can use RequestWithContext instead and remove the timeout field.

nussjustin · 2018-03-19T17:16:11Z

transport/nats/publisher.go

+
+		}
+
+		if p.reply == "" {


You can just use PublishMsg and than remove the rest of the method.

nussjustin · 2018-03-19T17:18:22Z

transport/nats/encode_decode.go

+type DecodeRequestFunc func(context.Context, *nats.Msg) (request interface{}, err error)
+
+
+// EncodeRequestFunc encodes the passed request object into the HTTP request


This comment still talks about HTTP

oh, great! thanks a lot! I didn't hope to so fast review) Actually, it was a draft and I supposed to modify comments and test code. One more time, thanks a lot! I will fix your remarks as soon as possible.

nussjustin · 2018-03-19T17:25:37Z

transport/nats/request_response_funcs.go

+// subscribers, after invoking the endpoint but prior to publishing a reply.
+type SubscriberResponseFunc func(context.Context, *nats.Conn) context.Context
+
+// ClientResponseFunc may take information from an HTTP request and make the


nussjustin · 2018-03-19T17:26:47Z

transport/nats/subscriber.go

+// ServeMsg provides nats.MsgHandler.
+func (s Subscriber) ServeMsg(nc *nats.Conn) func(msg *nats.Msg) {
+	return func(msg *nats.Msg) {
+		ctx := context.Background()


The context is never cancelled

nussjustin · 2018-03-19T17:36:24Z

transport/nats/publisher.go

+			ctx = f(ctx, &msg)
+		}
+
+		if !p.async {


The endpoint tries to handle both simple publish operations and requests which feels kinda odd. Also the !async case ignores the p.reply topic.

I suggest to remove the async case and the custom reply topic for know and only do a simple Request(WithContext). We can always add support for messages without reply later through options or a different type.

nussjustin · 2018-03-19T17:38:08Z

transport/nats/publisher.go

+		reply:     reply,
+		enc:     enc,
+		dec:     dec,
+		before:  []RequestFunc{},


No need to initialize this, Go already handles nil correct for all the cases needed.

nussjustin · 2018-03-19T17:39:31Z

transport/nats/subscriber.go

+// ServeMsg provides nats.MsgHandler.
+func (s Subscriber) ServeMsg(nc *nats.Conn) func(msg *nats.Msg) {
+	return func(msg *nats.Msg) {
+		ctx := context.Background()


ctx should be cancelled when the handler returns

nussjustin · 2018-03-19T17:43:48Z

transport/nats/publisher.go

+				return nil, err
+			}
+
+			return nil, nil


The funcs in p.after should probably still be called here and below with a nil *nats.Msg

kirooha · 2018-03-22T12:41:33Z

@nussjustin all remarks are fixed

nussjustin

1 small comment, but other than that the API looks good to me.

This needs some tests though ;-)

For testing publishers and subscribers you can start real NATS servers via the gnatsd/server package (see Server.Start and Server.ReadyForConnections).

nussjustin · 2018-03-25T10:58:25Z

transport/nats/request_response_funcs.go

+// SubscriberRequestFunc may take information from a publisher request and put it into a
+// request context. In Subscribers, RequestFuncs are executed prior to invoking the
+// endpoint.
+type SubscriberRequestFunc func(context.Context, *nats.Msg) context.Context


Sorry, you were right. This should be just RequestFunc. I forgot that this is used for both Publishers and Subscribers. That's what I get for not reviewing my comments before requesting changes...

kirooha · 2018-04-10T21:27:57Z

there's some weird behavior of ci in this pr. I did 3 force push without any changes (amend commit that added blank lines). 2 builds were failed and the latest is succeeded. I mean ci/circleci and continuous-integration/travis-ci/pr

kirooha · 2018-04-11T10:34:57Z

@nussjustin tests have been added

nussjustin · 2018-04-14T16:23:36Z

transport/nats/publisher_test.go

+	"strings"
+
+	"github.com/nats-io/go-nats"
+	natstransport "github.com/kirooha/kit/transport/nats"


Wrong import path

nussjustin · 2018-04-14T16:24:24Z

transport/nats/subscriber_test.go

+	"github.com/nats-io/go-nats"
+	"github.com/nats-io/gnatsd/server"
+
+	natstransport "github.com/kirooha/kit/transport/nats"


kirooha · 2018-04-15T14:53:02Z

@nussjustin fixed wrong paths, squashed commits to the only one with a more detailed comment. do you have any notes?

nussjustin · 2018-04-15T15:46:18Z

Could you please add a test for PublisherTimeout? From what I see it's the only publisher option that's not tested right now (as reported by go tool cover).

Otherwise LGTM so far.

kirooha · 2018-04-15T20:14:34Z

done

nussjustin · 2018-04-16T15:24:22Z

transport/nats/publisher_test.go

+	defer nc.Close()
+
+	sub, err := nc.QueueSubscribe("natstransport.test", "natstransport", func(msg *nats.Msg) {
+		ch := make(chan struct{})


You're leaking the goroutine here. Even if this is just a test it shouldn't leak goroutines. Just create the channel before the QueueSubscribe and close it after the test, like this:

ch := make(chan struct{}) defer close(ch) sub, err := nc.QueueSubscribe("natstransport.test", "natstransport", func(msg *nats.Msg) { <-ch })

NATS is an open-source, cloud-native messaging system (https://www.nats.io). The functional provides API that lets one works with NATS in a similar way as HTTP. - nats.MsgHandler could be used in queue or simple subscriber - Sync publisher

kirooha · 2018-04-16T16:27:14Z

thank you! awesome note, fixed

nussjustin

I don't see anything obviously wrong / bad right now sooo... passing on the review torch to @peterbourgon and co.

peterbourgon · 2018-04-16T17:18:28Z

Great, thank you. Give me a week or so; I'm on a work trip and have a vacation immediately afterwards.

peterbourgon · 2018-04-27T17:32:41Z

Yeah I really don't have any nits to pick here, this LGTM! Thanks very much for the hard work!

kirooha changed the title ~~Transport for nats~~ NATS transport Mar 15, 2018

MrEhbr reviewed Mar 16, 2018

View reviewed changes

vetcher mentioned this pull request Mar 16, 2018

Question: AMQP transport implementation? #681

Closed

nussjustin suggested changes Mar 19, 2018

View reviewed changes

nussjustin suggested changes Mar 25, 2018

View reviewed changes

nussjustin reviewed Apr 14, 2018

View reviewed changes

nussjustin suggested changes Apr 16, 2018

View reviewed changes

Add NATS transport

db58b3a

NATS is an open-source, cloud-native messaging system (https://www.nats.io). The functional provides API that lets one works with NATS in a similar way as HTTP. - nats.MsgHandler could be used in queue or simple subscriber - Sync publisher

nussjustin approved these changes Apr 16, 2018

View reviewed changes

peterbourgon merged commit 62e6b81 into go-kit:master Apr 27, 2018

This was referenced Apr 27, 2018

NATS transport for go-kit #623

Closed

Fix race condition in NATS transport tests #705

Merged

dzakaammar mentioned this pull request Aug 9, 2018

option to generate NATS transport kujtimiihoxha/kit#14

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

NATS transport #680

NATS transport #680

kirooha commented Mar 15, 2018

peterbourgon commented Mar 15, 2018

kirooha commented Mar 15, 2018

MrEhbr Mar 16, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

kirooha Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

nussjustin Mar 19, 2018

kirooha commented Mar 22, 2018

nussjustin left a comment

nussjustin Mar 25, 2018

kirooha commented Apr 10, 2018

kirooha commented Apr 11, 2018

nussjustin Apr 14, 2018

nussjustin Apr 14, 2018

kirooha commented Apr 15, 2018

nussjustin commented Apr 15, 2018

kirooha commented Apr 15, 2018

nussjustin Apr 16, 2018

kirooha commented Apr 16, 2018

nussjustin left a comment

peterbourgon commented Apr 16, 2018

peterbourgon commented Apr 27, 2018

		type DecodeRequestFunc func(context.Context, *nats.Msg) (request interface{}, err error)


		// EncodeRequestFunc encodes the passed request object into the HTTP request

NATS transport #680

NATS transport #680

Conversation

kirooha commented Mar 15, 2018

peterbourgon commented Mar 15, 2018

kirooha commented Mar 15, 2018

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

kirooha commented Mar 22, 2018

nussjustin left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirooha commented Apr 10, 2018

kirooha commented Apr 11, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kirooha commented Apr 15, 2018

nussjustin commented Apr 15, 2018

kirooha commented Apr 15, 2018

Choose a reason for hiding this comment

kirooha commented Apr 16, 2018

nussjustin left a comment

Choose a reason for hiding this comment

peterbourgon commented Apr 16, 2018

peterbourgon commented Apr 27, 2018