generated from kyma-project/template-repository
-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rely on the NATS cluster connection status when reporting the Eventin…
…g CR status
- Loading branch information
1 parent
0df1b52
commit 31c731c
Showing
22 changed files
with
774 additions
and
126 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func Test_getNATSConnectionBuilder(t *testing.T) { | ||
| natsConnectionBuilder, err := getNATSConnectionBuilder() | ||
| require.NoError(t, err) | ||
| require.NotNil(t, natsConnectionBuilder) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package nats | ||
|
|
||
| import ( | ||
| "strings" | ||
|
|
||
| natsio "github.com/nats-io/nats.go" | ||
|
|
||
| "github.com/kyma-project/eventing-manager/internal/connection/nats/errors" | ||
| ) | ||
|
|
||
| type Builder interface { | ||
| Build() NATS | ||
| } | ||
|
|
||
| type ConnectionBuilder struct { | ||
| url string | ||
| opts []natsio.Option | ||
| } | ||
|
|
||
| func NewBuilder(url, name string, opts ...natsio.Option) (*ConnectionBuilder, error) { | ||
| if len(strings.TrimSpace(url)) == 0 { | ||
| return nil, errors.ErrEmptyConnectionURL | ||
| } | ||
|
|
||
| if len(strings.TrimSpace(name)) == 0 { | ||
| return nil, errors.ErrEmptyConnectionName | ||
| } | ||
|
|
||
| opts = append(opts, natsio.Name(name)) // enforce configuring the connection name | ||
| return &ConnectionBuilder{url: url, opts: opts}, nil | ||
| } | ||
|
|
||
| func (b *ConnectionBuilder) Build() NATS { | ||
| return &connection{ | ||
| url: b.url, | ||
| conn: nil, | ||
| opts: b.opts, | ||
| reconnectHandlerRegistered: false, | ||
| disconnectErrHandlerRegistered: false, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,150 @@ | ||
| package nats | ||
|
|
||
| import ( | ||
| "reflect" | ||
| "testing" | ||
| "time" | ||
|
|
||
| natsio "github.com/nats-io/nats.go" | ||
| "github.com/stretchr/testify/require" | ||
|
|
||
| "github.com/kyma-project/eventing-manager/internal/connection/nats/errors" | ||
| ) | ||
|
|
||
| func TestNewBuilder(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| // given | ||
| const ( | ||
| url = "url" | ||
| name = "name" | ||
| maxReconnects = 10 | ||
| retryOnFailedConnect = true | ||
| reconnectWait = time.Minute | ||
| ) | ||
|
|
||
| type args struct { | ||
| url string | ||
| name string | ||
| opts []natsio.Option | ||
| } | ||
| tests := []struct { | ||
| name string | ||
| args args | ||
| wantBuilder *ConnectionBuilder | ||
| wantErr error | ||
| }{ | ||
| { | ||
| name: "should return an error if the URL is empty", | ||
| args: args{ | ||
| url: "", | ||
| name: name, | ||
| opts: nil, | ||
| }, | ||
| wantBuilder: nil, | ||
| wantErr: errors.ErrEmptyConnectionURL, | ||
| }, | ||
| { | ||
| name: "should return an error if the name is empty", | ||
| args: args{ | ||
| url: url, | ||
| name: "", | ||
| opts: nil, | ||
| }, | ||
| wantBuilder: nil, | ||
| wantErr: errors.ErrEmptyConnectionName, | ||
| }, | ||
| { | ||
| name: "should return a connection builder instance with no errors if there are no given options", | ||
| args: args{ | ||
| url: url, | ||
| name: name, | ||
| opts: nil, // no given options | ||
| }, | ||
| wantBuilder: &ConnectionBuilder{ | ||
| url: url, | ||
| opts: []natsio.Option{ | ||
| natsio.Name(name), | ||
| }, | ||
| }, | ||
| wantErr: nil, | ||
| }, | ||
| { | ||
| name: "should return a connection builder instance with no errors if there are given options", | ||
| args: args{ | ||
| url: url, | ||
| name: name, | ||
| opts: []natsio.Option{ | ||
| natsio.MaxReconnects(maxReconnects), | ||
| natsio.RetryOnFailedConnect(retryOnFailedConnect), | ||
| natsio.ReconnectWait(reconnectWait), | ||
| }, | ||
| }, | ||
| wantBuilder: &ConnectionBuilder{ | ||
| url: url, | ||
| opts: []natsio.Option{ | ||
| natsio.Name(name), | ||
| natsio.MaxReconnects(10), | ||
| natsio.RetryOnFailedConnect(true), | ||
| natsio.ReconnectWait(time.Minute), | ||
| }, | ||
| }, | ||
| wantErr: nil, | ||
| }, | ||
| } | ||
| for _, tt := range tests { | ||
| ttc := tt | ||
|
|
||
| t.Run(ttc.name, func(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| // when | ||
| gotBuilder, gotErr := NewBuilder(ttc.args.url, ttc.args.name, ttc.args.opts...) | ||
|
|
||
| // then | ||
| require.Equal(t, ttc.wantErr, gotErr) | ||
| require.True(t, builderDeepEqual(t, ttc.wantBuilder, gotBuilder)) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| func TestConnectionBuilder_Build(t *testing.T) { | ||
| b := ConnectionBuilder{ | ||
| url: "url", | ||
| opts: []natsio.Option{ | ||
| natsio.Name("name"), | ||
| natsio.MaxReconnects(10), | ||
| natsio.RetryOnFailedConnect(true), | ||
| natsio.ReconnectWait(time.Minute), | ||
| }, | ||
| } | ||
| require.NotNil(t, b.Build()) | ||
| } | ||
|
|
||
| func builderDeepEqual(t *testing.T, a, b *ConnectionBuilder) bool { | ||
| t.Helper() | ||
|
|
||
| if a == b { | ||
| return true | ||
| } | ||
|
|
||
| if a.url != b.url { | ||
| return false | ||
| } | ||
|
|
||
| if len(a.opts) != len(b.opts) { | ||
| return false | ||
| } | ||
|
|
||
| aOpts := &natsio.Options{} | ||
| for _, opt := range a.opts { | ||
| require.NoError(t, opt(aOpts)) | ||
| } | ||
|
|
||
| bOpts := &natsio.Options{} | ||
| for _, opt := range b.opts { | ||
| require.NoError(t, opt(bOpts)) | ||
| } | ||
|
|
||
| return reflect.DeepEqual(aOpts, bOpts) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,66 @@ | ||
| package nats | ||
|
|
||
| import ( | ||
| natsio "github.com/nats-io/nats.go" | ||
|
|
||
| natsconnectionerrors "github.com/kyma-project/eventing-manager/internal/connection/nats/errors" | ||
| ) | ||
|
|
||
| // compile-time check. | ||
| var _ NATS = &connection{} | ||
|
|
||
| // connection represents a NATS connection. | ||
| type connection struct { | ||
| url string | ||
| conn *natsio.Conn | ||
| opts []natsio.Option | ||
| reconnectHandlerRegistered bool | ||
| disconnectErrHandlerRegistered bool | ||
| } | ||
|
|
||
| func (c *connection) Connect() error { | ||
| if c.isConnected() { | ||
| return nil | ||
| } | ||
|
|
||
| var err error | ||
| if c.conn, err = natsio.Connect(c.url, c.opts...); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| if c.isConnected() { | ||
| return nil | ||
| } | ||
|
|
||
| return natsconnectionerrors.ErrCannotConnect | ||
| } | ||
|
|
||
| func (c *connection) Disconnect() { | ||
| if c.conn == nil || c.conn.IsClosed() { | ||
| return | ||
| } | ||
| c.conn.Close() | ||
| } | ||
|
|
||
| func (c *connection) isConnected() bool { | ||
| if c.conn == nil { | ||
| return false | ||
| } | ||
| return c.conn.IsConnected() | ||
| } | ||
|
|
||
| func (c *connection) RegisterReconnectHandlerIfNotRegistered(handler natsio.ConnHandler) { | ||
| if c.conn == nil || c.reconnectHandlerRegistered { | ||
| return | ||
| } | ||
| c.conn.SetReconnectHandler(handler) | ||
| c.reconnectHandlerRegistered = true | ||
| } | ||
|
|
||
| func (c *connection) RegisterDisconnectErrHandlerIfNotRegistered(handler natsio.ConnErrHandler) { | ||
| if c.conn == nil || c.disconnectErrHandlerRegistered { | ||
| return | ||
| } | ||
| c.conn.SetDisconnectErrHandler(handler) | ||
| c.disconnectErrHandlerRegistered = true | ||
| } |
Oops, something went wrong.