From 16961e8ec7079893522950e18bfcc8ccbb89692b Mon Sep 17 00:00:00 2001 From: stephen-totty-hpe Date: Tue, 10 Sep 2024 16:59:51 -0400 Subject: [PATCH] add v3 version of nats jetstream protocol with integration tests and samples Signed-off-by: stephen-totty-hpe --- protocol/nats_jetstream/v3/go.mod | 27 +++ protocol/nats_jetstream/v3/go.sum | 47 ++++ protocol/nats_jetstream/v3/message.go | 145 ++++++++++++ protocol/nats_jetstream/v3/message_test.go | 111 +++++++++ protocol/nats_jetstream/v3/options.go | 92 ++++++++ protocol/nats_jetstream/v3/options_test.go | 176 ++++++++++++++ protocol/nats_jetstream/v3/protocol.go | 100 ++++++++ protocol/nats_jetstream/v3/receiver.go | 219 ++++++++++++++++++ protocol/nats_jetstream/v3/sender.go | 122 ++++++++++ protocol/nats_jetstream/v3/write_message.go | 98 ++++++++ samples/nats_jetstream/v3/go.mod | 31 +++ samples/nats_jetstream/v3/go.sum | 47 ++++ samples/nats_jetstream/v3/receiver/main.go | 61 +++++ samples/nats_jetstream/v3/sender/main.go | 80 +++++++ test/integration/go.mod | 16 +- test/integration/go.sum | 28 +-- .../nats_jetstream/v3/nats_test.go | 162 +++++++++++++ 17 files changed, 1543 insertions(+), 19 deletions(-) create mode 100644 protocol/nats_jetstream/v3/go.mod create mode 100644 protocol/nats_jetstream/v3/go.sum create mode 100644 protocol/nats_jetstream/v3/message.go create mode 100644 protocol/nats_jetstream/v3/message_test.go create mode 100644 protocol/nats_jetstream/v3/options.go create mode 100644 protocol/nats_jetstream/v3/options_test.go create mode 100644 protocol/nats_jetstream/v3/protocol.go create mode 100644 protocol/nats_jetstream/v3/receiver.go create mode 100644 protocol/nats_jetstream/v3/sender.go create mode 100644 protocol/nats_jetstream/v3/write_message.go create mode 100644 samples/nats_jetstream/v3/go.mod create mode 100644 samples/nats_jetstream/v3/go.sum create mode 100644 samples/nats_jetstream/v3/receiver/main.go create mode 100644 samples/nats_jetstream/v3/sender/main.go create mode 100644 test/integration/nats_jetstream/v3/nats_test.go diff --git a/protocol/nats_jetstream/v3/go.mod b/protocol/nats_jetstream/v3/go.mod new file mode 100644 index 000000000..2b20a3062 --- /dev/null +++ b/protocol/nats_jetstream/v3/go.mod @@ -0,0 +1,27 @@ +module github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3 + +go 1.18 + +replace github.com/cloudevents/sdk-go/v2 => ../../../v2 + +require ( + github.com/cloudevents/sdk-go/v2 v2.15.2 + github.com/nats-io/nats.go v1.37.0 +) + +require ( + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/google/go-cmp v0.5.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/nats-io/nkeys v0.4.7 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/stretchr/testify v1.8.0 // indirect + golang.org/x/crypto v0.27.0 // indirect + golang.org/x/sys v0.25.0 // indirect + golang.org/x/text v0.18.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/protocol/nats_jetstream/v3/go.sum b/protocol/nats_jetstream/v3/go.sum new file mode 100644 index 000000000..a6562f7ad --- /dev/null +++ b/protocol/nats_jetstream/v3/go.sum @@ -0,0 +1,47 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w= +github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE= +github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8= +github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI= +github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU= +go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI= +go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM= +golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A= +golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70= +golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34= +golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224= +golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/protocol/nats_jetstream/v3/message.go b/protocol/nats_jetstream/v3/message.go new file mode 100644 index 000000000..b88779a3f --- /dev/null +++ b/protocol/nats_jetstream/v3/message.go @@ -0,0 +1,145 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "bytes" + "context" + "fmt" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/binding/format" + "github.com/cloudevents/sdk-go/v2/binding/spec" +) + +const ( + // see https://github.com/cloudevents/spec/blob/main/cloudevents/bindings/nats-protocol-binding.md + prefix = "ce-" + contentTypeHeader = "content-type" +) + +var specs = spec.WithPrefix(prefix) + +// Message implements binding.Message by wrapping an jetstream.Msg. +// This message *can* be read several times safely +type Message struct { + Msg jetstream.Msg + encoding binding.Encoding +} + +// NewMessage wraps an *nats.Msg in a binding.Message. +// The returned message *can* be read several times safely +// The default encoding returned is EncodingStructured unless the NATS message contains a specversion header. +func NewMessage(msg jetstream.Msg) *Message { + encoding := binding.EncodingStructured + if msg.Headers() != nil { + if msg.Headers().Get(specs.PrefixedSpecVersionName()) != "" { + encoding = binding.EncodingBinary + } + } + return &Message{Msg: msg, encoding: encoding} +} + +var _ binding.Message = (*Message)(nil) + +// ReadEncoding return the type of the message Encoding. +func (m *Message) ReadEncoding() binding.Encoding { + return m.encoding +} + +// ReadStructured transfers a structured-mode event to a StructuredWriter. +func (m *Message) ReadStructured(ctx context.Context, encoder binding.StructuredWriter) error { + if m.encoding != binding.EncodingStructured { + return binding.ErrNotStructured + } + return encoder.SetStructuredEvent(ctx, format.JSON, bytes.NewReader(m.Msg.Data())) +} + +// ReadBinary transfers a binary-mode event to an BinaryWriter. +func (m *Message) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) error { + if m.encoding != binding.EncodingBinary { + return binding.ErrNotBinary + } + + version := m.GetVersion() + if version == nil { + return binding.ErrNotBinary + } + + var err error + for k, v := range m.Msg.Headers() { + headerValue := v[0] + if strings.HasPrefix(k, prefix) { + attr := version.Attribute(k) + if attr != nil { + err = encoder.SetAttribute(attr, headerValue) + } else { + err = encoder.SetExtension(strings.TrimPrefix(k, prefix), headerValue) + } + } else if k == contentTypeHeader { + err = encoder.SetAttribute(version.AttributeFromKind(spec.DataContentType), headerValue) + } + if err != nil { + return err + } + } + + if m.Msg.Data() != nil { + err = encoder.SetData(bytes.NewBuffer(m.Msg.Data())) + } + + return err +} + +// Finish *must* be called when message from a Receiver can be forgotten by the receiver. +func (m *Message) Finish(err error) error { + return nil +} + +// GetAttribute implements binding.MessageMetadataReader +func (m *Message) GetAttribute(attributeKind spec.Kind) (spec.Attribute, interface{}) { + key := withPrefix(attributeKind.String()) + if m.Msg.Headers() != nil { + headerValue := m.Msg.Headers().Get(key) + if headerValue != "" { + version := m.GetVersion() + return version.Attribute(key), headerValue + } + } + return nil, nil +} + +// GetExtension implements binding.MessageMetadataReader +func (m *Message) GetExtension(name string) interface{} { + key := withPrefix(name) + if m.Msg.Headers() != nil { + headerValue := m.Msg.Headers().Get(key) + if headerValue != "" { + return headerValue + } + } + return nil +} + +// GetVersion looks for specVersion header and returns a Version object +func (m *Message) GetVersion() spec.Version { + if m.Msg.Headers() == nil { + return nil + } + versionValue := m.Msg.Headers().Get(specs.PrefixedSpecVersionName()) + if versionValue == "" { + return nil + } + return specs.Version(versionValue) +} + +// withPrefix prepends the prefix to the attribute name +func withPrefix(attributeName string) string { + return fmt.Sprintf("%s%s", prefix, attributeName) +} diff --git a/protocol/nats_jetstream/v3/message_test.go b/protocol/nats_jetstream/v3/message_test.go new file mode 100644 index 000000000..553540587 --- /dev/null +++ b/protocol/nats_jetstream/v3/message_test.go @@ -0,0 +1,111 @@ +package nats_jetstream + +import ( + "context" + "encoding/json" + "testing" + + "github.com/cloudevents/sdk-go/v2/binding/spec" + bindingtest "github.com/cloudevents/sdk-go/v2/binding/test" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/test" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +type jetStreamMsg struct { + jetstream.Msg + msg *nats.Msg +} + +func (j *jetStreamMsg) Data() []byte { return j.msg.Data } +func (j *jetStreamMsg) Headers() nats.Header { return j.msg.Header } + +var ( + outBinaryMessage = bindingtest.MockBinaryMessage{ + Metadata: map[spec.Attribute]interface{}{}, + Extensions: map[string]interface{}{}, + } + outStructMessage = bindingtest.MockStructuredMessage{} + + testEvent = test.FullEvent() + binaryData, _ = json.Marshal(map[string]string{ + "ce_type": testEvent.Type(), + "ce_source": testEvent.Source(), + "ce_id": testEvent.ID(), + "ce_time": test.Timestamp.String(), + "ce_specversion": "1.0", + "ce_dataschema": test.Schema.String(), + "ce_datacontenttype": "text/json", + "ce_subject": "receiverTopic", + "ce_exta": "someext", + }) + structuredConsumerMessage = &jetStreamMsg{ + msg: &nats.Msg{ + Subject: "hello", + Data: binaryData, + }, + } + binaryConsumerMessage = &jetStreamMsg{ + msg: &nats.Msg{ + Subject: "hello", + Data: testEvent.Data(), + Header: nats.Header{ + "ce-type": {testEvent.Type()}, + "ce-source": {testEvent.Source()}, + "ce-id": {testEvent.ID()}, + "ce-time": {test.Timestamp.String()}, + "ce-specversion": {"1.0"}, + "ce-dataschema": {test.Schema.String()}, + "ce-datacontenttype": {"text/json"}, + "ce-subject": {"receiverTopic"}, + "ce-exta": {"someext"}, + }, + }, + } +) + +func TestNewMessage(t *testing.T) { + tests := []struct { + name string + consumerMessage jetstream.Msg + expectedEncoding binding.Encoding + expectedStructuredError error + expectedBinaryError error + }{ + { + name: "Structured encoding", + consumerMessage: structuredConsumerMessage, + expectedEncoding: binding.EncodingStructured, + expectedStructuredError: nil, + expectedBinaryError: binding.ErrNotBinary, + }, + { + name: "Binary encoding", + consumerMessage: binaryConsumerMessage, + expectedEncoding: binding.EncodingBinary, + expectedStructuredError: binding.ErrNotStructured, + expectedBinaryError: nil, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := NewMessage(tt.consumerMessage) + if got == nil { + t.Errorf("Error in NewMessage!") + } + err := got.ReadBinary(context.TODO(), &outBinaryMessage) + if err != tt.expectedBinaryError { + t.Errorf("ReadBinary err:%s", err.Error()) + } + err = got.ReadStructured(context.TODO(), &outStructMessage) + if err != tt.expectedStructuredError { + t.Errorf("ReadStructured err:%s", err.Error()) + } + if got.ReadEncoding() != tt.expectedEncoding { + t.Errorf("ExpectedEncoding %s, while got %s", tt.expectedEncoding, got.ReadEncoding()) + } + }) + } +} diff --git a/protocol/nats_jetstream/v3/options.go b/protocol/nats_jetstream/v3/options.go new file mode 100644 index 000000000..82e4257a2 --- /dev/null +++ b/protocol/nats_jetstream/v3/options.go @@ -0,0 +1,92 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "errors" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +var ( + ErrNoFilterSubjects = errors.New("no filter subjects were given") + ErrMoreThanOneStream = errors.New("more than one stream for given filter subjects") + ErrNoConsumerConfig = errors.New("no consumer config was given") + ErrMoreThanOneConsumerConfig = errors.New("more than one consumer config given") +) + +// NatsOptions is a helper function to group a variadic nats.ProtocolOption into +// []nats.Option that can be used by either Sender, Consumer or Protocol +func NatsOptions(opts ...nats.Option) []nats.Option { + return opts +} + +// ProtocolOption is the function signature required to be considered an nats.ProtocolOption. +type ProtocolOption func(*Protocol) error + +// WithJetStreamOptions configures the Sender and/or Consumer +func WithJetStreamOptions(jetStreamOpts []jetstream.JetStreamOpt) ProtocolOption { + return func(p *Protocol) error { + if p.Sender != nil { + p.Sender.jetSteamOpts = jetStreamOpts + } + if p.Consumer != nil { + p.Consumer.jetSteamOpts = jetStreamOpts + } + return nil + } +} + +// WithPublishOptions configures the Sender +func WithPublishOptions(publishOpts []jetstream.PublishOpt) ProtocolOption { + return func(p *Protocol) error { + if p.Sender == nil { + return nil + } + p.Sender.publishOpts = publishOpts + return nil + } +} + +// WithConsumerConfig configures the Consumer with the given config +func WithConsumerConfig(consumerConfig *jetstream.ConsumerConfig) ProtocolOption { + return func(p *Protocol) error { + if p.Consumer == nil { + return nil + } + if p.Consumer.orderedConsumerConfig != nil { + return ErrMoreThanOneConsumerConfig + } + p.Consumer.consumerConfig = consumerConfig + return nil + } +} + +// WithOrderedConsumerConfig configures the Consumer with the given config +func WithOrderedConsumerConfig(orderedConsumerConfig *jetstream.OrderedConsumerConfig) ProtocolOption { + return func(p *Protocol) error { + if p.Consumer == nil { + return nil + } + if p.Consumer.consumerConfig != nil { + return ErrMoreThanOneConsumerConfig + } + p.Consumer.orderedConsumerConfig = orderedConsumerConfig + return nil + } +} + +// WithPullConsumerOptions configures the Consumer with the given pullConsumeOpts +func WithPullConsumerOptions(pullConsumeOpts []jetstream.PullConsumeOpt) ProtocolOption { + return func(p *Protocol) error { + if p.Consumer == nil { + return nil + } + p.Consumer.pullConsumeOpts = pullConsumeOpts + return nil + } +} diff --git a/protocol/nats_jetstream/v3/options_test.go b/protocol/nats_jetstream/v3/options_test.go new file mode 100644 index 000000000..faffff964 --- /dev/null +++ b/protocol/nats_jetstream/v3/options_test.go @@ -0,0 +1,176 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "reflect" + "testing" + + "github.com/nats-io/nats.go/jetstream" +) + +func TestWithConsumerConfig(t *testing.T) { + filterSubjects := []string{"normal"} + type args struct { + consumer *consumer + config *jetstream.ConsumerConfig + } + type wants struct { + err error + consumer *consumer + } + tests := []struct { + name string + args args + wants wants + }{ + { + name: "valid case", + args: args{ + consumer: &consumer{}, + config: &jetstream.ConsumerConfig{FilterSubjects: filterSubjects}, + }, + wants: wants{ + err: nil, + consumer: &consumer{ + consumerConfig: &jetstream.ConsumerConfig{FilterSubjects: filterSubjects}, + }, + }, + }, + { + name: "too many consumer options", + args: args{ + consumer: &consumer{orderedConsumerConfig: &jetstream.OrderedConsumerConfig{FilterSubjects: filterSubjects}}, + config: &jetstream.ConsumerConfig{FilterSubjects: filterSubjects}, + }, + wants: wants{ + err: ErrMoreThanOneConsumerConfig, + consumer: &consumer{orderedConsumerConfig: &jetstream.OrderedConsumerConfig{FilterSubjects: filterSubjects}}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotErr := tt.args.consumer.applyOptions(WithConsumerConfig(tt.args.config)) + if gotErr != tt.wants.err { + t.Errorf("applyOptions(WithConsumerConfig()) = %v, want %v", gotErr, tt.wants.err) + } + + if !reflect.DeepEqual(tt.args.consumer, tt.wants.consumer) { + t.Errorf("p = %v, want %v", tt.args.consumer, tt.wants.consumer) + } + }) + } +} + +func TestOrderedConsumerConfig(t *testing.T) { + filterSubjects := []string{"ordered"} + type args struct { + consumer *consumer + config *jetstream.OrderedConsumerConfig + } + type wants struct { + err error + consumer *consumer + } + tests := []struct { + name string + args args + wants wants + }{ + { + name: "valid case", + args: args{ + consumer: &consumer{}, + config: &jetstream.OrderedConsumerConfig{FilterSubjects: filterSubjects}, + }, + wants: wants{ + err: nil, + consumer: &consumer{ + orderedConsumerConfig: &jetstream.OrderedConsumerConfig{FilterSubjects: filterSubjects}, + }, + }, + }, + { + name: "too many consumer options", + args: args{ + consumer: &consumer{consumerConfig: &jetstream.ConsumerConfig{FilterSubjects: filterSubjects}}, + config: &jetstream.OrderedConsumerConfig{FilterSubjects: filterSubjects}, + }, + wants: wants{ + err: ErrMoreThanOneConsumerConfig, + consumer: &consumer{consumerConfig: &jetstream.ConsumerConfig{FilterSubjects: filterSubjects}}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotErr := tt.args.consumer.applyOptions(WithOrderedConsumerConfig(tt.args.config)) + if gotErr != tt.wants.err { + t.Errorf("applyOptions(WithOrderedConsumerConfig()) = %v, want %v", gotErr, tt.wants.err) + } + + if !reflect.DeepEqual(tt.args.consumer, tt.wants.consumer) { + t.Errorf("p = %v, want %v", tt.args.consumer, tt.wants.consumer) + } + }) + } +} + +func TestWithPullConsumeOptions(t *testing.T) { + maxMessages := jetstream.PullMaxMessages(1) + maxBytes := jetstream.PullMaxBytes(0) + type args struct { + consumer *consumer + config []jetstream.PullConsumeOpt + } + type wants struct { + err error + consumer *consumer + } + tests := []struct { + name string + args args + wants wants + }{ + { + name: "pull consumer option given", + args: args{ + consumer: &consumer{}, + config: []jetstream.PullConsumeOpt{maxMessages, maxBytes}, + }, + wants: wants{ + err: nil, + consumer: &consumer{ + pullConsumeOpts: []jetstream.PullConsumeOpt{maxMessages, maxBytes}, + }, + }, + }, + { + name: "no pull consumer option given", + args: args{ + consumer: &consumer{}, + config: nil, + }, + wants: wants{ + err: nil, + consumer: &consumer{pullConsumeOpts: nil}, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotErr := tt.args.consumer.applyOptions(WithPullConsumerOptions(tt.args.config)) + if gotErr != tt.wants.err { + t.Errorf("applyOptions(WithPullConsumerOptions()) = %v, want %v", gotErr, tt.wants.err) + } + + if !reflect.DeepEqual(tt.args.consumer, tt.wants.consumer) { + t.Errorf("p = %v, want %v", tt.args.consumer, tt.wants.consumer) + } + }) + } +} diff --git a/protocol/nats_jetstream/v3/protocol.go b/protocol/nats_jetstream/v3/protocol.go new file mode 100644 index 000000000..d64efb567 --- /dev/null +++ b/protocol/nats_jetstream/v3/protocol.go @@ -0,0 +1,100 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "context" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/protocol" + + "github.com/nats-io/nats.go" +) + +// Protocol is a reference implementation for using the CloudEvents binding +// integration. Protocol acts as both a NATS client and a NATS handler. +type Protocol struct { + conn *nats.Conn + + Consumer *consumer + + Sender *sender + + connOwned bool // whether this protocol created the nats connection +} + +// NewProtocol creates a new NATS protocol. +func NewProtocol(ctx context.Context, url, sendSubject string, natsOpts []nats.Option, opts ...ProtocolOption) (*Protocol, error) { + conn, err := nats.Connect(url, natsOpts...) + if err != nil { + return nil, err + } + + p, err := NewProtocolFromConn(ctx, conn, sendSubject, opts...) + if err != nil { + conn.Close() + return nil, err + } + + p.connOwned = true + + return p, nil +} + +// NewProtocolFromConn creates a new NATS protocol with the given nats connection. +func NewProtocolFromConn(ctx context.Context, conn *nats.Conn, sendSubject string, opts ...ProtocolOption) (*Protocol, error) { + var err error + p := &Protocol{ + conn: conn, + } + + if p.Consumer, err = NewConsumerFromConn(ctx, conn, opts...); err != nil { + return nil, err + } + + if p.Sender, err = NewSenderFromConn(ctx, conn, sendSubject, opts...); err != nil { + return nil, err + } + + return p, nil +} + +// Send implements Sender.Send +func (p *Protocol) Send(ctx context.Context, in binding.Message, transformers ...binding.Transformer) error { + return p.Sender.Send(ctx, in, transformers...) +} + +// OpenInbound implements Opener.OpenInbound +func (p *Protocol) OpenInbound(ctx context.Context) error { + return p.Consumer.OpenInbound(ctx) +} + +// Receive implements Receiver.Receive +func (p *Protocol) Receive(ctx context.Context) (binding.Message, error) { + return p.Consumer.Receive(ctx) +} + +// Close implements Closer.Close +func (p *Protocol) Close(ctx context.Context) error { + if p.connOwned { + defer p.conn.Close() + } + + if err := p.Consumer.Close(ctx); err != nil { + return err + } + + if err := p.Sender.Close(ctx); err != nil { + return err + } + + return nil +} + +var _ protocol.Receiver = (*Protocol)(nil) +var _ protocol.Sender = (*Protocol)(nil) +var _ protocol.Opener = (*Protocol)(nil) +var _ protocol.Closer = (*Protocol)(nil) diff --git a/protocol/nats_jetstream/v3/receiver.go b/protocol/nats_jetstream/v3/receiver.go new file mode 100644 index 000000000..91becfd7e --- /dev/null +++ b/protocol/nats_jetstream/v3/receiver.go @@ -0,0 +1,219 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "context" + "io" + "sync" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/protocol" +) + +type msgErr struct { + msg binding.Message +} + +type receiver struct { + incoming chan msgErr +} + +// newReceiver creates a new protocol.Receiver responsible for receiving messages. +func newReceiver() *receiver { + return &receiver{ + incoming: make(chan msgErr), + } +} + +// MsgHandler implements nats.MsgHandler and publishes messages onto our internal incoming channel to be delivered +// via r.Receive(ctx) +func (r *receiver) MsgHandler(msg jetstream.Msg) { + r.incoming <- msgErr{msg: NewMessage(msg)} +} + +// Receive implements Receiver.Receive. +func (r *receiver) Receive(ctx context.Context) (binding.Message, error) { + select { + case msgErr, ok := <-r.incoming: + if !ok { + return nil, io.EOF + } + return msgErr.msg, nil + case <-ctx.Done(): + return nil, io.EOF + } +} + +type consumer struct { + receiver + + conn *nats.Conn + jetSteamOpts []jetstream.JetStreamOpt + jetStream jetstream.JetStream + consumerConfig *jetstream.ConsumerConfig + orderedConsumerConfig *jetstream.OrderedConsumerConfig + pullConsumeOpts []jetstream.PullConsumeOpt + jetstreamConsumer jetstream.Consumer + + subMtx sync.Mutex + internalClose chan struct{} + connOwned bool +} + +// NewConsumer consumes from filterSubjects given in ConsumerConfig or OrderedConsumerConfig +// embedded in ConsumerOption. +// See: WithConsumerConfig(...) and WithOrderedConsumerConfig(...) +func NewConsumer(ctx context.Context, url string, natsOpts []nats.Option, opts ...ProtocolOption) (*consumer, error) { + conn, err := nats.Connect(url, natsOpts...) + if err != nil { + return nil, err + } + + c, err := NewConsumerFromConn(ctx, conn, opts...) + if err != nil { + conn.Close() + return nil, err + } + + c.connOwned = true + + return c, err +} + +// NewConsumerFromConn consumes from filterSubjects given in ConsumerConfig or OrderedConsumerConfig +// embedded in ConsumerOption. See: WithConsumerConfig(...) and WithOrderedConsumerConfig(...) +func NewConsumerFromConn(ctx context.Context, conn *nats.Conn, opts ...ProtocolOption) (*consumer, error) { + c := &consumer{ + receiver: *newReceiver(), + conn: conn, + internalClose: make(chan struct{}, 1), + } + + if err := c.applyOptions(opts...); err != nil { + return nil, err + } + + var err error + if c.jetStream, err = jetstream.New(conn, c.jetSteamOpts...); err != nil { + return nil, err + } + + return c, nil +} + +// OpenInbound implements Opener.OpenInbound +func (c *consumer) OpenInbound(ctx context.Context) error { + c.subMtx.Lock() + defer c.subMtx.Unlock() + + var consumeContext jetstream.ConsumeContext + var err error + if err = c.createConsumer(ctx); err != nil { + return err + } + if consumeContext, err = c.jetstreamConsumer.Consume(c.MsgHandler, c.pullConsumeOpts...); err != nil { + return err + } + + // Wait until external or internal context done + select { + case <-ctx.Done(): + case <-c.internalClose: + } + + // Finish to consume messages in the queue and close the subscription + if consumeContext != nil { + consumeContext.Drain() + } + return nil +} + +// Close implements Closer.Close +func (c *consumer) Close(ctx context.Context) error { + // 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 + c.internalClose <- struct{}{} + c.subMtx.Lock() + defer c.subMtx.Unlock() + + if c.connOwned { + c.conn.Close() + } + + close(c.internalClose) + + return nil +} + +func (c *consumer) applyOptions(opts ...ProtocolOption) error { + p := &Protocol{Consumer: c} + for _, fn := range opts { + if err := fn(p); err != nil { + return err + } + } + return nil +} + +// createConsumer creates a consumer based on the configured consumer config +func (c *consumer) createConsumer(ctx context.Context) error { + var err error + var stream string + if stream, err = c.getStreamFromSubjects(ctx); err != nil { + return err + } + var consumerErr error + if c.consumerConfig != nil { + c.jetstreamConsumer, consumerErr = c.jetStream.CreateOrUpdateConsumer(ctx, stream, *c.consumerConfig) + } else if c.orderedConsumerConfig != nil { + c.jetstreamConsumer, consumerErr = c.jetStream.OrderedConsumer(ctx, stream, *c.orderedConsumerConfig) + } else { + return ErrNoConsumerConfig + } + return consumerErr +} + +// getStreamFromSubjects finds the unique stream for the set of filter subjects +// If more than one stream is found, returns ErrMoreThanOneStream +func (c *consumer) getStreamFromSubjects(ctx context.Context) (string, error) { + var subjects []string + if c.consumerConfig != nil && c.consumerConfig.FilterSubject != "" { + subjects = []string{c.consumerConfig.FilterSubject} + } + if c.consumerConfig != nil && len(c.consumerConfig.FilterSubjects) > 0 { + subjects = c.consumerConfig.FilterSubjects + } + if c.orderedConsumerConfig != nil && len(c.orderedConsumerConfig.FilterSubjects) > 0 { + subjects = c.orderedConsumerConfig.FilterSubjects + } + if len(subjects) == 0 { + return "", ErrNoFilterSubjects + } + var finalStream string + for i, subject := range subjects { + currentStream, err := c.jetStream.StreamNameBySubject(ctx, subject) + if err != nil { + return "", err + } + if i == 0 { + finalStream = currentStream + continue + } + if finalStream != currentStream { + return "", ErrMoreThanOneStream + } + } + return finalStream, nil +} + +var _ protocol.Opener = (*consumer)(nil) +var _ protocol.Receiver = (*consumer)(nil) +var _ protocol.Closer = (*consumer)(nil) diff --git a/protocol/nats_jetstream/v3/sender.go b/protocol/nats_jetstream/v3/sender.go new file mode 100644 index 000000000..5c5820e11 --- /dev/null +++ b/protocol/nats_jetstream/v3/sender.go @@ -0,0 +1,122 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "bytes" + "context" + "fmt" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/protocol" +) + +type sender struct { + jetSteamOpts []jetstream.JetStreamOpt + jetStream jetstream.JetStream + publishOpts []jetstream.PublishOpt + conn *nats.Conn + subject string + connOwned bool +} + +// NewSender creates a new protocol.Sender responsible for opening and closing the NATS connection +func NewSender(ctx context.Context, url, subject string, natsOpts []nats.Option, opts ...ProtocolOption) (*sender, error) { + conn, err := nats.Connect(url, natsOpts...) + if err != nil { + return nil, err + } + + s, err := NewSenderFromConn(ctx, conn, subject, opts...) + if err != nil { + conn.Close() + return nil, err + } + + s.connOwned = true + + return s, nil +} + +// NewSenderFromConn creates a new protocol.Sender which leaves responsibility for opening and closing the NATS +// connection to the caller +func NewSenderFromConn(ctx context.Context, conn *nats.Conn, subject string, opts ...ProtocolOption) (*sender, error) { + s := &sender{ + conn: conn, + subject: subject, + } + + if err := s.applyOptions(opts...); err != nil { + return nil, err + } + + var err error + if s.jetStream, err = jetstream.New(conn, s.jetSteamOpts...); err != nil { + return nil, err + } + + return s, nil +} + +// Close implements Sender.Sender +// Sender sends messages. +func (s *sender) Send(ctx context.Context, in binding.Message, transformers ...binding.Transformer) (err error) { + defer func() { + if err2 := in.Finish(err); err2 != nil { + if err == nil { + err = err2 + } else { + err = fmt.Errorf("failed to call in.Finish() when error already occurred: %s: %w", err2.Error(), err) + } + } + }() + + if _, err = s.jetStream.StreamNameBySubject(ctx, s.subject); err != nil { + return err + } + + writer := new(bytes.Buffer) + header, err := WriteMsg(ctx, in, writer, transformers...) + if err != nil { + return err + } + + natsMsg := &nats.Msg{ + Subject: s.subject, + Data: writer.Bytes(), + Header: header, + } + + _, err = s.jetStream.PublishMsg(ctx, natsMsg, s.publishOpts...) + + return err +} + +// Close implements Closer.Close +// This method only closes the connection if the Sender opened it +func (s *sender) Close(_ context.Context) error { + if s.connOwned { + s.conn.Close() + } + + return nil +} + +func (s *sender) applyOptions(opts ...ProtocolOption) error { + p := &Protocol{Sender: s} + for _, fn := range opts { + if err := fn(p); err != nil { + return err + } + } + return nil +} + +var _ protocol.Sender = (*sender)(nil) +var _ protocol.Closer = (*Protocol)(nil) diff --git a/protocol/nats_jetstream/v3/write_message.go b/protocol/nats_jetstream/v3/write_message.go new file mode 100644 index 000000000..2fe3c598c --- /dev/null +++ b/protocol/nats_jetstream/v3/write_message.go @@ -0,0 +1,98 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "context" + "fmt" + "io" + "time" + + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/binding/format" + "github.com/cloudevents/sdk-go/v2/binding/spec" + "github.com/nats-io/nats.go" +) + +// WriteMsg fills the provided writer with the bindings.Message m. +// Using context you can tweak the encoding processing (more details on binding.Write documentation). +// The nats.Header returned is not deep-copied. The header values should be deep-copied to an event object. +func WriteMsg(ctx context.Context, m binding.Message, writer io.ReaderFrom, transformers ...binding.Transformer) (nats.Header, error) { + structuredWriter := &natsMessageWriter{writer} + binaryWriter := &natsBinaryMessageWriter{ReaderFrom: writer} + + _, err := binding.Write( + ctx, + m, + structuredWriter, + binaryWriter, + transformers..., + ) + natsHeader := binaryWriter.header + + return natsHeader, err +} + +type natsMessageWriter struct { + io.ReaderFrom +} + +// StructuredWriter implements StructuredWriter.SetStructuredEvent +func (w *natsMessageWriter) SetStructuredEvent(_ context.Context, _ format.Format, event io.Reader) error { + if _, err := w.ReadFrom(event); err != nil { + return err + } + + return nil +} + +var _ binding.StructuredWriter = (*natsMessageWriter)(nil) // Test it conforms to the interface + +type natsBinaryMessageWriter struct { + io.ReaderFrom + header nats.Header +} + +// SetAttribute implements MessageMetadataWriter.SetAttribute +func (w *natsBinaryMessageWriter) SetAttribute(attribute spec.Attribute, value interface{}) error { + prefixedName := withPrefix(attribute.Name()) + convertedValue := fmt.Sprint(value) + switch attribute.Kind().String() { + case spec.Time.String(): + timeValue := value.(time.Time) + convertedValue = timeValue.Format(time.RFC3339Nano) + } + w.header.Set(prefixedName, convertedValue) + return nil +} + +// SetExtension implements MessageMetadataWriter.SetExtension +func (w *natsBinaryMessageWriter) SetExtension(name string, value interface{}) error { + prefixedName := withPrefix(name) + convertedValue := fmt.Sprint(value) + w.header.Set(prefixedName, convertedValue) + return nil +} + +// Start implements BinaryWriter.Start +func (w *natsBinaryMessageWriter) Start(ctx context.Context) error { + w.header = nats.Header{} + return nil +} + +// SetData implements BinaryWriter.SetData +func (w *natsBinaryMessageWriter) SetData(data io.Reader) error { + if _, err := w.ReadFrom(data); err != nil { + return err + } + + return nil +} + +// End implements BinaryWriter.End +func (w *natsBinaryMessageWriter) End(ctx context.Context) error { + return nil +} diff --git a/samples/nats_jetstream/v3/go.mod b/samples/nats_jetstream/v3/go.mod new file mode 100644 index 000000000..9c8e89cec --- /dev/null +++ b/samples/nats_jetstream/v3/go.mod @@ -0,0 +1,31 @@ +module github.com/cloudevents/sdk-go/samples/nats_jetstream/v3 + +go 1.18 + +require ( + github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3 v3.0.0 + github.com/cloudevents/sdk-go/v2 v2.15.2 + github.com/google/uuid v1.1.1 + github.com/kelseyhightower/envconfig v1.4.0 + github.com/nats-io/nats.go v1.37.0 +) + +require ( + github.com/json-iterator/go v1.1.12 // indirect + github.com/klauspost/compress v1.17.9 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/nats-io/nkeys v0.4.7 // indirect + github.com/nats-io/nuid v1.0.1 // indirect + go.uber.org/atomic v1.4.0 // indirect + go.uber.org/multierr v1.1.0 // indirect + go.uber.org/zap v1.10.0 // indirect + golang.org/x/crypto v0.27.0 // indirect + golang.org/x/sys v0.25.0 // indirect + golang.org/x/text v0.18.0 // indirect + golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect +) + +replace github.com/cloudevents/sdk-go/v2 => ../../../v2 + +replace github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3 => ./../../../protocol/nats_jetstream/v3 diff --git a/samples/nats_jetstream/v3/go.sum b/samples/nats_jetstream/v3/go.sum new file mode 100644 index 000000000..986929c03 --- /dev/null +++ b/samples/nats_jetstream/v3/go.sum @@ -0,0 +1,47 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/google/go-cmp v0.5.0 h1:/QaMHBdZ26BB3SSst0Iwl10Epc+xhTquomWX0oZEB6w= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY= +github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/kelseyhightower/envconfig v1.4.0 h1:Im6hONhd3pLkfDFsbRgu68RDNkGF1r3dvMUtDTo2cv8= +github.com/kelseyhightower/envconfig v1.4.0/go.mod h1:cccZRl6mQpaq41TPp5QxidR+Sa3axMbJDNb//FQX6Gg= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE= +github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8= +github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI= +github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= +go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU= +go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= +go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI= +go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= +go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM= +go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q= +golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A= +golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70= +golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34= +golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224= +golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= +golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 h1:GZokNIeuVkl3aZHJchRrr13WCsols02MLUcz1U9is6M= +golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/samples/nats_jetstream/v3/receiver/main.go b/samples/nats_jetstream/v3/receiver/main.go new file mode 100644 index 000000000..2328bb407 --- /dev/null +++ b/samples/nats_jetstream/v3/receiver/main.go @@ -0,0 +1,61 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package main + +import ( + "context" + "fmt" + "log" + + "github.com/nats-io/nats.go/jetstream" + + cejsm "github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3" + cloudevents "github.com/cloudevents/sdk-go/v2" +) + +func main() { + ctx := context.Background() + + natsURL := "nats://localhost:4222" + natsSubject := "sample" + + consumerOpt := cejsm.WithConsumerConfig(&jetstream.ConsumerConfig{FilterSubjects: []string{natsSubject}}) + consumer, err := cejsm.NewConsumer(ctx, natsURL, cejsm.NatsOptions(), consumerOpt) + if err != nil { + log.Fatalf("failed to create nats protocol, %s", err.Error()) + } + + defer consumer.Close(ctx) + + c, err := cloudevents.NewClient(consumer) + if err != nil { + log.Fatalf("failed to create client, %s", err.Error()) + } + + for { + if err := c.StartReceiver(ctx, receive); err != nil { + log.Printf("failed to start nats receiver, %s", err.Error()) + } + } +} + +type Example struct { + Sequence int `json:"id"` + Message string `json:"message"` +} + +func receive(ctx context.Context, event cloudevents.Event) error { + fmt.Printf("Got Event Context: %+v\n", event.Context) + + data := &Example{} + if err := event.DataAs(data); err != nil { + fmt.Printf("Got Data Error: %s\n", err.Error()) + } + fmt.Printf("Got Data: %+v\n", data) + + fmt.Printf("----------------------------\n") + return nil +} diff --git a/samples/nats_jetstream/v3/sender/main.go b/samples/nats_jetstream/v3/sender/main.go new file mode 100644 index 000000000..8434435d1 --- /dev/null +++ b/samples/nats_jetstream/v3/sender/main.go @@ -0,0 +1,80 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package main + +import ( + "context" + "fmt" + "log" + "time" + + "github.com/google/uuid" + "github.com/kelseyhightower/envconfig" + + cejsm "github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3" + cloudevents "github.com/cloudevents/sdk-go/v2" +) + +const ( + count = 10 +) + +type envConfig struct { + // NATSServer URL to connect to the nats server. + NATSServer string `envconfig:"NATS_SERVER" default:"http://localhost:4222" required:"true"` + + // Subject is the nats subject to publish cloudevents on. + Subject string `envconfig:"SUBJECT" default:"sample" required:"true"` +} + +type Example struct { + Sequence int `json:"id"` + Message string `json:"message"` +} + +func main() { + var env envConfig + if err := envconfig.Process("", &env); err != nil { + log.Fatalf("Failed to process env var: %s", err) + } + + natsURL := "nats://localhost:4222" + natsSubject := "sample" + + ctx := context.Background() + sender, err := cejsm.NewSender(ctx, natsURL, natsSubject, cejsm.NatsOptions(), nil) + if err != nil { + log.Fatalf("Failed to create nats protocol, %s", err.Error()) + } + + defer sender.Close(context.Background()) + + c, err := cloudevents.NewClient(sender) + if err != nil { + log.Fatalf("Failed to create client, %s", err.Error()) + } + + for _, contentType := range []string{"application/json", "application/xml"} { + for i := 0; i < count; i++ { + e := cloudevents.NewEvent() + e.SetID(uuid.New().String()) + e.SetType("com.cloudevents.sample.sent") + e.SetTime(time.Now()) + e.SetSource("https://github.com/cloudevents/sdk-go/v2/samples/sender") + _ = e.SetData(contentType, &Example{ + Sequence: i, + Message: fmt.Sprintf("Hello, %s!", contentType), + }) + + if result := c.Send(context.Background(), e); cloudevents.IsUndelivered(result) { + log.Printf("failed to send: %v", err) + } else { + log.Printf("sent: %d, accepted: %t", i, cloudevents.IsACK(result)) + } + time.Sleep(100 * time.Millisecond) + } + } +} diff --git a/test/integration/go.mod b/test/integration/go.mod index b80046249..2ffc9a039 100644 --- a/test/integration/go.mod +++ b/test/integration/go.mod @@ -14,6 +14,8 @@ replace github.com/cloudevents/sdk-go/protocol/nats/v2 => ../../protocol/nats/v2 replace github.com/cloudevents/sdk-go/protocol/nats_jetstream/v2 => ../../protocol/nats_jetstream/v2 +replace github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3 => ../../protocol/nats_jetstream/v3 + replace github.com/cloudevents/sdk-go/protocol/kafka_sarama/v2 => ../../protocol/kafka_sarama/v2 replace github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 => ../../protocol/mqtt_paho/v2 @@ -29,13 +31,14 @@ require ( github.com/cloudevents/sdk-go/protocol/mqtt_paho/v2 v2.0.0-00010101000000-000000000000 github.com/cloudevents/sdk-go/protocol/nats/v2 v2.5.0 github.com/cloudevents/sdk-go/protocol/nats_jetstream/v2 v2.0.0-00010101000000-000000000000 + github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3 v3.0.0-00010101000000-000000000000 github.com/cloudevents/sdk-go/protocol/stan/v2 v2.5.0 github.com/cloudevents/sdk-go/v2 v2.15.2 github.com/confluentinc/confluent-kafka-go/v2 v2.3.0 github.com/eclipse/paho.golang v0.12.0 github.com/google/go-cmp v0.6.0 github.com/google/uuid v1.3.0 - github.com/nats-io/nats.go v1.31.0 + github.com/nats-io/nats.go v1.37.0 github.com/nats-io/stan.go v0.10.4 github.com/stretchr/testify v1.8.4 go.uber.org/atomic v1.4.0 @@ -61,14 +64,14 @@ require ( github.com/jcmturner/gokrb5/v8 v8.4.3 // indirect github.com/jcmturner/rpc/v2 v2.0.3 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.17.2 // indirect + github.com/klauspost/compress v1.17.9 // indirect github.com/mattn/go-colorable v0.1.13 // indirect github.com/mattn/go-isatty v0.0.16 // indirect github.com/minio/highwayhash v1.0.2 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 // indirect - github.com/nats-io/nkeys v0.4.6 // indirect + github.com/nats-io/nkeys v0.4.7 // indirect github.com/nats-io/nuid v1.0.1 // indirect github.com/pierrec/lz4/v4 v4.1.17 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect @@ -77,10 +80,11 @@ require ( go.etcd.io/bbolt v1.3.6 // indirect go.uber.org/multierr v1.1.0 // indirect go.uber.org/zap v1.10.0 // indirect - golang.org/x/crypto v0.21.0 // indirect + golang.org/x/crypto v0.27.0 // indirect golang.org/x/net v0.23.0 // indirect - golang.org/x/sync v0.4.0 // indirect - golang.org/x/sys v0.18.0 // indirect + golang.org/x/sync v0.8.0 // indirect + golang.org/x/sys v0.25.0 // indirect + golang.org/x/text v0.18.0 // indirect golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/test/integration/go.sum b/test/integration/go.sum index 422913dd6..b0115c807 100644 --- a/test/integration/go.sum +++ b/test/integration/go.sum @@ -90,8 +90,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= -github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4= -github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA= +github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw= github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/magiconair/properties v1.8.6 h1:5ibWZ6iY0NctNGWo87LalDlEZ6R41TqbbDamhfG/Qzo= @@ -117,13 +117,13 @@ github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296 h1:vU9tpM3apjYlLLeY23zRWJ9Zktr5jp+mloR942LEOpY= github.com/nats-io/jwt/v2 v2.2.1-0.20220113022732-58e87895b296/go.mod h1:0tqz9Hlu6bCBFLWAASKhE5vUA4c24L9KPUUgvwumE/k= github.com/nats-io/nats-server/v2 v2.9.23 h1:6Wj6H6QpP9FMlpCyWUaNu2yeZ/qGj+mdRkZ1wbikExU= -github.com/nats-io/nats-streaming-server v0.24.3 h1:uZez8jBkXscua++jaDsK7DhpSAkizdetar6yWbPMRco= +github.com/nats-io/nats-streaming-server v0.24.6 h1:iIZXuPSznnYkiy0P3L0AP9zEN9Etp+tITbbX1KKeq4Q= github.com/nats-io/nats.go v1.22.1/go.mod h1:tLqubohF7t4z3du1QDPYJIQQyhb4wl6DhjxEajSI7UA= -github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E= -github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8= +github.com/nats-io/nats.go v1.37.0 h1:07rauXbVnnJvv1gfIyghFEo6lUcYRY0WXc3x7x0vUxE= +github.com/nats-io/nats.go v1.37.0/go.mod h1:Ubdu4Nh9exXdSz0RVWRFBbRfrbSxOYd26oF0wkWclB8= github.com/nats-io/nkeys v0.3.0/go.mod h1:gvUNGjVcM2IPr5rCsRsC6Wb3Hr2CQAm08dsxtV6A5y4= -github.com/nats-io/nkeys v0.4.6 h1:IzVe95ru2CT6ta874rt9saQRkWfe2nFj1NtvYSLqMzY= -github.com/nats-io/nkeys v0.4.6/go.mod h1:4DxZNzenSVd1cYQoAa8948QY3QDjrHfcfVADymtkpts= +github.com/nats-io/nkeys v0.4.7 h1:RwNJbbIdYCoClSDNY7QVKZlyb/wfT6ugvFCiKy6vDvI= +github.com/nats-io/nkeys v0.4.7/go.mod h1:kqXRgRDPlGy7nGaEDMuYzmiJCIAAWDK0IMBtDmGD0nc= github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/nats-io/stan.go v0.10.4 h1:19GS/eD1SeQJaVkeM9EkvEYattnvnWrZ3wkSWSw4uXw= @@ -179,8 +179,8 @@ golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b/go.mod h1:T9bdIzuCu7OtxOm golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU= -golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA= -golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs= +golang.org/x/crypto v0.27.0 h1:GXm2NjJrPaiv/h1tb2UH8QfgC/hOf/+z0p6PT8o1w7A= +golang.org/x/crypto v0.27.0/go.mod h1:1Xngt8kV6Dvbssa53Ziq6Eqn0HqbZi5Z6R0ZpwQzt70= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= @@ -203,8 +203,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.4.0 h1:zxkM55ReGkDlKSM+Fu41A+zmbZuaPVbGMzvvdUPznYQ= -golang.org/x/sync v0.4.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= +golang.org/x/sync v0.8.0 h1:3NFvSEYkUoMifnESzZl15y791HH1qU2xm6eCJU5ZPXQ= +golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -219,8 +219,8 @@ golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.18.0 h1:DBdB3niSjOA/O0blCZBqDefyWNYveAYMNF1Wum0DYQ4= -golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= +golang.org/x/sys v0.25.0 h1:r+8e+loiHxRqhXVl6ML1nO3l1+oFoWbnlu2Ehimmi34= +golang.org/x/sys v0.25.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= golang.org/x/term v0.4.0/go.mod h1:9P2UbLfCdcvo3p/nzKvsmas4TnlujnuoV9hGgYzW1lQ= @@ -229,6 +229,8 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= +golang.org/x/text v0.18.0 h1:XvMDiNzPAl0jr17s6W9lcaIhGUfUORdGCNsuLmPG224= +golang.org/x/text v0.18.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY= golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 h1:GZokNIeuVkl3aZHJchRrr13WCsols02MLUcz1U9is6M= golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= diff --git a/test/integration/nats_jetstream/v3/nats_test.go b/test/integration/nats_jetstream/v3/nats_test.go new file mode 100644 index 000000000..f49ead06c --- /dev/null +++ b/test/integration/nats_jetstream/v3/nats_test.go @@ -0,0 +1,162 @@ +/* + Copyright 2021 The CloudEvents Authors + SPDX-License-Identifier: Apache-2.0 +*/ + +package nats_jetstream + +import ( + "context" + "os" + "testing" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + ce_nats "github.com/cloudevents/sdk-go/protocol/nats_jetstream/v3" + "github.com/cloudevents/sdk-go/v2/binding" + "github.com/cloudevents/sdk-go/v2/event" + bindings "github.com/cloudevents/sdk-go/v2/protocol" + "github.com/cloudevents/sdk-go/v2/protocol/test" + . "github.com/cloudevents/sdk-go/v2/test" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + . "github.com/cloudevents/sdk-go/v2/binding/test" +) + +func TestSendReceiveStructuredAndBinary(t *testing.T) { + conn := testConn(t) + defer conn.Close() + + type args struct { + opts []ce_nats.ProtocolOption + bindingEncoding binding.Encoding + consumerConfig any + } + tests := []struct { + name string + args args + }{ + { + name: "regular consumer - structured", + args: args{ + consumerConfig: &jetstream.ConsumerConfig{}, + bindingEncoding: binding.EncodingStructured, + }, + }, + { + name: "ordered consumer - structured", + args: args{ + consumerConfig: &jetstream.OrderedConsumerConfig{}, + bindingEncoding: binding.EncodingStructured, + }, + }, + { + name: "regular consumer - binary", + args: args{ + consumerConfig: &jetstream.ConsumerConfig{}, + bindingEncoding: binding.EncodingBinary, + }, + }, { + name: "ordered consumer - binary", + args: args{ + consumerConfig: &jetstream.OrderedConsumerConfig{}, + bindingEncoding: binding.EncodingBinary, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + cleanup, s, r := testProtocol(ctx, t, conn, tt.args.consumerConfig, tt.args.opts...) + defer cleanup() + EachEvent(t, Events(), func(t *testing.T, eventIn event.Event) { + eventIn = ConvertEventExtensionsToString(t, eventIn) + + var in binding.Message + switch tt.args.bindingEncoding { + case binding.EncodingStructured: + in = MustCreateMockStructuredMessage(t, eventIn) + case binding.EncodingBinary: + in = MustCreateMockBinaryMessage(eventIn) + } + + test.SendReceive(t, binding.WithPreferredEventEncoding(context.TODO(), tt.args.bindingEncoding), in, s, r, func(out binding.Message) { + eventOut := MustToEvent(t, context.Background(), out) + assert.Equal(t, tt.args.bindingEncoding, out.ReadEncoding()) + AssertEventEquals(t, eventIn, ConvertEventExtensionsToString(t, eventOut)) + }) + }) + }) + } +} + +func testConn(t testing.TB) *nats.Conn { + t.Helper() + // STAN connections actually connect to NATS, so the env var is named appropriately + s := os.Getenv("TEST_NATS_SERVER") + if s == "" { + s = "nats://localhost:4222" + } + + conn, err := nats.Connect(s) + if err != nil { + t.Skipf("Cannot create STAN client to NATS server [%s]: %v", s, err) + } + + return conn +} + +func testProtocol(ctx context.Context, t testing.TB, natsConn *nats.Conn, consumerConfig any, opts ...ce_nats.ProtocolOption) (func(), bindings.Sender, + bindings.Receiver) { + // STAN connections actually connect to NATS, so the env var is named appropriately + s := os.Getenv("TEST_NATS_SERVER") + if s == "" { + s = "nats://localhost:4222" + } + + stream := "test-ce-client-" + uuid.New().String() + subject := stream + ".test" + + var js jetstream.JetStream + var err error + js, err = jetstream.New(natsConn) + require.NoError(t, err) + + streamConfig := jetstream.StreamConfig{Name: stream, Subjects: []string{subject}} + _, err = js.CreateOrUpdateStream(ctx, streamConfig) + require.NoError(t, err) + + if normalConsumerConfig, ok := consumerConfig.(*jetstream.ConsumerConfig); ok { + normalConsumerConfig.FilterSubjects = []string{subject} + opts = append(opts, ce_nats.WithConsumerConfig(normalConsumerConfig)) + } + if orderedConsumerConfig, ok := consumerConfig.(*jetstream.OrderedConsumerConfig); ok { + orderedConsumerConfig.FilterSubjects = []string{subject} + opts = append(opts, ce_nats.WithOrderedConsumerConfig(orderedConsumerConfig)) + } + + // use NewProtocol rather than individual Consumer and Sender since this gives us more coverage + p, err := ce_nats.NewProtocol(ctx, s, subject, ce_nats.NatsOptions(), opts...) + require.NoError(t, err) + + go func() { + require.NoError(t, p.OpenInbound(context.TODO())) + }() + + return func() { + err = p.Close(context.TODO()) + require.NoError(t, err) + }, p.Sender, p.Consumer +} + +func BenchmarkSendReceive(b *testing.B) { + ctx := context.Background() + conn := testConn(b) + defer conn.Close() + c, s, r := testProtocol(ctx, b, conn, &jetstream.ConsumerConfig{}) + defer c() // Cleanup + test.BenchmarkSendReceive(b, s, r) +}