add v3 version of nats jetstream protocol with integration tests and samples #1095

stephen-totty-hpe · 2024-09-10T21:18:07Z

As been discussed previously, there has been a desire to implement the CloudEventSDK nats jetstream protocol using the newer jetstream package. The newer jetstream package exposes more jetstream specific functionality that should make it more flexible to use.

Much of this code is stolen from the v2 implementation and changed where necessary. It builds upon comments made in:
#1083

Under the covers, this uses a jetstream.Consumer, which can be "normal" or "ordered". This is done using consumer options WithConsumerConfig and WithOrderedConsumerConfig.

Also, the need to know the stream names upfront is not needed. Many of the internal options are available if needed, although the only required option is the ConsumerOption with the correct config.

embano1

Regarding the samples, please take a look at how we use a protocol in https://github.com/cloudevents/sdk-go/blob/main/samples/kafka_confluent/sender/main.go

protocol/nats_jetstream/v3/options.go

protocol/nats_jetstream/v3/options_test.go

protocol/nats_jetstream/v3/sender.go

samples/nats_jetstream/v3/message-interoperability/main.go

samples/nats_jetstream/v3/receiver/main.go

protocol/nats_jetstream/v3/options.go

protocol/nats_jetstream/v3/protocol.go

protocol/nats_jetstream/v3/message.go

protocol/nats_jetstream/v3/protocol.go

protocol/nats_jetstream/v3/receiver.go

yanmxa · 2024-10-08T02:37:00Z

@stephen-totty-hpe Thanks! It looks good to me. Leave it to @embano1 ~

protocol/nats_jetstream/v3/message_test.go

embano1 · 2024-10-08T07:31:07Z

protocol/nats_jetstream/v3/options.go

+type ProtocolOption func(*Protocol) error
+
+// WithURL configures the Sender and/or Receiver
+func WithURL(url string, natsOpts ...nats.Option) ProtocolOption {


There's quite a lot of logic in this option. I thought URL was always required? If so, better to make it a constructor parameter and configure these fields after all options are applied.

You can use one of two options:

URL +nats.Options // if you do not have a connection yet

*nats.Conn // if you already have a connection.

This requires either:

two constructors
a) context.Context, URL , nats.Options, ...ProtocolOption
b) context.Context, *nats.Conn, ...ProtocolOption

Options that makes the two cases "mutually exclusive"

Simplified more. One constructor.

Ok, thx for clarifying. So we need those two options then. I don't think having an option with additional options here is what we want though. Please remove the additional options slice in this option then.

Suggested change

func WithURL(url string, natsOpts ...nats.Option) ProtocolOption {

func WithURL(url string) ProtocolOption {

protocol/nats_jetstream/v3/options.go

embano1 · 2024-10-08T07:32:36Z

protocol/nats_jetstream/v3/options.go

+	}
+}
+
+// WithConnection configures the Sender and/or Receiver


Please update all option comments with a clear description of what they do e.g.,

Suggested change

// WithConnection configures the Sender and/or Receiver

// WithConnection uses the provided connection in the protocol sender and receiver

protocol/nats_jetstream/v3/options.go

samples/nats_jetstream/v3/receiver/main.go

samples/nats_jetstream/v3/sender/main.go

embano1 · 2024-10-08T07:54:34Z

protocol/nats_jetstream/v3/sender.go

+
+// NewSender creates a new protocol.Sender responsible for opening and closing the NATS connection
+// Use WithURL() OR WithConnection()
+func NewSender(ctx context.Context, subject string, opts ...ProtocolOption) (*sender, error) {


Can we make these sender/receiver constructors private?

The protocol constructor makes both a sender and receiver. in some cases you might only want a sender OR receiver.

I can move sendSubject to an Option.
Then I can check:
A) ConsumerConfig/OrderedConsumerConfig to see if the use wants a receiver
B) SendSubject to see if the user wants a sender

protocol/nats_jetstream/v3/receiver.go

embano1

Thx for all the hard work you put into this! Some minor comments on the Options implementation (really like it) and I think we can further reduce the complexity of this code base, especially in the sender/receiver implementations. I suggested to take a look at the HTTP/Kafka (Confluent) protocol implementation for how to simplify the internal logic of this protocol.

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

embano1 · 2024-10-09T05:34:31Z

protocol/nats_jetstream/v3/message_test.go

+			if got == nil {
+				t.Errorf("Error in NewMessage!")
+			}
+			err := got.ReadBinary(context.TODO(), &outBinaryMessage)


nit: shouldn't you also assert that the contents of the messages are as expected?

embano1 · 2024-10-09T05:35:53Z

protocol/nats_jetstream/v3/options.go

+}
+
+// WithConsumerConfig creates a unordered consumer used in the protocol receiver.
+// If WithOrderedConsumerConfig is also given, the receiver will error.


Suggested change

// If WithOrderedConsumerConfig is also given, the receiver will error.

// This option is mutually exclusive with WithOrderedConsumerConfig.

embano1 · 2024-10-09T05:36:14Z

protocol/nats_jetstream/v3/options.go

+}
+
+// WithOrderedConsumerConfig creates a ordered consumer used in the protocol receiver.
+// If WithConsumerConfig is also given, the receiver will error.


Suggested change

// If WithConsumerConfig is also given, the receiver will error.

// This option is mutually exclusive with WithConsumerConfig.

embano1 · 2024-10-09T05:37:32Z

protocol/nats_jetstream/v3/options.go

+type ProtocolOption func(*Protocol) error
+
+// WithURL configures the Sender and/or Receiver
+func WithURL(url string, natsOpts ...nats.Option) ProtocolOption {


Ok, thx for clarifying. So we need those two options then. I don't think having an option with additional options here is what we want though. Please remove the additional options slice in this option then.

Suggested change

func WithURL(url string, natsOpts ...nats.Option) ProtocolOption {

func WithURL(url string) ProtocolOption {

embano1 · 2024-10-09T05:39:04Z

protocol/nats_jetstream/v3/options_test.go

+ SPDX-License-Identifier: Apache-2.0
+*/
+
+package nats_jetstream


nit: many of these tests don't assert error cases. If you run your tests with coverage, you'll see that several code paths are not checked

embano1 · 2024-10-09T05:44:50Z

protocol/nats_jetstream/v3/protocol.go

+	// Before closing, let's be sure OpenInbound completes
+	// We send a signal to close and then we lock on subMtx in order
+	// to wait OpenInbound to finish draining the queue
+	p.internalClose <- struct{}{}


If OpenInbound is already done (ctx cancelled) this will deadlock

A quick fix IMHO is a buffered channel of 1

embano1 · 2024-10-09T05:46:00Z

protocol/nats_jetstream/v3/protocol.go

+
+// validateOptions runs after all options have been applied and makes sure needed options were set correctly.
+func (p *Protocol) validateOptions() error {
+	if p.url == "" && p.conn == nil {


you also need to check for conflicting options here

embano1 · 2024-10-09T05:49:21Z

protocol/nats_jetstream/v3/protocol.go

+	p.incoming <- msgErr{msg: NewMessage(msg)}
+}
+
+// Close implements Closer.Close


Suggested change

// Close implements Closer.Close

// Close must be called after use to releases internal resources and close the underlying connection

embano1 · 2024-10-09T05:50:53Z

test/integration/nats_jetstream/v3/nats_test.go

+	return conn
+}
+
+func testProtocol(ctx context.Context, t testing.TB, natsConn *nats.Conn, consumerConfig any, opts ...ce_nats.ProtocolOption) (func(), bindings.Sender,


Shouldn't this be TestProtocol?

embano1 · 2024-10-09T05:51:16Z

test/integration/nats_jetstream/v3/nats_test.go

+	}
+}
+
+func testConn(t testing.TB) *nats.Conn {


Helper, therefore let's not confuse names here

Suggested change

func testConn(t testing.TB) *nats.Conn {

func createTestConnection(t testing.TB) *nats.Conn {

stephen-totty-hpe requested a review from a team as a code owner September 10, 2024 21:18

embano1 reviewed Sep 15, 2024

View reviewed changes

stephen-totty-hpe force-pushed the totty/nats-v3 branch 6 times, most recently from 16961e8 to 6f1ff3d Compare September 16, 2024 20:47

embano1 reviewed Sep 17, 2024

View reviewed changes

stephen-totty-hpe force-pushed the totty/nats-v3 branch from 6f1ff3d to df22aee Compare September 26, 2024 15:39

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/message.go Outdated Show resolved Hide resolved

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/message.go Outdated Show resolved Hide resolved

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/message.go Show resolved Hide resolved

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/protocol.go Outdated Show resolved Hide resolved

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/receiver.go Outdated Show resolved Hide resolved

yanmxa reviewed Sep 28, 2024

View reviewed changes

protocol/nats_jetstream/v3/receiver.go Outdated Show resolved Hide resolved

stephen-totty-hpe force-pushed the totty/nats-v3 branch 4 times, most recently from 0e2da9c to 2cbe682 Compare September 30, 2024 19:33

stephen-totty-hpe requested review from embano1 and yanmxa October 4, 2024 17:12

embano1 reviewed Oct 8, 2024

View reviewed changes

embano1 requested changes Oct 8, 2024

View reviewed changes

stephen-totty-hpe force-pushed the totty/nats-v3 branch from 2cbe682 to 9d9a30e Compare October 8, 2024 18:01

stephen-totty-hpe force-pushed the totty/nats-v3 branch from 9d9a30e to 5a16005 Compare October 8, 2024 18:02

add v3 version of nats jetstream protocol with integration tests and …

7f198df

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

stephen-totty-hpe force-pushed the totty/nats-v3 branch from 5a16005 to 7f198df Compare October 8, 2024 18:11

stephen-totty-hpe requested a review from embano1 October 8, 2024 18:18

embano1 reviewed Oct 9, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add v3 version of nats jetstream protocol with integration tests and samples #1095

add v3 version of nats jetstream protocol with integration tests and samples #1095

stephen-totty-hpe commented Sep 10, 2024

embano1 left a comment

yanmxa commented Oct 8, 2024

embano1 Oct 8, 2024

stephen-totty-hpe Oct 8, 2024

stephen-totty-hpe Oct 8, 2024

embano1 Oct 9, 2024

embano1 Oct 8, 2024

embano1 Oct 8, 2024

stephen-totty-hpe Oct 8, 2024

stephen-totty-hpe Oct 8, 2024

embano1 left a comment

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

embano1 Oct 9, 2024

	func WithURL(url string, natsOpts ...nats.Option) ProtocolOption {
	func WithURL(url string) ProtocolOption {

	// WithConnection configures the Sender and/or Receiver
	// WithConnection uses the provided connection in the protocol sender and receiver

	// If WithOrderedConsumerConfig is also given, the receiver will error.
	// This option is mutually exclusive with WithOrderedConsumerConfig.

	// If WithConsumerConfig is also given, the receiver will error.
	// This option is mutually exclusive with WithConsumerConfig.

	// Close implements Closer.Close
	// Close must be called after use to releases internal resources and close the underlying connection

	func testConn(t testing.TB) *nats.Conn {
	func createTestConnection(t testing.TB) *nats.Conn {

add v3 version of nats jetstream protocol with integration tests and samples #1095

Are you sure you want to change the base?

add v3 version of nats jetstream protocol with integration tests and samples #1095

Conversation

stephen-totty-hpe commented Sep 10, 2024

embano1 left a comment

Choose a reason for hiding this comment

yanmxa commented Oct 8, 2024

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

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