Implement nats_jetstream using new jetstream package #1083

stephen-totty-hpe · 2024-07-29T21:52:35Z

Implement nats_jetstream using new jetstream package.

Because many of the options were incompatible, I was forced to create new constructors for Consumer, Sender, and Protocol.

Also, *nats.Msg does not implement jetstream.Msg. Because of this, there is a MsgHandlerV2.

Integration tests are modified to test the new code.

@embano1, let me know if you would rather create a V2 implementation elsewhere

embano1

I'm still torn between adding a new V2 API vs introducing a breaking change and remove the old code.

embano1 · 2024-08-03T07:13:41Z

protocol/nats_jetstream/v2/options.go

 )

 var ErrInvalidQueueName = errors.New("invalid queue name for QueueSubscriber")
+var ErrNoConsumerConfig = errors.New("no consumer config was given")


Question: do we need all those typed errors? They're now part of the API. Also, in those cases a more simple pattern is to use an error struct, for example https://github.com/aws-controllers-k8s/eventbridge-controller/blob/67a9375a6ac7f95f1a3eb8b1b49f329755305c24/pkg/resource/rule/hooks.go#L29-L43

Actually, with errors.As, using simple exported errors works rather than needing to do something more complicated unless you want additional methods.

embano1 · 2024-08-03T07:16:44Z

protocol/nats_jetstream/v2/protocol.go

@@ -67,6 +86,37 @@ func NewProtocolFromConn(conn *nats.Conn, stream, sendSubject, receiveSubject st
 	return p, nil
 }

+func NewProtocolFromConnV2(conn *nats.Conn, stream, sendSubject string, jsOpts []jetstream.JetStreamOpt, opts ...ProtocolOption) (*Protocol, error) {


@evankanderson what do you think about the V2 change? We have to trade backwards-compatibility with maintenance and simple APIs.

@stephen-totty-hpe
If we go down this route, we need to mark the other method(s) deprecated so linters can catch it. Also, should we use this as opportunity to also add context.Context as first parameter to the API?

BTW, I have other PRs to modify behavior in the existing implementation that are not needed using the implementation I have here with the jetstream package. So however it is done, the new jetstream library would be a welcome change. For instance, the new jetstream consumers are pull consumers. Also, an array of filter subjects are exposed. An there are error handlers that can be passed through the "config" objects.

embano1 · 2024-08-03T07:17:33Z

protocol/nats_jetstream/v2/protocol.go

+	var err error
+	var js jetstream.JetStream


Do we need to define those? Is it because of the err in L96 and L100?

It is because of the combined "if"+"assignment" on line 100. The js would be undefined later on line 105.

embano1 · 2024-08-03T07:18:22Z

protocol/nats_jetstream/v2/protocol.go

+	if js, err = jetstream.New(conn, jsOpts...); err != nil {
+		return nil, err
+	}
+	ctx := context.Background()


Ah, see earlier comment on ctx :)

embano1 · 2024-08-03T07:19:49Z

protocol/nats_jetstream/v2/receiver.go

+// NewConsumerV2 consumes from filterSubjects given in ConsumerConfig or OrderedConsumerConfig
+// embedded in ConsumerOption.
+// See: WithConsumerConfig(...) and WithOrderedConsumerConfig(...)
+func NewConsumerV2(url string, natsOpts []nats.Option, jsOpts []jetstream.JetStreamOpt, opts ...ConsumerOption) (*Consumer, error) {


Question: do those need to be public?

I assume you mean the options? Maybe possible to drop jetstream.JetStreamOpt

nats.Option:
About 40 to 50 options

jetstream.JetStreamOpt:

WithClientTrace(ct *ClientTrace) JetStreamOpt

WithPublishAsyncErrHandler(cb MsgErrHandler) JetStreamOpt

WithPublishAsyncMaxPending(max int) JetStreamOpt

The ConsumerOption options are needed because there are variations of how the NATS consumer is created.
There are two variations, "normal" and "ordered". Each has mutiple fields on how to configure.

embano1 · 2024-08-03T07:21:56Z

protocol/nats_jetstream/v2/receiver.go

+	} else if c.JetStream != nil {
+		err = c.createConsumer(ctx)
+		if err != nil {
+			return err
+		}
+		consumeContext, err = c.JetstreamConsumer.Consume(c.MsgHandlerV2, c.PullConsumeOpt...)
+		if err != nil {
+			return err
+		}
+	} else {
+		return ErrNoJetstream


can be simplified to:

just if c.JetStream != nil {...}

remove final else and return error directly

(avoids else logic)

embano1 · 2024-08-03T07:22:18Z

protocol/nats_jetstream/v2/receiver.go

-	return sub.Drain()
+	if sub != nil {
+		return sub.Drain()
+	} else if consumeContext != nil {


same as above

embano1 · 2024-08-03T07:22:57Z

protocol/nats_jetstream/v2/receiver.go

+	if c.OrderedConsumerConfig != nil {
+		consumer, err = c.JetStream.OrderedConsumer(ctx, stream, *c.OrderedConsumerConfig)
+	} else if c.ConsumerConfig != nil {
+		consumer, err = c.JetStream.CreateOrUpdateConsumer(ctx, stream, *c.ConsumerConfig)
+	} else {
+		return ErrNoConsumerConfig
+	}


can we make this switch/case style?

embano1 · 2024-08-03T07:23:18Z

protocol/nats_jetstream/v2/receiver.go

+// getStreamFromSubjects finds the unique stream for the set of filter subjects
+// If more than one stream is found, returns ErrMoreThanOneStream
+func (c *Consumer) getStreamFromSubjects() (string, error) {
+	ctx := context.Background()


allow passing ctx as parameter?

stephen-totty-hpe · 2024-08-05T17:02:23Z

I'm still torn between adding a new V2 API vs introducing a breaking change and remove the old code.

@embano1
Yeah, I made this PR as a quick proof of concept to get some feedback on which way to go. I have the same questions about how to proceed.

Signed-off-by: stephen-totty-hpe <[email protected]>

embano1 · 2024-08-11T07:26:16Z

@duglin @lionelvillard what are your thoughts on changing the NATS JetStream implementation to the latest NATS SDK which provides several advantages for users, and is the recommended way to use JetStream. However, this means:

We either need to define V2 methods on this package to expose the new capabilities
Introduce a protocol/nats_jetstream/v3 Go module
Break existing APIs without introducing new Go module/V2 above

What are your preferences?

evankanderson · 2024-08-11T12:05:03Z

protocol/nats_jetstream/v2/options.go

 )

 var ErrInvalidQueueName = errors.New("invalid queue name for QueueSubscriber")
+var ErrNoConsumerConfig = errors.New("no consumer config was given")


Actually, with errors.As, using simple exported errors works rather than needing to do something more complicated unless you want additional methods.

evankanderson · 2024-08-11T12:14:49Z

protocol/nats_jetstream/v2/receiver.go

+	natsMessage := &nats.Msg{
+		Subject: msg.Subject(),
+		Reply:   msg.Reply(),
+		Header:  msg.Headers(),
+		Data:    msg.Data(),
+	}
+	r.incoming <- msgErr{msg: NewMessage(natsMessage)}


Another option would be to define an interface that both nats.Msg and jetstream.Msg could be wrapped to implement, e.g.

type natsData interface { Subject() string Reply() string Headers() nats.Header Data() []byte }

But this seems likely to be just as effective.

evankanderson · 2024-08-11T12:16:59Z

protocol/nats_jetstream/v2/receiver.go

@@ -54,12 +69,20 @@ func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {
 type Consumer struct {


Given that we've exposed this as a struct with named v1 fields, it feels like maybe introducing a new v3 version of nats_jetstream might be the way to go. I don't have a really strong feeling about it, though.

stephen-totty-hpe · 2024-08-23T13:47:19Z

embano1

@embano1 @evankanderson
Any thoughts on path forward? I am fine either way.

embano1 · 2024-08-26T08:31:35Z

@duglin @lionelvillard any thoughts on #1083 (comment) ?

stephen-totty-hpe · 2024-09-26T15:23:34Z

embano1

@embano1 @evankanderson Any thoughts on path forward? I am fine either way.

I will close this PR if #1095 gets merged.

stephen-totty-hpe requested a review from a team as a code owner July 29, 2024 21:52

stephen-totty-hpe force-pushed the totty/nats-newjetstream branch 2 times, most recently from ca8466a to a4a3019 Compare July 30, 2024 13:43

embano1 reviewed Aug 3, 2024

View reviewed changes

embano1 mentioned this pull request Aug 4, 2024

Add flexibility in subject matching for nats_jetstream #1084

Merged

stephen-totty-hpe force-pushed the totty/nats-newjetstream branch from a4a3019 to a64334f Compare August 6, 2024 13:26

implement nats_jetstream using new jetstream package

4a199db

Signed-off-by: stephen-totty-hpe <[email protected]>

stephen-totty-hpe force-pushed the totty/nats-newjetstream branch from a64334f to 4a199db Compare August 6, 2024 15:35

evankanderson reviewed Aug 11, 2024

View reviewed changes

stephen-totty-hpe mentioned this pull request Sep 10, 2024

Add v3 version of NATS JetStream protocol #1095

Merged

stephen-totty-hpe closed this Oct 15, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement nats_jetstream using new jetstream package #1083

Implement nats_jetstream using new jetstream package #1083

stephen-totty-hpe commented Jul 29, 2024

embano1 left a comment

embano1 Aug 3, 2024

evankanderson Aug 11, 2024

embano1 Aug 3, 2024

stephen-totty-hpe Aug 5, 2024

embano1 Aug 3, 2024

stephen-totty-hpe Aug 6, 2024

embano1 Aug 3, 2024

embano1 Aug 3, 2024

stephen-totty-hpe Aug 6, 2024

embano1 Aug 3, 2024

embano1 Aug 3, 2024

embano1 Aug 3, 2024

embano1 Aug 3, 2024

embano1 Aug 3, 2024

stephen-totty-hpe commented Aug 5, 2024

embano1 commented Aug 11, 2024

evankanderson Aug 11, 2024

evankanderson Aug 11, 2024

evankanderson Aug 11, 2024

stephen-totty-hpe commented Aug 23, 2024

embano1 commented Aug 26, 2024

stephen-totty-hpe commented Sep 26, 2024

		@@ -54,12 +69,20 @@ func (r *Receiver) Receive(ctx context.Context) (binding.Message, error) {
		type Consumer struct {

Implement nats_jetstream using new jetstream package #1083

Implement nats_jetstream using new jetstream package #1083

Conversation

stephen-totty-hpe commented Jul 29, 2024

embano1 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephen-totty-hpe commented Aug 5, 2024

embano1 commented Aug 11, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephen-totty-hpe commented Aug 23, 2024

embano1 commented Aug 26, 2024

stephen-totty-hpe commented Sep 26, 2024